entrouvert
/
hobo
16
0
Fork 0
Code Issues Pull Requests 15 Releases Activity

hobo_deploy: handle url change on a service (#58908)

This commit is contained in:
Emmanuel Cazenave 2021-12-01 15:19:45 +01:00
parent 8463aa66f4
commit f8041c4457
11 changed files with 508 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
hobo/agent/authentic2/management/commands/hobo_deploy.py
View File

@ -155,9 +155,21 @@ class Command(hobo_deploy.Command):
continue
metadata_text = metadata_response.text
provider, service_created = LibertyProvider.objects.get_or_create(
entity_id=sp_url, protocol_conformance=lasso.PROTOCOL_SAML_2_0
)
provider, service_created = None, False
for legacy_urls in service.get('legacy_urls', []):
try:
provider = LibertyProvider.objects.get(
entity_id=legacy_urls['saml-sp-metadata-url'],
protocol_conformance=lasso.PROTOCOL_SAML_2_0,
)
provider.entity_id = sp_url
break
except LibertyProvider.DoesNotExist:
pass
if not provider:
provider, service_created = LibertyProvider.objects.get_or_create(
entity_id=sp_url, protocol_conformance=lasso.PROTOCOL_SAML_2_0
)
provider.name = service['title']
provider.slug = service['slug']
provider.federation_source = 'hobo'

13
hobo/agent/common/management/commands/hobo_deploy.py
View File

@ -74,12 +74,21 @@ class Command(BaseCommand):
# early exit, we don't redeploy secondary services
return
domain = urlparse.urlparse(self.me.get('base_url')).netloc.split(':')[0]
legacy_domain = None
try:
tenant = TenantMiddleware.get_tenant_by_hostname(domain)
except TenantNotFound:
# create tenant for domain
call_command('create_tenant', domain)
# might be a domain change request
for legacy_urls in self.me.get('legacy_urls', []):
old_domain = urlparse.urlparse(legacy_urls['base_url']).netloc.split(':')[0]
try:
tenant = TenantMiddleware.get_tenant_by_hostname(old_domain)
legacy_domain = old_domain
break
except TenantNotFound:
pass
call_command('create_tenant', domain, legacy_hostname=legacy_domain)
tenant = TenantMiddleware.get_tenant_by_hostname(domain)
timestamp = hobo_environment.get('timestamp')

36
hobo/agent/hobo/management/commands/hobo_deploy.py
View File

@ -5,6 +5,7 @@ from tenant_schemas.utils import tenant_context
from hobo.agent.common.management.commands import hobo_deploy
from hobo.deploy.signals import notify_agents
from hobo.environment.models import AVAILABLE_SERVICES, Hobo, Variable
from hobo.environment.utils import get_or_create_local_hobo
from hobo.multitenant.middleware import TenantMiddleware, TenantNotFound
from hobo.profile.models import AttributeDefinition
@ -38,6 +39,11 @@ class Command(hobo_deploy.Command):
if service_dict.get('secondary'):
continue
if service_dict.get('base_url') == me['base_url']:
# URL might have changed, need to update local hobo
local_hobo = get_or_create_local_hobo()
if local_hobo and local_hobo.get_base_url_path() != service_dict.get('base_url'):
local_hobo.change_base_url(service_dict.get('base_url'))
local_hobo.save()
continue
for service_klass in AVAILABLE_SERVICES:
if service_klass.Extra.service_id == service_dict.get('service-id'):
@ -63,15 +69,27 @@ class Command(hobo_deploy.Command):
slug_prefix = '_%s_' % hobo_environment['variables']['ou-slug']
service_slug = '%s%s' % (slug_prefix, service_dict['slug'])
service, created = service_klass.objects.get_or_create(
base_url=service_dict['base_url'],
secondary=True,
defaults={
'title': service_dict['title'],
'slug': service_slug,
'secret_key': service_dict.get('secret_key'),
},
)
service, created = None, False
for legacy_urls in service_dict.get('legacy_urls', []):
try:
service = service_klass.objects.get(base_url=legacy_urls['base_url'], secondary=True)
service.change_base_url(service_dict['base_url'])
break
except service_klass.DoesNotExist:
pass
if not service:
service, created = service_klass.objects.get_or_create(
base_url=service_dict['base_url'],
secondary=True,
defaults={
'title': service_dict['title'],
'slug': service_slug,
'secret_key': service_dict.get('secret_key'),
},
)
service.title = service_dict['title']
service.secret_key = service_dict.get('secret_key')
service.template_name = service_dict.get('template_name') or ''

59
hobo/environment/migrations/0025_legacy_urls.py Normal file
View File

@ -0,0 +1,59 @@
# Generated by Django 2.2.24 on 2021-12-02 13:55
import django.contrib.postgres.fields.jsonb
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('environment', '0024_remove_local_hobo'),
]
operations = [
migrations.AddField(
model_name='authentic',
name='legacy_urls',
field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),
),
migrations.AddField(
model_name='bijoe',
name='legacy_urls',
field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),
),
migrations.AddField(
model_name='chrono',
name='legacy_urls',
field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),
),
migrations.AddField(
model_name='combo',
name='legacy_urls',
field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),
),
migrations.AddField(
model_name='fargo',
name='legacy_urls',
field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),
),
migrations.AddField(
model_name='hobo',
name='legacy_urls',
field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),
),
migrations.AddField(
model_name='passerelle',
name='legacy_urls',
field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),
),
migrations.AddField(
model_name='wcs',
name='legacy_urls',
field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),
),
migrations.AddField(
model_name='welco',
name='legacy_urls',
field=django.contrib.postgres.fields.jsonb.JSONField(blank=True, default=list, null=True),
),
]

25
hobo/environment/models.py
View File

@ -19,11 +19,13 @@ import json
import random
import re
import socket
import time
import requests
from django.conf import settings
from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import JSONField
from django.core.cache import cache
from django.core.exceptions import ValidationError
from django.core.validators import URLValidator
@ -104,6 +106,7 @@ class ServiceBase(models.Model):
title = models.CharField(_('Title'), max_length=50)
slug = models.SlugField(_('Slug'))
base_url = models.CharField(_('Base URL'), max_length=200, validators=[URLValidator()])
legacy_urls = JSONField(null=True, default=list, blank=True)
secret_key = models.CharField(_('Secret Key'), max_length=60)
template_name = models.CharField(_('Template'), max_length=60, blank=True)
secondary = models.BooleanField(_('Secondary Service'), default=False)
@ -159,6 +162,8 @@ class ServiceBase(models.Model):
]
)
as_dict['base_url'] = self.get_base_url_path()
if self.legacy_urls:
as_dict['legacy_urls'] = self.legacy_urls
as_dict['service-id'] = self.Extra.service_id
as_dict['service-label'] = force_text(self.Extra.service_label)
as_dict['variables'] = dict(((v.name, v.json) for v in self.variables.all()))
@ -294,6 +299,26 @@ class ServiceBase(models.Model):
result[name] = value
return result
def change_base_url(self, base_url):
service_dict = self.as_dict()
timestamp = datetime.datetime.now()
legacy_urls = {
'base_url': service_dict['base_url'],
'timestamp': str(time.mktime(timestamp.timetuple()) + timestamp.microsecond / 1e6),
}
for url_key in (
'saml-sp-metadata-url',
'saml-idp-metadata-url',
'backoffice-menu-url',
'provisionning-url',
):
if url_key in service_dict:
legacy_urls[url_key] = service_dict[url_key]
if not self.legacy_urls:
self.legacy_urls = []
self.legacy_urls.insert(0, legacy_urls)
self.base_url = base_url
class Authentic(ServiceBase):
use_as_idp_for_self = models.BooleanField(verbose_name=_('Use as IdP'), default=False)

4
tests/test_hobo_deploy.py
View File

@ -152,7 +152,9 @@ def test_deploy(mocked_get_tenant_by_hostname, tmpdir):
mocked_get_tenant_by_hostname.side_effect = [TenantNotFound, tenant]
with patch('hobo.agent.common.management.commands.hobo_deploy.call_command') as mocked_call_command:
command.deploy(base_url, ENVIRONMENT, None)
assert mocked_call_command.mock_calls == [call('create_tenant', 'combo.dev.publik.love')]
assert mocked_call_command.mock_calls == [
call('create_tenant', 'combo.dev.publik.love', legacy_hostname=None)
]
assert_deployed()
# already there (timestamp do not change)

178
tests_authentic/test_hobo_deploy.py
View File

@ -579,3 +579,181 @@ def test_import_template(db, tenant_base):
export_ref = sort_and_remove_uuid(export_site())
file_ref = sort_and_remove_uuid(json.loads(content))
assert export_ref == file_ref
def test_hobo_deploy_with_legacy_urls(monkeypatch, tenant_base, mocker, skeleton_dir, tmp_path):
from django.core.management import call_command
from hobo.agent.authentic2.management.commands.hobo_deploy import Command as HoboDeployCommand
requests_get = mocker.patch('requests.get')
meta1 = '''<?xml version="1.0"?>
<EntityDescriptor xmlns="urn:oasis:names:tc:SAML:2.0:metadata"
xmlns:saml="urn:oasis:names:tc:SAML:2.0:assertion"
xmlns:ds="http://www.w3.org/2000/09/xmldsig#"
entityID="http://passerelle.example.net/saml/metadata">
<SPSSODescriptor
AuthnRequestsSigned="true" WantAssertionsSigned="true"
protocolSupportEnumeration="urn:oasis:names:tc:SAML:2.0:protocol">
<KeyDescriptor use="signing">
<ds:KeyInfo xmlns:ds="http://www.w3.org/2000/09/xmldsig#">
<KeyValue xmlns="http://www.w3.org/2000/09/xmldsig#">
<RSAKeyValue>
<Modulus>nJpkBznHNbvE+RAC6mU+NPQnIWs8gFNCm6I3FPcUKYpaJbXaurJ4cJgvnaEiqIXPQDcbHxuLeCbYbId9yascWZirvQbh8d/r+Vv+24bPG++9gW+i3Nnz1VW8V+z0b+puHWvM/FjJjBNJgWkI38gaupz47U6/02CtWx00stitiwk=</Modulus>
<Exponent>AQAB</Exponent>
</RSAKeyValue>
</KeyValue>
</ds:KeyInfo>
</KeyDescriptor>
<KeyDescriptor use="encryption">
<ds:KeyInfo xmlns:ds="http://www.w3.org/2000/09/xmldsig#">
<KeyValue xmlns="http://www.w3.org/2000/09/xmldsig#">
<RSAKeyValue>
<Modulus>3BxSiAzGvY1Yuqa31L7Zr2WHM/8cn5oX+Q6A2SYgzjuvAgnWyizN8YgW/fHR4G7MtkmZ5RFJLXfcSLwbUfpFHV6KO1ikbgViYuFempM+SWtjqEI7ribm9GaI5kUzHJZBrH3/Q9XAd9/GLLALxurGjbKDeLfc0D+7el26g4sYmA8=</Modulus>
<Exponent>AQAB</Exponent>
</RSAKeyValue>
</KeyValue>
</ds:KeyInfo>
</KeyDescriptor>
<SingleLogoutService
Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
Location="http://passerelle.example.net/saml/singleLogout"
ResponseLocation="http://passerelle.example.net/saml/singleLogoutReturn" />
<SingleLogoutService
Binding="urn:oasis:names:tc:SAML:2.0:bindings:SOAP"
Location="http://passerelle.example.net/saml/singleLogoutSOAP" />
<ManageNameIDService
Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
Location="http://passerelle.example.net/saml/manageNameId"
ResponseLocation="http://passerelle.example.net/saml/manageNameIdReturn" />
<ManageNameIDService
Binding="urn:oasis:names:tc:SAML:2.0:bindings:SOAP"
Location="http://passerelle.example.net/saml/manageNameIdSOAP" />
<AssertionConsumerService isDefault="true" index="0"
Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Artifact"
Location="http://passerelle.example.net/saml/assertionConsumerArtifact" />
<AssertionConsumerService index="1"
Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-POST"
Location="http://passerelle.example.net/saml/assertionConsumerPost" />
<AssertionConsumerService index="2"
Binding="urn:oasis:names:tc:SAML:2.0:bindings:PAOS"
Location="http://passerelle.example.net/saml/assertionConsumerSOAP" />
<AssertionConsumerService index="3"
Binding="urn:oasis:names:tc:SAML:2.0:bindings:HTTP-Redirect"
Location="http://passerelle.example.net/saml/assertionConsumerRedirect" />
</SPSSODescriptor>
</EntityDescriptor>'''
meta2 = meta1.replace('passerelle.example.net', 'new-passerelle.example.net')
monkeypatch.setattr(HoboDeployCommand, 'backoff_factor', 0.0001)
side_effect_iter = iter([meta1, meta2])
def side_effect(*args, **kwargs):
for v in side_effect_iter:
m = mock.Mock()
m.text = v
return m
requests_get.side_effect = side_effect
def hobo_json(env_dict):
with tempfile.NamedTemporaryFile(mode='w', dir=str(tmp_path), delete=False) as hobo_json:
hobo_json_content = json.dumps(env_dict)
hobo_json.write(hobo_json_content)
return hobo_json.name
env = {
'services': [
{
'service-id': 'authentic',
'slug': 'test',
'title': 'Test',
'this': True,
'secret_key': '12345',
'base_url': 'http://sso.example.net',
'variables': {
'other_variable': 'bar',
},
},
{
'service-id': 'passerelle',
'slug': 'passerelle',
'title': u'Passerelle',
'base_url': 'http://passerelle.example.net',
'saml-sp-metadata-url': 'http://passerelle.example.net/saml/metadata',
},
],
'users': [],
'profile': {'fields': []},
}
with mock.patch('hobo.agent.authentic2.provisionning.notify_agents'):
call_command('hobo_deploy', 'http://sso.example.net', hobo_json(env))
from hobo.multitenant.middleware import TenantMiddleware
tenants = list(TenantMiddleware.get_tenants())
assert len(tenants) == 1
tenant = tenants[0]
assert tenant.domain_url == 'sso.example.net'
assert tenant.schema_name == 'sso_example_net'
tenant_directory = tenant.get_directory()
assert tenant_directory == os.path.join(tenant_base, tenant.domain_url)
assert os.path.exists(os.path.join(tenant_directory, 'saml.crt'))
assert os.path.exists(os.path.join(tenant_directory, 'saml.key'))
from tenant_schemas.utils import tenant_context
with tenant_context(tenant):
# SAML checks
from authentic2.saml.models import LibertyProvider
assert LibertyProvider.objects.count() == 1
provider = LibertyProvider.objects.first()
provider_id = provider.pk
assert provider.entity_id == 'http://passerelle.example.net/saml/metadata'
assert provider.metadata == meta1
new_env = {
'services': [
{
'service-id': 'authentic',
'slug': 'test',
'title': 'Test',
'this': True,
'secret_key': '12345',
'base_url': 'http://sso.example.net',
'variables': {
'other_variable': 'bar',
},
},
{
'service-id': 'passerelle',
'slug': 'passerelle',
'title': u'Passerelle',
'base_url': 'http://new-passerelle.example.net',
'saml-sp-metadata-url': 'http://new-passerelle.example.net/saml/metadata',
'legacy_urls': [
{
'base_url': 'http://passerelle.example.net',
'saml-sp-metadata-url': 'http://passerelle.example.net/saml/metadata',
}
],
},
],
'users': [],
'profile': {'fields': []},
}
with mock.patch('hobo.agent.authentic2.provisionning.notify_agents'):
call_command('hobo_deploy', '--ignore-timestamp', 'http://sso.example.net', hobo_json(new_env))
# check that liberty provider is updated
with tenant_context(tenant):
from authentic2.saml.models import LibertyProvider
assert LibertyProvider.objects.count() == 1
provider = LibertyProvider.objects.first()
assert provider.metadata == meta2
assert provider.entity_id == 'http://new-passerelle.example.net/saml/metadata'
assert provider.pk == provider_id

9
tests_multipublik/conftest.py
View File

@ -21,14 +21,19 @@ def tenants(transactional_db, request, settings):
t.create_schema()
return t
tenants = [make_tenant('tenant1.example.net')]
tenants = [
make_tenant('tenant1.example.net'),
make_tenant('hobo2.example.net'),
make_tenant('hobo3.example.net'),
]
def fin():
from django.db import connection
connection.set_schema_to_public()
for t in tenants:
t.delete(True)
if os.path.exists(t.get_directory()):
t.delete(True)
shutil.rmtree(base)
request.addfinalizer(fin)

116
tests_multipublik/test_multipublik.py
View File

@ -345,3 +345,119 @@ def test_multipublik(tenants, mocker):
with tenant_context(hobo2):
assert Combo.objects.filter(secondary=True).count() == 1
assert Combo.objects.filter(secondary=False).count() == 1
# URL change in interco portal
with tenant_context(hobo1):
combo = Combo.objects.get(slug='portal')
assert combo.base_url == 'https://combo1.example.net/'
combo.change_base_url('https://new-combo1.example.net')
combo.save()
# check the interco hobo json
hobo_json = get_hobo_json()
for service in hobo_json['services']:
if service['slug'] == 'portal':
assert service['base_url'] == 'https://new-combo1.example.net/'
assert len(service['legacy_urls']) == 1
assert service['legacy_urls'][0]['base_url'] == 'https://combo1.example.net/'
break
else:
assert False, "no portal found"
# inform coll2 about interco environment
HoboDeployCommand().handle(hobo2.base_url, get_hobo_json_filename(hobo1))
with tenant_context(hobo2):
# no extra combo created in coll 2
assert Combo.objects.filter().count() == 2
# interco portal url changed
combo = Combo.objects.get(slug='_interco_portal')
assert combo.base_url == 'https://new-combo1.example.net/'
assert len(combo.legacy_urls) == 1
assert combo.legacy_urls[0]['base_url'] == 'https://combo1.example.net/'
# inform coll2 about interco environment a second time, check that nothing changes
HoboDeployCommand().handle(hobo2.base_url, get_hobo_json_filename(hobo1))
with tenant_context(hobo2):
assert Combo.objects.filter().count() == 2
combo = Combo.objects.get(slug='_interco_portal')
assert combo.base_url == 'https://new-combo1.example.net/'
assert len(combo.legacy_urls) == 1
assert combo.legacy_urls[0]['base_url'] == 'https://combo1.example.net/'
# URL change in coll2 portal
with tenant_context(hobo2):
combo = Combo.objects.get(slug='portal')
assert combo.base_url == 'https://combo2.example.net/'
combo.change_base_url('https://new-combo2.example.net')
combo.save()
# check the coll2 hobo json
hobo_json = get_hobo_json()
for service in hobo_json['services']:
if service['slug'] == 'portal':
assert service['base_url'] == 'https://new-combo2.example.net/'
assert len(service['legacy_urls']) == 1
assert service['legacy_urls'][0]['base_url'] == 'https://combo2.example.net/'
break
else:
assert False, "no portal found"
# inform interco about coll2 environment
HoboDeployCommand().handle(hobo1.base_url, get_hobo_json_filename(hobo2))
with tenant_context(hobo1):
# no extra combo created in interco
assert Combo.objects.filter().count() == 3
# coll2 portal url changed
combo = Combo.objects.get(slug='_hobo-coll2_portal')
assert combo.base_url == 'https://new-combo2.example.net/'
assert len(combo.legacy_urls) == 1
assert combo.legacy_urls[0]['base_url'] == 'https://combo2.example.net/'
# URL change on the primary hobo
with tenant_context(hobo1):
hobo = Hobo.objects.get(slug='hobo')
assert hobo.base_url == 'https://tenant1.example.net/'
hobo.change_base_url('https://new-tenant1.example.net')
hobo.save()
# check the interco hobo json
hobo_json = get_hobo_json()
for service in hobo_json['services']:
if service['slug'] == 'hobo':
assert service['base_url'] == 'https://new-tenant1.example.net/'
assert len(service['legacy_urls']) == 1
assert service['legacy_urls'][0]['base_url'] == 'https://tenant1.example.net/'
break
else:
assert False, 'no hobo found'
# inform coll2 about interco environment
HoboDeployCommand().handle(hobo2.base_url, get_hobo_json_filename(hobo1))
with tenant_context(hobo2):
# no extra hobo created in coll 2
assert Hobo.objects.count() == 3
# interco hobo url changed
hobo = Hobo.objects.get(slug='_interco_hobo')
assert hobo.base_url == 'https://new-tenant1.example.net/'
assert len(hobo.legacy_urls) == 1
assert hobo.legacy_urls[0]['base_url'] == 'https://tenant1.example.net/'
# URL change on coll2 hobo (initiated by the interco tenant)
with tenant_context(hobo1):
hobo2 = Hobo.objects.get(slug='hobo-coll2')
assert hobo2.base_url == 'https://hobo2.example.net/'
hobo2.change_base_url('https://new-hobo2.example.net')
hobo2.save()
# inform coll2 about interco environment
HoboDeployCommand().handle(hobo2.base_url, get_hobo_json_filename(hobo1))
hobo2 = TenantMiddleware.get_tenant_by_hostname('new-hobo2.example.net')
with tenant_context(hobo2):
# no extra hobo created in coll 2
assert Hobo.objects.count() == 3
# coll2 hobo url changed
hobo = Hobo.objects.get(slug='hobo')
assert hobo.base_url == 'https://new-hobo2.example.net/'
assert len(hobo.legacy_urls) == 1
assert hobo.legacy_urls[0]['base_url'] == 'https://hobo2.example.net/'

45
tests_schemas/legacy_urls_chrono_env.json Normal file
View File

@ -0,0 +1,45 @@
{
"services": [
{
"service-id": "chrono",
"base_url": "https://new-chrono.dev.publik.love/",
"slug": "agendas",
"title": "CHRONO",
"secret_key": "123",
"template_name": "import_me",
"legacy_urls": [
{
"base_url": "https://chrono.dev.publik.love/"
}
]
},
{
"service-id": "wcs",
"base_url": "https://wcs.dev.publik.love/",
"slug": "eservices",
"title": "WCS"
},
{
"service-id": "hobo",
"base_url": "https://hobo.dev.publik.love/",
"slug": "hobo",
"title": "HOBO",
"secret_key": "123"
},
{
"service-id": "combo",
"base_url": "https://combo.dev.publik.love/",
"slug": "portal",
"title": "COMBO",
"secret_key": "123",
"template_name": "import_me"
},
{
"service-id": "authentic",
"base_url": "https://authentic.dev.publik.love/",
"slug": "idp",
"title": "A2",
"secret_key": "123"
}
]
}

25
tests_schemas/test_hobo_deploy.py
View File

@ -1,11 +1,12 @@
import os
import mock
import pytest
from django.core.management import call_command, get_commands, load_command_class
from tenant_schemas.utils import tenant_context
from hobo.environment.models import Variable
from hobo.multitenant.middleware import TenantMiddleware
from hobo.multitenant.middleware import TenantMiddleware, TenantNotFound
def assert_deployed(domain):
@ -26,11 +27,11 @@ def test_import_template(mocked_get_commands, mocked_call_command, db):
command = load_command_class('hobo.agent.common', 'hobo_deploy')
domain = 'chrono.dev.publik.love'
def my_call_command(command, parameter):
def my_call_command(command, parameter, **kwargs):
if command == 'import_template':
my_call_command.import_template_was_called = True
return
call_command(command, parameter)
call_command(command, parameter, **kwargs)
mocked_get_commands.return_value = ['import_template']
mocked_call_command.side_effect = my_call_command
@ -77,3 +78,21 @@ def test_deploy_specifics_on_hobo_agent(db):
# fails to simulate call from bijoe agent, that overload deploy_specifics()
# $ bijoe-mange hobo-deploy
# here, because this code is not implemented here
def test_deploy_with_legacy_urls(db):
command = load_command_class('hobo.agent.common', 'hobo_deploy')
domain = 'chrono.dev.publik.love'
command.handle('https://%s/' % domain, 'tests_schemas/env.json')
assert_deployed(domain)
tenant_directory = TenantMiddleware.get_tenant_by_hostname(domain).get_directory()
# change domain
new_domain = 'new-chrono.dev.publik.love'
command.handle('https://%s/' % new_domain, 'tests_schemas/legacy_urls_chrono_env.json')
assert_deployed(new_domain)
# check old tenant is gone
with pytest.raises(TenantNotFound):
TenantMiddleware.get_tenant_by_hostname(domain)
assert not os.path.exists(tenant_directory)