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/>.

import urllib.parse

from django.core.management import call_command

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

from tenant_schemas.utils import tenant_context

from hobo.deploy.signals import notify_agents

from hobo.environment.utils import get_or_create_local_hobo

from hobo.multitenant.middleware import TenantMiddleware, TenantNotFound

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)

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

        verbosity = kwargs.get('verbosity')

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

        legacy_domain = urllib.parse.urlparse(src_url).netloc.split(':')[0]

        target_domain = urllib.parse.urlparse(target_url).netloc.split(':')[0]

        try:

            tenant = TenantMiddleware.get_tenant_by_hostname(legacy_domain)

        except TenantNotFound:

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

        call_command('create_tenant', target_domain, legacy_hostname=legacy_domain)

        try:

            tenant = TenantMiddleware.get_tenant_by_hostname(target_domain)

        except TenantNotFound:

            raise CommandError('Could not find the new tenant %s' % target_domain)

        with tenant_context(tenant):

            local_hobo = get_or_create_local_hobo()

            local_hobo.change_base_url(target_url)

            local_hobo.save()

            notify_agents(None)

        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, get_or_create_local_hobo

from hobo.multitenant.middleware import TenantMiddleware, TenantNotFound

@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_hobo_service',

            'https://unkown-hobo-instance-name.dev.signalpublik.com/',

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

        )

        assert 'No service matches https://unkown-hobo-instance-name.dev.signalpublik.com/' in str(

            e_info.value

        )

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

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

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

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

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

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

    call_command(

        'migrate_hobo_service',

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

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

    )

    with pytest.raises(TenantNotFound) as e_info:

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

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

    with tenant_context(tenant):

        local_hobo = get_or_create_local_hobo()

        assert local_hobo.get_base_url_path() == 'https://new-hobo-instance-name.dev.signalpublik.com/'

        assert len(local_hobo.legacy_urls) == 1

        assert local_hobo.legacy_urls[0]['base_url'] == 'https://hobo-instance-name.dev.signalpublik.com/'

    # cleanup

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