tests: add unit tests for hobo_deploy.py (#33224)
This commit is contained in:
parent
cbd08f81b7
commit
7dcc824782
|
|
@ -0,0 +1,561 @@
|
|||
""" unit tests (mainly for code coverage)
|
||||
"""
|
||||
import StringIO
|
||||
import os
|
||||
import sys
|
||||
|
||||
import json
|
||||
import pytest
|
||||
from mock import call, patch, Mock
|
||||
from requests import Response, exceptions
|
||||
|
||||
from hobo.agent.combo.management.commands.hobo_deploy import Command as ComboCommand
|
||||
from hobo.agent.common.management.commands.hobo_deploy import (
|
||||
replace_file, Command, CommandError)
|
||||
from hobo.agent.hobo.management.commands.hobo_deploy import Command as HoboCommand
|
||||
from hobo.environment.models import Variable, Combo, Hobo, Wcs
|
||||
from hobo.multitenant.middleware import TenantNotFound
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fake_themes(settings, tmpdir):
|
||||
THEMES="""
|
||||
[
|
||||
{
|
||||
"id": "alfortville",
|
||||
"label": "Alfortville",
|
||||
"variables": {
|
||||
"css_variant": "alfortville",
|
||||
"no_extra_js": false,
|
||||
"theme_color": "#804697"
|
||||
},
|
||||
"overlay": "foobar"
|
||||
},
|
||||
{
|
||||
"id": "publik",
|
||||
"label": "Publik",
|
||||
"variables": {
|
||||
"css_variant": "publik",
|
||||
"no_extra_js": false,
|
||||
"theme_color": "#E80E89"
|
||||
}
|
||||
}
|
||||
]
|
||||
"""
|
||||
base_dir = str(tmpdir.mkdir('themes'))
|
||||
settings.THEMES_DIRECTORY = base_dir
|
||||
|
||||
themes_dir = os.path.join(base_dir, 'publik-base')
|
||||
os.mkdir(themes_dir)
|
||||
with open(os.path.join(themes_dir, 'themes.json'), 'w') as handler:
|
||||
handler.write(THEMES)
|
||||
|
||||
# populate 'foobar' overlay
|
||||
themes_dir = os.path.join(base_dir, 'foobar')
|
||||
os.mkdir(themes_dir)
|
||||
for part in ('static', 'templates'):
|
||||
os.mkdir(os.path.join(themes_dir, part))
|
||||
|
||||
|
||||
def test_replace_file(tmpdir):
|
||||
path = str(tmpdir) + '/my_file.txt'
|
||||
|
||||
content = 'content of my new file'
|
||||
replace_file(path, content)
|
||||
with open(path, 'r') as handler:
|
||||
assert handler.read() == content
|
||||
|
||||
content = 'new content for my file'
|
||||
replace_file(path, content)
|
||||
with open(path, 'r') as handler:
|
||||
assert handler.read() == content
|
||||
|
||||
|
||||
def test_handle_from_scratch():
|
||||
"""API using JSON from file or from stdin"""
|
||||
command = Command()
|
||||
CONTENT = """
|
||||
{
|
||||
"services": [{
|
||||
"service-id": "combo"
|
||||
}]
|
||||
}
|
||||
"""
|
||||
command.deploy = Mock()
|
||||
EXPECTED = [call('https://combo.dev.publik.love/',
|
||||
{'services': [{'service-id': 'combo'}]}, None)]
|
||||
|
||||
# handle from file
|
||||
command.deploy.reset_mock()
|
||||
with patch('hobo.agent.common.management.commands.hobo_deploy.file') as mocked_open:
|
||||
mocked_open.side_effect = [StringIO.StringIO(CONTENT)]
|
||||
command.handle('https://combo.dev.publik.love/', 'envbof.json')
|
||||
assert command.deploy.mock_calls == EXPECTED
|
||||
|
||||
# handle using a pipe
|
||||
command.deploy.reset_mock()
|
||||
backup = sys.stdin
|
||||
sys.stdin = StringIO.StringIO(CONTENT)
|
||||
command.handle('https://combo.dev.publik.love/', '-')
|
||||
sys.stdin = backup
|
||||
assert command.deploy.mock_calls == EXPECTED
|
||||
|
||||
# JSON having syntax error
|
||||
command.deploy.reset_mock()
|
||||
with patch('hobo.agent.common.management.commands.hobo_deploy.file') as mocked_open:
|
||||
mocked_open.side_effect = [StringIO.StringIO('malformated JSON')]
|
||||
with pytest.raises(ValueError, match='No JSON object could be decoded'):
|
||||
command.handle('https://combo.dev.publik.love/', 'env.json')
|
||||
assert command.deploy.mock_calls == []
|
||||
|
||||
# missing args
|
||||
with pytest.raises(CommandError, match='missing args'):
|
||||
command.handle('https://combo.dev.publik.love/')
|
||||
with pytest.raises(CommandError, match='missing args'):
|
||||
command.handle(json_filename='env.json')
|
||||
|
||||
|
||||
@patch('hobo.agent.common.management.commands.hobo_deploy.TenantMiddleware.get_tenants')
|
||||
def test_handle_redeploy_case(mocked_get_tenants):
|
||||
"""API using JSON from previous one we put on tenant"""
|
||||
command = Command()
|
||||
command.deploy = Mock()
|
||||
tenant = Mock()
|
||||
|
||||
# redeploy
|
||||
command.deploy.reset_mock()
|
||||
ENVIRONMENT = {'services': [{'this': True,
|
||||
'base_url': 'https://combo.dev.publik.love/',
|
||||
'service-id': 'combo'}]}
|
||||
tenant.get_hobo_json = Mock(return_value=ENVIRONMENT)
|
||||
mocked_get_tenants.return_value = [tenant]
|
||||
command.handle(redeploy=True)
|
||||
assert command.deploy.mock_calls == [ # ignore_timestamp is set to True
|
||||
call('https://combo.dev.publik.love/', ENVIRONMENT, True)]
|
||||
|
||||
# redeploy having wrong JSON content: no 'base_url' entry
|
||||
command.deploy.reset_mock()
|
||||
tenant.get_hobo_json = Mock(return_value=
|
||||
{'services': [{'service-id': 'combo', 'this': True}]})
|
||||
mocked_get_tenants.return_value = [tenant]
|
||||
with pytest.raises(KeyError, match='base_url'):
|
||||
command.handle(redeploy=True)
|
||||
assert command.deploy.mock_calls == []
|
||||
|
||||
# redeploy having wrong JSON content: 'this' entry not found
|
||||
command.deploy.reset_mock()
|
||||
tenant.get_hobo_json = Mock(return_value={'services': [{'service-id': 'combo'}]})
|
||||
mocked_get_tenants.return_value = [tenant]
|
||||
command.handle(redeploy=True)
|
||||
assert command.deploy.mock_calls == []
|
||||
|
||||
# IOError
|
||||
command.deploy.reset_mock()
|
||||
tenant.get_hobo_json = Mock(side_effect=IOError)
|
||||
mocked_get_tenants.return_value = [tenant]
|
||||
command.handle(redeploy=True)
|
||||
assert command.deploy.mock_calls == []
|
||||
|
||||
|
||||
@patch('hobo.agent.common.management.commands.hobo_deploy.TenantMiddleware.get_tenant_by_hostname')
|
||||
def test_deploy(mocked_get_tenant_by_hostname, tmpdir):
|
||||
command = Command()
|
||||
command.deploy_specifics = Mock()
|
||||
tenant = Mock()
|
||||
tenant.get_directory = Mock(return_value=str(tmpdir))
|
||||
base_url = 'https://combo.dev.publik.love/'
|
||||
tenant_hobo_json = os.path.join(str(tmpdir), 'hobo.json')
|
||||
ENVIRONMENT = {'services': [{'service-id': 'combo', 'base_url': base_url}],
|
||||
'timestamp': '001'}
|
||||
|
||||
def assert_deployed():
|
||||
assert mocked_get_tenant_by_hostname.mock_calls[0] == call('combo.dev.publik.love')
|
||||
assert command.deploy_specifics.mock_calls == [
|
||||
call({'services': [{'base_url': 'https://combo.dev.publik.love/',
|
||||
'service-id': 'combo',
|
||||
'this': True,}],
|
||||
'timestamp': '001'},
|
||||
tenant)]
|
||||
with open(tenant_hobo_json, 'r') as handler: # hobo.json file
|
||||
content = json.load(handler)
|
||||
assert ENVIRONMENT['services'][0]['this'] is True # new entry added
|
||||
assert json.dumps(content, sort_keys=True), json.dumps(ENVIRONMENT, sort_keys=True)
|
||||
|
||||
# create tenant first
|
||||
command.deploy_specifics.reset_mock()
|
||||
mocked_get_tenant_by_hostname.reset_mock()
|
||||
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_deployed()
|
||||
|
||||
# already there (timestamp do not change)
|
||||
command.deploy_specifics.reset_mock()
|
||||
mocked_get_tenant_by_hostname.side_effect = [tenant]
|
||||
command.deploy(base_url, ENVIRONMENT, None)
|
||||
assert command.deploy_specifics.mock_calls == []
|
||||
|
||||
# force re-deploy
|
||||
command.deploy_specifics.reset_mock()
|
||||
mocked_get_tenant_by_hostname.reset_mock()
|
||||
mocked_get_tenant_by_hostname.side_effect = [tenant]
|
||||
command.deploy(base_url, ENVIRONMENT, True)
|
||||
assert_deployed()
|
||||
|
||||
# early exit, we don't redeploy secondary services
|
||||
command.deploy_specifics.reset_mock()
|
||||
env = dict(ENVIRONMENT)
|
||||
env['services'][0]['secondary'] = True
|
||||
command.deploy(base_url, env, None)
|
||||
assert command.deploy_specifics.mock_calls == []
|
||||
|
||||
|
||||
def test_deploy_specific():
|
||||
"""stupid test"""
|
||||
command = Command()
|
||||
command.generate_saml_keys = Mock()
|
||||
command.configure_service_provider = Mock()
|
||||
command.configure_theme = Mock()
|
||||
command.configure_template = Mock()
|
||||
|
||||
command.deploy_specifics('my_hobo_env', 'my_tenant')
|
||||
assert command.generate_saml_keys.mock_calls == [call('my_tenant', prefix='sp-')]
|
||||
assert command.configure_service_provider.mock_calls == [call('my_hobo_env', 'my_tenant')]
|
||||
assert command.configure_theme.mock_calls == [call('my_hobo_env', 'my_tenant')]
|
||||
assert command.configure_template.mock_calls == [call('my_hobo_env', 'my_tenant')]
|
||||
|
||||
|
||||
def test_generate_saml_keys(tmpdir):
|
||||
"""create an RSA key and its x509 certificate"""
|
||||
command = Command()
|
||||
tenant = Mock()
|
||||
tenant.get_directory = Mock(return_value=str(tmpdir))
|
||||
tenant.domain_url = 'combo.dev.publik.love'
|
||||
|
||||
command.generate_saml_keys(tenant)
|
||||
with open('%s/saml.key' % str(tmpdir), 'r') as handler:
|
||||
key = handler.read()
|
||||
with open('%s/saml.crt' % str(tmpdir), 'r') as handler:
|
||||
crt = handler.read()
|
||||
|
||||
# if files exist don't regenerate them
|
||||
command.generate_saml_keys(tenant)
|
||||
with open('%s/saml.key' % str(tmpdir), 'r') as handler:
|
||||
assert key == handler.read()
|
||||
with open('%s/saml.crt' % str(tmpdir), 'r') as handler:
|
||||
assert crt == handler.read()
|
||||
|
||||
|
||||
@patch('hobo.agent.common.management.commands.hobo_deploy.requests.get')
|
||||
def test_configure_service_provider(mocked_get, tmpdir):
|
||||
"""create TENANT/idp-metadata-ID.xml file"""
|
||||
command = Command()
|
||||
tenant = Mock()
|
||||
tenant.get_directory = Mock(return_value=str(tmpdir))
|
||||
response1 = Response()
|
||||
response1._content = 'my saml idp metadata (1)'
|
||||
response1.status_code = 200
|
||||
response2 = Response()
|
||||
response2._content = 'my saml idp metadata (2)'
|
||||
response2.status_code = 200
|
||||
tenant_idp_metadata = '%s/idp-metadata-%s.xml' % (str(tmpdir), '1')
|
||||
|
||||
ENVIRONMENT = {'services': [
|
||||
{'service-id': 'combo',
|
||||
'saml-idp-metadata-url': 'https://combo.dev.publik.love/accounts/mellon/metadata/',
|
||||
'id': 1},
|
||||
{'service-id': 'wcs',
|
||||
'saml-idp-metadata-url': 'https://wcs.dev.publik.love/saml/metadata',
|
||||
'id': 1},
|
||||
]}
|
||||
|
||||
# normal case (stop when configuration on a service success for this tenant)
|
||||
mocked_get.side_effect = [response1]
|
||||
command.configure_service_provider(ENVIRONMENT, tenant)
|
||||
with open(tenant_idp_metadata, 'r') as handler:
|
||||
assert handler.read() == 'my saml idp metadata (1)'
|
||||
|
||||
# no 'idp_url' JSON entry
|
||||
env = {'services': [{'service-id': 'combo', 'id': 1}]}
|
||||
os.remove(tenant_idp_metadata)
|
||||
command.configure_service_provider(env, tenant)
|
||||
with pytest.raises(IOError, match='No such file or directory'):
|
||||
open(tenant_idp_metadata, 'r')
|
||||
|
||||
# idp not available
|
||||
response1.status_code = 500
|
||||
mocked_get.side_effect = [exceptions.RequestException, response1]
|
||||
command.configure_service_provider(ENVIRONMENT, tenant)
|
||||
with pytest.raises(IOError, match='No such file or directory'):
|
||||
open(tenant_idp_metadata, 'r')
|
||||
|
||||
# case when idp is becoming available
|
||||
mocked_get.side_effect = [response1, response2]
|
||||
command.configure_service_provider(ENVIRONMENT, tenant)
|
||||
with open(tenant_idp_metadata, 'r') as handler:
|
||||
assert handler.read() == 'my saml idp metadata (2)'
|
||||
|
||||
|
||||
def test_get_theme(fake_themes):
|
||||
"""return the service's theme"""
|
||||
COMBO = {'service-id': 'combo'}
|
||||
WCS = {'service-id': 'hobo', 'variables': {'theme': 'alfortville'}}
|
||||
HOBO = {'service-id': 'hobo', 'variables': {'theme': 'unknown_theme'}}
|
||||
ENVIRONMENT = {'services': [COMBO, WCS, HOBO], 'variables': {'theme': 'publik'}}
|
||||
command = Command()
|
||||
|
||||
# main case
|
||||
command.me = COMBO
|
||||
assert command.get_theme(ENVIRONMENT)['id'] == 'publik'
|
||||
|
||||
# service having theme overloaded
|
||||
command.me = WCS
|
||||
assert command.get_theme(ENVIRONMENT)['id'] == 'alfortville'
|
||||
|
||||
# theme not specified
|
||||
command.me = COMBO
|
||||
assert command.get_theme({'services': [COMBO, WCS, HOBO]}) is None
|
||||
|
||||
# unknown theme
|
||||
command.me = HOBO
|
||||
assert command.get_theme(ENVIRONMENT) is None
|
||||
|
||||
# can't test last return case as it could never append
|
||||
# if not theme.get('module'): # TODO: dead code to remove
|
||||
|
||||
|
||||
def test_configure_theme(fake_themes, tmpdir):
|
||||
"""make symlink for TENANT/theme
|
||||
and TENANT/static, TENANT/templates on overlay
|
||||
"""
|
||||
COMBO = {'service-id': 'combo'}
|
||||
WCS = {'service-id': 'hobo', 'variables': {'theme': 'alfortville'}} # overlay on fixture
|
||||
HOBO = {'service-id': 'hobo', 'variables': {'theme': 'unknown_theme'}}
|
||||
ENVIRONMENT = {'services': [COMBO, WCS, HOBO], 'variables': {'theme': 'publik'}}
|
||||
command = Command()
|
||||
command.me = COMBO
|
||||
|
||||
tenant = Mock()
|
||||
tenant.get_directory = Mock(return_value=str(tmpdir))
|
||||
theme_path = '%s/theme' % str(tmpdir)
|
||||
static_path = '%s/static' % str(tmpdir)
|
||||
templates_path = '%s/templates' % str(tmpdir)
|
||||
|
||||
# main case: (remove 'static' and/or 'templates' links and empty directories too)
|
||||
os.mkdir(static_path)
|
||||
os.symlink(str(tmpdir), templates_path)
|
||||
command.configure_theme(ENVIRONMENT, tenant)
|
||||
assert os.path.islink(theme_path)
|
||||
assert os.path.exists(static_path) is False
|
||||
assert os.path.exists(templates_path) is False
|
||||
|
||||
# overlay provided: build links to the 'overlay' directory
|
||||
os.symlink(str(tmpdir), static_path)
|
||||
os.mkdir(templates_path)
|
||||
command.me = WCS
|
||||
command.configure_theme(ENVIRONMENT, tenant)
|
||||
assert os.readlink(static_path) == '%s/themes/foobar/static' % str(tmpdir)
|
||||
assert os.readlink(templates_path) == '%s/themes/foobar/templates' % str(tmpdir)
|
||||
|
||||
# retrieve on error (if server hangs during 'atomic_symlink')
|
||||
os.unlink(theme_path)
|
||||
os.symlink(str(tmpdir), theme_path + '.tmp')
|
||||
command.configure_theme(ENVIRONMENT, tenant)
|
||||
assert os.path.islink(theme_path)
|
||||
assert os.path.islink(static_path) # to compare with result we get bellow
|
||||
|
||||
# do not 'overlay' non-empty 'static' and/or 'templates' true directories
|
||||
os.unlink(static_path)
|
||||
os.mkdir(static_path)
|
||||
os.mkdir(os.path.join(static_path, 'some_dir'))
|
||||
command.configure_theme(ENVIRONMENT, tenant)
|
||||
assert os.path.isdir(static_path)
|
||||
|
||||
# no theme provided
|
||||
os.unlink(theme_path)
|
||||
command.me = HOBO
|
||||
command.configure_theme(ENVIRONMENT, tenant)
|
||||
assert os.path.exists(theme_path) is False
|
||||
|
||||
|
||||
@patch('hobo.agent.common.management.commands.hobo_deploy.get_commands')
|
||||
@patch('hobo.agent.common.management.commands.hobo_deploy.call_command')
|
||||
def test_configure_template(mocked_call_command, mocked_get_commands):
|
||||
command = Command()
|
||||
ENVIRONMENT = {'services': [{'service-id': 'combo',
|
||||
'template_name': 'my_template',
|
||||
'this': True}]}
|
||||
|
||||
# import_template.py located into hobo/agent/common: always belongs to command object
|
||||
# TODO: dead condition
|
||||
mocked_get_commands.return_value = ['import_template', '...']
|
||||
|
||||
# main case
|
||||
command.configure_template(ENVIRONMENT, 'unused') # TODO: dead tenant parameter
|
||||
assert mocked_call_command.mock_calls == [call('import_template', 'my_template')]
|
||||
|
||||
# no template_name entry provided
|
||||
mocked_call_command.reset_mock()
|
||||
del ENVIRONMENT['services'][0]['template_name']
|
||||
command.configure_template(ENVIRONMENT, 'unused')
|
||||
assert mocked_call_command.mock_calls == []
|
||||
|
||||
|
||||
def test_combo_get_theme(fake_themes):
|
||||
# IMHO, JSON's 'this' entry can't miss
|
||||
# cf: hobo/agent/common/management/commands/hobo_deploy.py::configure_template
|
||||
# cf: hobo/agent/hobo/management/commands/hobo_deploy.py::deploy_specifics
|
||||
# TODO: dead file (I mean it should be removed)
|
||||
|
||||
COMBO = {'service-id': 'combo', 'this': True}
|
||||
ENVIRONMENT = {'services': [COMBO], 'variables': {'theme': 'publik'}}
|
||||
command = ComboCommand()
|
||||
|
||||
# main case
|
||||
assert command.get_theme(ENVIRONMENT)['id'] == 'publik'
|
||||
|
||||
|
||||
@patch('hobo.agent.common.management.commands.hobo_deploy.Command.deploy_specifics')
|
||||
@patch('hobo.agent.hobo.management.commands.hobo_deploy.tenant_context')
|
||||
def test_hobo_deploy_specifics_alone(mocked_tenant_context, mocked_super, db):
|
||||
"""build database content used later to generate hobo.json.
|
||||
tests using tenants are located into tests_multipublik/test_multipublik.py
|
||||
"""
|
||||
command = HoboCommand()
|
||||
ENVIRONMENT = {'services': [{'service-id': 'hobo',
|
||||
'slug': 'hobo',
|
||||
'title': 'Hobo',
|
||||
'base_url': 'https://hobo.dev.publik.love/',
|
||||
'this': True},
|
||||
{'service-id': 'combo',
|
||||
'slug': 'portal',
|
||||
'title': 'Compte citoyen',
|
||||
'base_url': 'https://combo.dev.publik.love/'},
|
||||
{'service-id': 'wcs',
|
||||
'slug': 'eservices',
|
||||
'title': 'D\u00e9marches',
|
||||
'base_url': 'wcs.dev.publik.love'},
|
||||
{'service-id': 'wrong id',
|
||||
'base_url': 'https://wrong.dev.publik.love/'}]}
|
||||
|
||||
command.deploy_specifics(ENVIRONMENT, 'tenant')
|
||||
|
||||
assert mocked_super.mock_calls == [call(ENVIRONMENT, 'tenant')]
|
||||
assert len(Hobo.objects.all()) == 0
|
||||
assert len(Combo.objects.all()) == 1
|
||||
assert len(Wcs.objects.all()) == 1
|
||||
combo = Combo.objects.all()[0]
|
||||
assert combo.base_url == 'https://combo.dev.publik.love/'
|
||||
assert combo.slug == '_interco_portal'
|
||||
|
||||
# define global variables (twice)
|
||||
ou_label = Variable.objects.get(name='ou-label')
|
||||
ou_slug = Variable.objects.get(name='ou-slug')
|
||||
assert ou_label.value == 'Hobo'
|
||||
assert ou_slug.value == 'hobo'
|
||||
assert ou_label.service_pk is None
|
||||
assert ou_slug.service_pk is None
|
||||
|
||||
|
||||
@patch('hobo.agent.common.management.commands.hobo_deploy.Command.deploy_specifics')
|
||||
@patch('hobo.agent.hobo.management.commands.hobo_deploy.tenant_context')
|
||||
def test_hobo_deploy_specifics_primary(mocked_tenant_context, mocked_super, db):
|
||||
"""first deployement: do nothing for secondary hobo-services
|
||||
same behaviour as previous 'alone' test
|
||||
"""
|
||||
command = HoboCommand()
|
||||
wcs_url = 'https://wcs.dev.publik.love/'
|
||||
ENVIRONMENT = {'services': [{'service-id': 'hobo',
|
||||
'slug': 'hobo',
|
||||
'title': 'Hobo primary',
|
||||
'base_url': 'https://hobo1.dev.publik.love/',
|
||||
'this': True},
|
||||
{'service-id': 'combo',
|
||||
'slug': 'portal',
|
||||
'title': 'Compte citoyen',
|
||||
'base_url': 'https://combo.dev.publik.love/'},
|
||||
{'service-id': 'wcs',
|
||||
'slug': 'eservices',
|
||||
'title': 'skipped as url already in use',
|
||||
'base_url': wcs_url},
|
||||
{'service-id': 'hobo',
|
||||
'slug': 'hobo',
|
||||
'title': 'Hobo secondary',
|
||||
'base_url': 'https://hobo2.dev.publik.love/',
|
||||
'secondary': True}]}
|
||||
|
||||
# make the wcs service already deployed
|
||||
Wcs.objects.get_or_create(base_url=wcs_url, secondary=False)
|
||||
|
||||
command.deploy_specifics(ENVIRONMENT, 'tenant')
|
||||
|
||||
assert mocked_super.mock_calls == [call(ENVIRONMENT, 'tenant')]
|
||||
assert len(Hobo.objects.all()) == 0
|
||||
assert len(Combo.objects.all()) == 1
|
||||
assert len(Wcs.objects.all()) == 1 # the wcs service already set upper
|
||||
combo = Combo.objects.all()[0]
|
||||
assert combo.base_url == 'https://combo.dev.publik.love/'
|
||||
assert combo.slug == '_interco_portal'
|
||||
|
||||
# define global variables
|
||||
ou_label = Variable.objects.get(name='ou-label')
|
||||
ou_slug = Variable.objects.get(name='ou-slug')
|
||||
assert ou_label.value == 'Hobo primary'
|
||||
assert ou_slug.value == 'hobo'
|
||||
assert ou_label.service_pk is None
|
||||
assert ou_slug.service_pk is None
|
||||
|
||||
|
||||
@patch('hobo.agent.common.management.commands.hobo_deploy.Command.deploy_specifics')
|
||||
@patch('hobo.agent.hobo.management.commands.hobo_deploy.tenant_context')
|
||||
def test_hobo_deploy_specifics_secondary(mocked_tenant_context, mocked_super, db):
|
||||
"""next deployements: skip generic tasks and add 'ou' variables
|
||||
"""
|
||||
command = HoboCommand()
|
||||
ENVIRONMENT = {'services': [{'service-id': 'hobo',
|
||||
'slug': 'hobo',
|
||||
'title': 'Hobo primary',
|
||||
'base_url': 'https://hobo1.dev.publik.love/'},
|
||||
{'service-id': 'combo',
|
||||
'slug': 'portal',
|
||||
'title': 'Compte citoyen',
|
||||
'base_url': 'https://combo.dev.publik.love/'},
|
||||
{'service-id': 'hobo',
|
||||
'slug': 'hobo',
|
||||
'title': 'Hobo secondary',
|
||||
'base_url': 'https://hobo2.dev.publik.love/',
|
||||
'secondary': True,
|
||||
'this': True}],
|
||||
'variables': {'ou-label': 'my ou-label',
|
||||
'ou-slug': 'my ou-slug'}}
|
||||
|
||||
# assert primary hobo is wanted
|
||||
command.deploy_specifics(ENVIRONMENT, 'tenant')
|
||||
assert len(Combo.objects.all()) == 0
|
||||
|
||||
# make primary hobo already deployed
|
||||
Hobo.objects.get_or_create(base_url='hobo1.dev.publik.love', secondary=False)
|
||||
command.deploy_specifics(ENVIRONMENT, 'tenant')
|
||||
|
||||
assert mocked_super.mock_calls == []
|
||||
assert len(Hobo.objects.all()) == 1 # primary hobo set upper
|
||||
assert len(Combo.objects.all()) == 1
|
||||
combo = Combo.objects.all()[0]
|
||||
assert combo.base_url == 'https://combo.dev.publik.love/'
|
||||
assert combo.slug == '_my ou-slug_portal'
|
||||
|
||||
# variables are added to the services
|
||||
ou_label = Variable.objects.get(name='ou-label')
|
||||
ou_slug = Variable.objects.get(name='ou-slug')
|
||||
assert ou_label.value == 'my ou-label'
|
||||
assert ou_slug.value == 'my ou-slug'
|
||||
assert ou_label.service_pk == combo.id
|
||||
assert ou_slug.service_pk == combo.id
|
||||
|
||||
|
||||
# Can't write test_authentic2_deploy_specifics as cannot import A2Command here:
|
||||
#from hobo.agent.authentic2.management.commands.hobo_deploy import Command as A2Command
|
||||
# -> tests_authentic/test_hobo_deploy.py
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
{
|
||||
"services": [
|
||||
{
|
||||
"service-id": "chrono",
|
||||
"base_url": "https://chrono.dev.publik.love/",
|
||||
"slug": "agendas",
|
||||
"title": "CHRONO",
|
||||
"secret_key": "123",
|
||||
"template_name": "import_me.txt"
|
||||
},
|
||||
{
|
||||
"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.txt"
|
||||
},
|
||||
{
|
||||
"service-id": "authentic",
|
||||
"base_url": "https://authentic.dev.publik.love/",
|
||||
"slug": "idp",
|
||||
"title": "A2",
|
||||
"secret_key": "123"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
|
@ -12,3 +12,4 @@ DATABASE_ROUTERS = ('tenant_schemas.routers.TenantSyncRouter',)
|
|||
INSTALLED_APPS = ('hobo.multitenant', 'hobo') + INSTALLED_APPS
|
||||
SHARED_APPS = ()
|
||||
TENANT_APPS = INSTALLED_APPS
|
||||
PROJECT_NAME = 'testing'
|
||||
|
|
|
|||
|
|
@ -0,0 +1,91 @@
|
|||
import os
|
||||
|
||||
import mock
|
||||
from django.core.management import call_command, get_commands, load_command_class
|
||||
|
||||
from hobo.environment.models import Variable
|
||||
from hobo.multitenant.middleware import TenantMiddleware
|
||||
from tenant_schemas.utils import tenant_context
|
||||
|
||||
|
||||
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)
|
||||
|
||||
|
||||
@mock.patch('hobo.agent.common.management.commands.hobo_deploy.call_command')
|
||||
@mock.patch('hobo.agent.common.management.commands.hobo_deploy.get_commands')
|
||||
def test_import_template(mocked_get_commands, mocked_call_command, db):
|
||||
"""check 'hobo-deploy' result in '$ chrono-manage import_template' call
|
||||
$ chrono-manage hobo_deploy env.json # simulate this bash query
|
||||
|
||||
warning:
|
||||
this test must be the first one executed else 'get_commands' fails to be mocked
|
||||
"""
|
||||
command = load_command_class('hobo.agent.common', 'hobo_deploy')
|
||||
domain = 'chrono.dev.publik.love'
|
||||
|
||||
def my_call_command(command, parameter):
|
||||
if command == 'import_template':
|
||||
my_call_command.import_template_was_called = True
|
||||
return
|
||||
call_command(command, parameter)
|
||||
|
||||
mocked_get_commands.return_value = ['import_template']
|
||||
mocked_call_command.side_effect = my_call_command
|
||||
my_call_command.import_template_was_called = False
|
||||
|
||||
command.handle('https://%s/' % domain, 'tests_schemas/env.json')
|
||||
assert_deployed(domain)
|
||||
|
||||
# assert the 'import_template' command was called
|
||||
assert my_call_command.import_template_was_called is True
|
||||
|
||||
|
||||
def test_deploy_on_common_agent(db):
|
||||
"""deploy basic case (for chrono here):
|
||||
$ chrono-manage hobo_deploy env.json # simulate this bash query
|
||||
"""
|
||||
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)
|
||||
|
||||
|
||||
def test_deploy_specifics_on_hobo_agent(db):
|
||||
"""overloaded case for hobo:
|
||||
$ hobo-manage hobo-deploy env.json # simulate this bash query
|
||||
"""
|
||||
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)
|
||||
|
||||
# reach deploy_specifics() in hobo/agent/hobo/management/commands/hobo_deploy.py
|
||||
tenant = TenantMiddleware.get_tenant_by_hostname(domain)
|
||||
with tenant_context(tenant):
|
||||
assert Variable.objects.get(name='ou-label').value == 'HOBO'
|
||||
|
||||
|
||||
# fails to simulate call from authentic2 agent, that overload deploy_specifics()
|
||||
# $ authentic-multitenant-manage hobo-deploy
|
||||
# here, because this agent need to import authentic2 modules
|
||||
|
||||
|
||||
# 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_get_theme_on_combo_agent(db):
|
||||
"""overloaded case for combo:
|
||||
$ combo-manage hobo-deploy env.json # simulate this bash query
|
||||
"""
|
||||
command = load_command_class('hobo.agent.combo', 'hobo_deploy')
|
||||
domain = 'combo.dev.publik.love'
|
||||
command.handle('https://%s/' % domain, 'tests_schemas/env.json')
|
||||
assert_deployed(domain)
|
||||
# lack an assert statement proving we do reach
|
||||
# get_theme() in hobo/agent/combo/management/commands/hobo_deploy.py
|
||||
Loading…
Reference in New Issue