Produits Entr'ouvert » Hobo

from hobo.environment.utils import wait_operationals

        self.terminal_width = 0

        wait_operationals(services, timeout, self.verbosity, self.terminal_width, notify_agents)

# hobo - portal to configure and deploy applications

# Copyright (C) 2021  Entr'ouvert

#

# This program is free software: you can redistribute it and/or modify it

# under the terms of the GNU Affero General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU Affero General Public License for more details.

#

# You should have received a copy of the GNU Affero General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import subprocess

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

from hobo.deploy.signals import notify_agents

from hobo.environment.utils import get_installed_services, wait_operationals

def normalize_url(url):

    if not url.endswith('/'):

        url += '/'

    return url

class Command(BaseCommand):

    verbosity = 1

    def add_arguments(self, parser):

        parser.add_argument('src_url', type=str)

        parser.add_argument('target_url', type=str)

        parser.add_argument(

            '--timeout',

            type=int,

            action='store',

            default=120,

            help='set the timeout for the wait_operationals method',

        )

    def handle(self, src_url, target_url, *args, **kwargs):

        timeout = kwargs.get('timeout')

        verbosity = kwargs.get('verbosity')

        terminal_width = 0

        if self.verbosity > 1:

            try:

                terminal_width = int(subprocess.check_output(['tput', 'cols']).strip())

            except OSError:

                terminal_width = 80

        src_url, target_url = normalize_url(src_url), normalize_url(target_url)

        target_service = None

        for service in get_installed_services():

            if service.get_base_url_path() == src_url:

                target_service = service

                break

        if target_service is None:

            raise CommandError('No service matches %s' % src_url)

        if target_service.Extra.service_id in ('authentic', 'hobo', 'wcs'):

            raise CommandError("%s service type is not supported" % target_service.Extra.service_id)

        target_service.change_base_url(target_url)

        target_service.last_operational_check_timestamp = None

        target_service.last_operational_success_timestamp = None

        target_service.save()

        notify_agents(None)

        wait_operationals([target_service], timeout, verbosity, terminal_width, notify_agents)

        if self.verbosity:

            print('Service migrated successfully, check it out : %s' % target_service.get_base_url_path())

import sys

import time

from django.core.management.base import CommandError

def wait_operationals(services, timeout, verbosity, terminal_width, notify_agents_func):

    t0 = time.time()

    i = 0

    last_service_url = None

    last_notification = t0

    while len(services) > 0:

        if time.time() - last_notification > 15:

            last_notification = time.time()

            notify_agents_func(None)

        for service in services[:]:

            if service.last_operational_success_timestamp:

                services.remove(service)

                continue

            service.check_operational()

        if len(services) == 0:

            break

        if verbosity == 1:

            sys.stderr.write('.')

        elif verbosity > 1:

            if last_service_url != services[0].base_url:

                last_service_url = services[0].base_url

                i = 0

            elif i == (terminal_width - len(services[0].base_url) - 25):

                i = 0

            i += 1

            sys.stderr.write('\rWaiting for %s ' % services[0].base_url)

            sys.stderr.write('%5ds ' % (timeout - (time.time() - t0)))

            sys.stderr.write('.' * i)

            sys.stderr.flush()

        time.sleep(0.5)

        if time.time() - t0 > timeout:

            if verbosity:

                sys.stderr.write('\n')

            raise CommandError('timeout waiting for %s' % ', '.join([x.base_url for x in services]))

    command.terminal_width = 80

from hobo.environment.management.commands.migrate_service import Command as MigrateServiceCommand

def test_deploy_specifics_on_hobo_agent(transactional_db):

import os

import pytest

from django.core.management import call_command, load_command_class

from django.core.management.base import CommandError

from mock import Mock

from tenant_schemas.utils import tenant_context

from hobo.environment.models import Chrono

from hobo.environment.utils import get_installed_services

from hobo.multitenant.middleware import TenantMiddleware

def assert_deployed(domain):

    tenant = TenantMiddleware.get_tenant_by_hostname(domain)

    tenant_hobo_json = os.path.join(tenant.get_directory(), 'hobo.json')

    assert os.path.exists(tenant_hobo_json)

@pytest.fixture()

def hobo_tenant(db):

    name = 'hobo.dev.publik.love'

    yield call_command('create_hobo_tenant', name)

    call_command('delete_tenant', name)

def test_unknown_service(hobo_tenant):

    command = load_command_class('hobo.agent.hobo', 'hobo_deploy')

    domain = 'hobo.dev.publik.love'

    command.handle('https://%s/' % domain, 'tests_schemas/env.json')

    assert_deployed(domain)

    tenant = TenantMiddleware.get_tenant_by_hostname('hobo.dev.publik.love')

    with tenant_context(tenant):

        assert get_installed_services()

        with pytest.raises(CommandError) as e_info:

            call_command(

                'migrate_service', 'https://unkown.dev.publik.love/', 'https://new-chrono.dev.publik.love/'

            )

        assert 'No service matches https://unkown.dev.publik.love/' in str(e_info.value)

def test_unsupported_service(hobo_tenant):

    command = load_command_class('hobo.agent.hobo', 'hobo_deploy')

    domain = 'hobo.dev.publik.love'

    command.handle('https://%s/' % domain, 'tests_schemas/env.json')

    assert_deployed(domain)

    tenant = TenantMiddleware.get_tenant_by_hostname('hobo.dev.publik.love')

    with tenant_context(tenant):

        assert get_installed_services()

        with pytest.raises(CommandError) as e_info:

            call_command(

                'migrate_service', 'https://wcs.dev.publik.love/', 'https://new-wcs.dev.publik.love/'

            )

        assert 'wcs service type is not supported' in str(e_info.value)

def test_change_base_url(hobo_tenant, monkeypatch):

    deploy_command = load_command_class('hobo.agent.hobo', 'hobo_deploy')

    domain = 'hobo.dev.publik.love'

    deploy_command.handle('https://%s/' % domain, 'tests_schemas/env.json')

    assert_deployed(domain)

    tenant = TenantMiddleware.get_tenant_by_hostname(domain)

    with tenant_context(tenant):

        assert Chrono.objects.count() == 1

        chrono_service = Chrono.objects.first()

        assert chrono_service.get_base_url_path() == 'https://chrono.dev.publik.love/'

        assert get_installed_services()

        monkeypatch.setattr('hobo.environment.management.commands.migrate_service.notify_agents', Mock())

        monkeypatch.setattr('hobo.environment.management.commands.migrate_service.wait_operationals', Mock())

        call_command(

            'migrate_service', 'https://chrono.dev.publik.love/', 'https://new-chrono.dev.publik.love/'

        )

        assert Chrono.objects.count() == 1

        chrono_service = Chrono.objects.first()

        assert chrono_service.get_base_url_path() == 'https://new-chrono.dev.publik.love/'

        assert len(chrono_service.legacy_urls) == 1

        assert chrono_service.legacy_urls[0]['base_url'] == 'https://chrono.dev.publik.love/'