Produits Entr'ouvert » Hobo

import json

from django.contrib.auth.models import User

from django.contrib.contenttypes.models import ContentType

from hobo.environment.models import Authentic, Combo, ServiceBase, Variable

from hobo.profile.models import AttributeDefinition

def test_handle():

    expected = "[call('recipe_me.json')]"

    kwargs = {'verbosity': 'verbosity value',

              'timeout': 'timeout value',

              'permissive': 'permissive value'}

    command = Command()

    command.run_cook = mock.Mock()

    command.handle('recipe_me.json', **kwargs)

    assert command.verbosity == 'verbosity value'

    assert command.timeout == 'timeout value'

    assert command.permissive == 'permissive value'

    assert str(command.run_cook.mock_calls) == expected

@mock.patch('hobo.environment.management.commands.cook.open')

@mock.patch('hobo.environment.management.commands.cook.notify_agents')

def test_run_cook(mocked_notify_agents, mocked_open):

    receipe = """{

  "steps": [

    {"create-hobo": {

      "url": "https://entrouvert.org/"

    }}

  ]

}"""

    expected1 = "[call(u'create-hobo', {u'url': u'https://entrouvert.org/'})]"

    expected2 = "[call(url=u'https://entrouvert.org/')]"

    expected3 = "[]"

    expected4 = "[call(timeout=42)]"

    command = Command()

    command.permissive = False

    command.must_notify = True

    command.timeout = 42

    command.check_action = mock.Mock()

    command.create_hobo = mock.Mock()

    mocked_notify_agents = mock.Mock()

    command.wait_operationals = mock.Mock()

    mocked_open.side_effect = [StringIO.StringIO(receipe)]

    command.run_cook('recipe_me.json')

    assert str(command.check_action.mock_calls) == expected1

    assert str(command.create_hobo.mock_calls) == expected2

    assert str(mocked_notify_agents.mock_calls) == expected3

    assert str(command.wait_operationals.mock_calls) == expected4

    command.create_authentic = mock.Mock()

@mock.patch('hobo.environment.management.commands.cook.open')

def test_load_variables_from(mocked_open):

    """load variables from a file"""

    receipe = """{

  "load-variables-from": "variables.json",

  "variables": {

    "domain": "entrouvert.org",

    "bar": "bar2"

  },

  "steps": [

    {"create-hobo": {

      "url": "https://${domain}/"

    }},

    {"create-authentic": {

      "url": "https://${foo}/"

    }},

    {"create-combo": {

      "url": "https://${bar}/",

      "not_a_string": []

    }}

  ]

}"""

    variables = """{

  "foo": "foo1",

  "bar": "bar1"

}"""

    expected_hobo = "[call(url=u'https://entrouvert.org/')]"

    expected_a2 = "[call(url=u'https://foo1/')]"

    expected_combo = "[call(not_a_string=[], url=u'https://bar2/')]"

    command = Command()

    command.permissive = True

    command.create_hobo = mock.Mock()

    command.create_authentic = mock.Mock()

    command.create_combo = mock.Mock()

    mocked_open.side_effect = [StringIO.StringIO(receipe),

                               StringIO.StringIO(variables)]

    command.run_cook('recipe_me.json')

    assert str(command.create_hobo.mock_calls) == expected_hobo

    assert str(command.create_authentic.mock_calls) == expected_a2

    assert str(command.create_combo.mock_calls) == expected_combo

def test_wait_operationals(db, monkeypatch):

    service1 = mock.Mock()

    service2 = mock.Mock()

    obj1 = mock.Mock()

    obj2 = mock.Mock()

    def do_nothing(*args, **kwargs):

        pass

    obj1.check_operational = do_nothing

    obj2.check_operational = do_nothing

    obj1.base_url = 'url1'

    obj2.base_url = 'url2'

    service1.objects.all = mock.Mock(return_value = [obj1])

    service2.objects.all = mock.Mock(return_value = [obj2])

    monkeypatch.setattr(

        'hobo.environment.management.commands.cook.AVAILABLE_SERVICES',

        [service1, service2])

    command = Command()

    # already operational

    obj1.last_operational_success_timestamp = 'some date'

    obj2.last_operational_success_timestamp = 'some date'

    command.wait_operationals(2)

    assert True

    # not operational

    obj2.last_operational_success_timestamp = None

    with pytest.raises(CommandError) as e_info:

        command.wait_operationals(.6)

    assert str(e_info).find('CommandError: timeout waiting for url2') != -1

def test_set_variable(db):

    command = Command()

    command.set_variable('foo', 'bar')

    value = Variable.objects.get(name='foo').value

    label = Variable.objects.get(name='foo').label

    assert value == 'bar'

    assert label == 'foo'

    command.set_variable('foo', 'bar', label='FOO')

    value = Variable.objects.get(name='foo').value

    label = Variable.objects.get(name='foo').label

    assert value == 'bar'

    assert label == 'FOO'

    command.set_variable('foo', ['bar', 'baz'])

    value = Variable.objects.get(name='foo').value

    assert value == '["bar", "baz"]'

    command.set_variable('foo', {'key1': 'bar', 'key2': 'baz'})

    value = Variable.objects.get(name='foo').value

    ordered_dump = json.dumps(json.loads(value), sort_keys=True)

    assert  ordered_dump == '{"key1": "bar", "key2": "baz"}'

def test_set_attribute(db):

    command = Command()

    command.set_attribute('foo_name', 'foo_label')

    values = AttributeDefinition.objects.filter(name='foo_name')

    assert values.count() == 1

    assert values[0].label == 'foo_label'

    assert values[0].disabled is False

    command.set_attribute('foo_name', 'foo_label', disabled=True)

    values = AttributeDefinition.objects.filter(name='foo_name')

    assert values.count() == 1

    assert values[0].label == 'foo_label'

    assert values[0].disabled is True

def test_disable_attribute(db):

    command = Command()

    command.set_attribute('foo_name', 'foo_label')

    values = AttributeDefinition.objects.filter(name='foo_name')

    assert values.count() == 1

    assert values[0].label == 'foo_label'

    assert values[0].disabled is False

    command.disable_attribute('foo_name')

    values = AttributeDefinition.objects.filter(name='foo_name')

    assert values[0].disabled is True

    command.disable_attribute('not_defined')

    values = AttributeDefinition.objects.filter(name='not_defined')

    assert values.count() == 0

def test_enable_attribute(db):

    command = Command()

    command.set_attribute('foo_name', 'foo_label', disabled=True)

    values = AttributeDefinition.objects.filter(name='foo_name')

    assert values.count() == 1

    assert values[0].label == 'foo_label'

    assert values[0].disabled is True

    command.enable_attribute('foo_name')

    values = AttributeDefinition.objects.filter(name='foo_name')

    assert values[0].disabled is False

    command.enable_attribute('not_defined')

    values = AttributeDefinition.objects.filter(name='not_defined')

    assert values.count() == 0

def test_create_superuser(db):

    command = Command()

    command.create_superuser()

    assert User.objects.count() == 1

    user = User.objects.all()[0]

    assert user.username == 'admin'

    assert user.is_superuser is True

def test_create_site(db):

    """

    code redondant ?

      service_type=obj_type,

      service_pk=obj.id,

    """

    command = Command()

    base_url = 'http://entrouvert.org'

    title = 'site title'

    slug = None

    template_name = ''

    variables = {'foo': {'label': 'FOO', 'value': {'key': 'bar'}}}

    command.create_site(Combo, base_url, title, slug, template_name,

                        variables)

    # Combo object

    assert Combo.objects.count() == 1

    combo = Combo.objects.all()[0]

    assert combo.title == title

    assert combo.base_url == base_url + '/'

    assert combo.template_name == ''

    # ContentType object

    obj_type = ContentType.objects.get_for_model(Combo)

    # Variables

    variable = Variable.objects.get(name='foo')

    assert variable.label == 'FOO'

    assert variable.value == '{"key": "bar"}'

    assert variable.service_type == obj_type

    assert variable.service_pk == combo.id

    with pytest.raises(CommandError) as e_info:

        command.create_site(Combo, 'unvalid_url', 'site title', 'a slug', '', '')

    assert 'Enter a valid URL.' in str(e_info.value)

@mock.patch('hobo.environment.management.commands.cook.connection')

@mock.patch('hobo.environment.management.commands.cook.call_command')

@mock.patch('hobo.environment.management.commands.cook.TenantMiddleware')

def test_create_hobo_primary(mocked_TenantMiddleware, mocked_call_command,

                             mocked_connection):

    expected1 = '[]'

    expected2 = "[call('create_hobo_tenant', 'entrouvert.org')]"

    expected3 = "[call('/bof/base_url', 'w'),\n" + \

      " call().write('http://entrouvert.org/and_much'),\n" + \

      " call().close()]"

    command = Command()

    command.create_site = mock.Mock()

    tenant = mock.Mock()

    tenant.schema_name = 'public'

    tenant.get_directory = mock.Mock(return_value='/bof')

    mocked_connection.get_tenant = mock.Mock(return_value=tenant)

    mocked_connection.set_tenant = mock.Mock()

    mocked_TenantMiddleware.get_tenant_by_hostname = mock.Mock(return_value=tenant)

    mocked_call_command.side_effect = CommandError

    mocked_open = mock.mock_open()

    with mock.patch('hobo.environment.management.commands.cook.open', mocked_open,

                    create=True):

        command.create_hobo('http://entrouvert.org/and_much')

    assert str(command.create_site.mock_calls) == expected1

    assert str(mocked_call_command.mock_calls) == expected2

    assert len(mocked_connection.set_tenant.mock_calls) == 1

    assert str(mocked_open.mock_calls) == expected3

@mock.patch('hobo.environment.management.commands.cook.connection')

@mock.patch('hobo.environment.management.commands.cook.call_command')

@mock.patch('hobo.environment.management.commands.cook.TenantMiddleware')

def test_create_hobo_not_primary(mocked_TenantMiddleware, mocked_call_command,

                     mocked_connection):

    command = Command()

    command.create_site = mock.Mock()

    tenant = mock.Mock()

    tenant.schema_name = 'public'

    tenant.get_directory = mock.Mock(return_value='/bof')

    mocked_connection.get_tenant = mock.Mock(return_value=tenant)

    mocked_connection.set_tenant = mock.Mock()

    mocked_TenantMiddleware.get_tenant_by_hostname = mock.Mock(return_value=tenant)

    expected1 = "[call(<class 'hobo.environment.models.Hobo'>, " + \

      "'http://entrouvert.org/and_much', None, u'hobo-none', " + \

      "template_name='', variables=None)]"

    expected2 = "[]"

    expected3 = "[]"

    tenant.schema_name = 'other than public'

    mocked_open = mock.mock_open()

    with mock.patch('hobo.environment.management.commands.cook.open', mocked_open,

                    create=True):

        command.create_hobo('http://entrouvert.org/and_much')

    assert str(command.create_site.mock_calls) == expected1

    assert str(mocked_call_command.mock_calls) == expected2

    assert len(mocked_connection.set_tenant.mock_calls) == 1

    assert str(mocked_open.mock_calls) == expected3

def test_create_services():

    endings = ", 'url', 'title', None, '', None)"

    expected = "[call(<class 'hobo.environment.models.Authentic'>" + endings

    expected += ",\n call(<class 'hobo.environment.models.Combo'>" + endings

    expected += ",\n call(<class 'hobo.environment.models.Wcs'>" + endings

    expected += ",\n call(<class 'hobo.environment.models.Passerelle'>" + endings

    expected += ",\n call(<class 'hobo.environment.models.Fargo'>" + endings

    expected += ",\n call(<class 'hobo.environment.models.Welco'>" + endings

    expected += ",\n call(<class 'hobo.environment.models.Chrono'>" + endings

    expected += ",\n call(<class 'hobo.environment.models.Corbo'>" + endings

    expected += ",\n call(<class 'hobo.environment.models.BiJoe'>" + endings

    expected += "]"

    command = Command()

    command.create_site = mock.Mock()

    command.create_authentic('url', 'title')

    command.create_combo('url', 'title')

    command.create_wcs('url', 'title')

    command.create_passerelle('url', 'title')

    command.create_fargo('url', 'title')

    command.create_welco('url', 'title')

    command.create_chrono('url', 'title')

    command.create_corbo('url', 'title')

    command.create_bijoe('url', 'title')

    assert len(command.create_site.mock_calls) == 9

    assert str(command.create_site.mock_calls) == expected

def test_set_idp(db):

    command = Command()

    # TODO: maybe we should handle this into cook.py ?

    with pytest.raises(Authentic.DoesNotExist,

                       match='Authentic matching query does not exist'):

        command.set_idp('url')

    with pytest.raises(IndexError, match='list index out of range'):

        command.set_idp()

    # objects sorted on title: [obj1, obj2]

    obj1, ignored = Authentic.objects.get_or_create(

        slug='slug1', defaults={'title': 'bar'})

    obj2, ignored = Authentic.objects.get_or_create(

        slug='slug2', defaults={'title': 'foo', 'base_url': 'http://example.org'})

    assert obj1.use_as_idp_for_self is False

    assert obj2.use_as_idp_for_self is False

    # set first

    command.set_idp('')

    obj = Authentic.objects.get(title='bar')

    assert obj.use_as_idp_for_self is True

    # set using url

    command.set_idp('http://example.org/')

    obj = Authentic.objects.get(title='foo')

    assert obj.use_as_idp_for_self is True

@mock.patch('hobo.environment.management.commands.cook.set_theme')

@mock.patch('hobo.environment.management.commands.cook.connection')

@mock.patch('hobo.agent.common.management.commands.hobo_deploy.Command.configure_theme')

def test_set_theme(mocked_configure_theme, mocked_connection, mocked_set_theme):

    expected1 = "[call('the theme')]"

    expected2 = "[call({'variables': {'theme': 'the theme'}}, 'the tenant')]"

    mocked_connection.get_tenant = mock.Mock(return_value='the tenant')

    command = Command()

    command.set_theme('the theme')

    assert str(mocked_set_theme.mock_calls) == expected1

    assert len(mocked_connection.get_tenant.mock_calls) == 1

    assert str(mocked_configure_theme.mock_calls) == expected2

@mock.patch('hobo.environment.management.commands.cook.connection')

def test_cook(mocked_connection):

    expected1 = "[call('a-recipe-file.json')]"

    expected2 = "[call('the tenant')]"

    mocked_connection.get_tenant = mock.Mock(return_value='the tenant')

    mocked_connection.set_tenant = mock.Mock()

    command = Command()

    command.run_cook = mock.Mock()

    command.cook('a-recipe-file.json')

    assert len(mocked_connection.get_tenant.mock_calls) == 1

    assert str(command.run_cook.mock_calls) == expected1

    assert str(mocked_connection.set_tenant.mock_calls) == expected2