Produits Entr'ouvert » Hobo

import requests

import subprocess

import sys

import tempfile

from django.conf import settings

from django.core.management.base import BaseCommand, CommandError

from django.core.management import call_command, get_commands

from django.utils.six.moves.urllib import parse as urlparse

from tenant_schemas.utils import tenant_context

from hobo.multitenant.middleware import TenantMiddleware, TenantNotFound

from hobo.theme.utils import get_theme

# TODO: move this to settings

KEY_SIZE = 1024

DAYS = 3652

def replace_file(path, content):

    dirname = os.path.dirname(path)

    fd, temp = tempfile.mkstemp(dir=dirname,

            prefix='.tmp-'+os.path.basename(path)+'-')

    f = os.fdopen(fd, 'w')

    f.flush()

    os.fsync(f.fileno())

    f.close()

    os.rename(temp, path)

class Command(BaseCommand):

    early_secondary_exit = True

import fnmatch

import json

import multiprocessing

import multiprocessing.pool

import os

import subprocess

import sys

from django.utils.six.moves.urllib import parse as urlparse

from . import settings

class BaseService(object):

    tenants_dir = None

        return False

    def execute(self, environment):

        if not os.path.exists(self.service_manage_try_cmd):

            return

        cmd = self.service_manage_cmd + ' hobo_deploy ' + self.base_url + ' -'

        cmd_process = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE,

                stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        if cmd_process.returncode != 0:

            raise RuntimeError('command "%s" failed: %r %r' % (cmd, stdout, stderr))

    @classmethod

    def notify(cls, data):

        if not os.path.exists(cls.service_manage_try_cmd):

            return

        tenants = cls.get_actual_tenants()

            else:

                return

        cmd = cls.service_manage_cmd + ' hobo_notify -'

        try:

            cmd_process = subprocess.Popen(cmd, shell=True, stdin=subprocess.PIPE,

                                           stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        except OSError:

            return

        if cmd_process.returncode != 0:

            raise RuntimeError('command "%s" failed: %r %r' % (cmd, stdout, stderr))

class Passerelle(BaseService):

    service_id = 'passerelle'

    service_manage_cmd = settings.PASSERELLE_MANAGE_COMMAND

    service_manage_try_cmd = settings.PASSERELLE_MANAGE_TRY_COMMAND

# -*- coding: utf-8 -*-

import importlib

import json

import mock

import os

import pytest

import six

from hobo.agent.worker.services import deploy, notify

from hobo.agent.worker import settings

ENVIRONMENT = {

    u'timestamp': '2022-02-22',

    u'services': [

        {

    assert '/usr/bin/wcsctl hobo_deploy https://wcs.dev.publik.love/ -' in mock_calls

    assert '/usr/lib/combo/manage.py hobo_deploy https://combo.dev.publik.love/ -' in mock_calls

    assert '/usr/bin/hobo-manage hobo_deploy https://hobo2.dev.publik.love/ -' in mock_calls

    # environment sent

    mock_calls = mocked_communicate.mock_calls

    assert len(mock_calls) == 4

    for i in range(0, len(mock_calls)):

    mocked_opened.returncode = 1

    with pytest.raises(RuntimeError, match='failed: '):

        deploy(ENVIRONMENT)

@mock.patch('hobo.agent.worker.services.os.path.exists', return_value=True)

@mock.patch('hobo.agent.worker.services.subprocess')

def test_deploy_host_with_agent_patterns(mocked_subprocess, mocked_exists, local_settings):

    # process called

    mock_calls = set(x[1][0] for x in mocked_subprocess.Popen.mock_calls)

    assert mock_calls == {'/usr/lib/combo/manage.py hobo_notify -'}

    # notification sent

    mock_calls = mocked_communicate.mock_calls

    assert len(mock_calls) == 1

    mocked_opened.returncode = 1

    with pytest.raises(RuntimeError, match='failed: '):

        notify(NOTIFICATION)

    # ignore subprocess error

    mocked_communicate.reset_mock()

    mocked_subprocess.Popen.side_effect = OSError