Produits Entr'ouvert » Hobo

defined, the `execute()` method of the instance is called, passing it his

`base_url` and the full environment dictionary.

adapted in the AUTHENTIC_MANAGE_COMMAND setting. It should be run with the

""" 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

{

    "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"

	}

    ]

}

PROJECT_NAME = 'testing'

import os

from django.core.management import 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)

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

import mock

import os

from django.core.management import call_command, load_command_class

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)

@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

    this test must be isolated from the others on test_deploy.py

    else 'get_commands' which is lru_cached 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