Produits Entr'ouvert » Hobo

# hobo - portal to configure and deploy applications

# Copyright (C) 2022  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/>.

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):

        self.timeout = kwargs.get('timeout')

        self.verbosity = kwargs.get('verbosity')

        self.terminal_width = 0

        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:

                if service.secondary:

                    raise CommandError(

                        'Cannot migrate a secondary service, you must run the command against the hobo tenant that is primary holding it'

                    )

                target_service = service

                break

        if target_service is None:

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

        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], self.timeout, self.verbosity, self.terminal_width, notify_agents)

        if self.verbosity:

            print('Service migrated successfully, check it out : %s' % target_url)

import pytest

from django.core.management import call_command

from django.core.management.base import CommandError

from mock import Mock

from tenant_schemas.utils import tenant_context

from hobo.environment.models import Passerelle, ServiceBase

from hobo.environment.utils import get_installed_services

from hobo.multitenant.middleware import TenantMiddleware

@pytest.fixture()

def hobo_tenant(db, fake_notify, monkeypatch, fake_themes):

    monkeypatch.setattr(ServiceBase, 'is_resolvable', lambda x: True)

    monkeypatch.setattr(ServiceBase, 'has_valid_certificate', lambda x: True)

    yield call_command('cook', 'tests_schemas/example_recipe.json')

    call_command('delete_tenant', 'hobo-instance-name.dev.signalpublik.com')

def test_unknown_service(hobo_tenant):

    tenant = TenantMiddleware.get_tenant_by_hostname('hobo-instance-name.dev.signalpublik.com')

    with tenant_context(tenant):

        assert get_installed_services()

        with pytest.raises(CommandError) as e_info:

            call_command(

                'migrate_service',

                'https://unkown.dev.signalpublik.com/',

                'https://new-passerelle-instance-name.dev.signalpublik.com',

            )

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

def test_secondary_service(hobo_tenant):

    tenant = TenantMiddleware.get_tenant_by_hostname('hobo-instance-name.dev.signalpublik.com')

    with tenant_context(tenant):

        assert get_installed_services()

        Passerelle.objects.create(

            title='other passerelle',

            slug='other-passerelle',

            base_url='https://other-passerelle-instance-name.dev.signalpublik.com',

            secondary=True,

        )

        with pytest.raises(CommandError) as e_info:

            call_command(

                'migrate_service',

                'https://other-passerelle-instance-name.dev.signalpublik.com',

                'https://new-other-passerelle-instance-name.dev.signalpublik.com',

            )

        assert (

            'Cannot migrate a secondary service, you must run the command against the hobo tenant that is primary holding it'

            in str(e_info.value)

        )

def test_migrate_service_succes(hobo_tenant, monkeypatch):

    tenant = TenantMiddleware.get_tenant_by_hostname('hobo-instance-name.dev.signalpublik.com')

    with tenant_context(tenant):

        assert Passerelle.objects.count() == 1

        passerelle_service = Passerelle.objects.first()

        assert (

            passerelle_service.get_base_url_path() == 'https://passerelle-instance-name.dev.signalpublik.com/'

        )

        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://passerelle-instance-name.dev.signalpublik.com/',

            'https://new-passerelle-instance-name.dev.signalpublik.com/',

        )

        assert Passerelle.objects.count() == 1

        passerelle_service = Passerelle.objects.first()

        assert (

            passerelle_service.get_base_url_path()

            == 'https://new-passerelle-instance-name.dev.signalpublik.com/'

        )

        assert len(passerelle_service.legacy_urls) == 1

        assert (

            passerelle_service.legacy_urls[0]['base_url']

            == 'https://passerelle-instance-name.dev.signalpublik.com/'

        )