Produits Entr'ouvert » Hobo

stale roles, default is to not delete them.  

Tests

-----

For testing hobo server, do in a virtualenv:

   pip install pytest pytest-django

   DJANGO_SETTINGS_MODULE=hobo.settings HOBO_SETTINGS_FILE=tests/settings.py py.tests tests

For testing multitenant framework, do in a virtualenv:

   pip install pytest pytest-django memcached mock .

   cd tests_multitenant ; PYTHONPATH=. DJANGO_SETTINGS_MODULE=settings py.test .

from django.conf.urls import patterns, url

from django.http import HttpResponse

def helloworld(request):

    return HttpResponse('Hello world')

urlpatterns = patterns('',

    url(r'^$', helloworld),

)

import os

import tempfile

import shutil

import json

import pytest

@pytest.fixture(scope='function')

def tenants(db, request, settings):

    from hobo.multitenant.models import Tenant

    base = tempfile.mkdtemp('combo-tenant-base')

    settings.TENANT_BASE = base

    @pytest.mark.django_db

    def make_tenant(name):

        tenant_dir = os.path.join(base, name)

        os.mkdir(tenant_dir)

        with open(os.path.join(tenant_dir, 'unsecure'), 'w') as fd:

            fd.write('1')

        with open(os.path.join(tenant_dir, 'settings.json'), 'w') as fd:

            json.dump({'HOBO_TEST_VARIABLE': name}, fd)

        with open(os.path.join(tenant_dir, 'hobo.json'), 'w') as fd:

            json.dump({

                'variables': {

                    'hobo_test_variable': True,

                    'other_variable': 'foo',

                },

                'services': [

                    {'slug': 'test',

                     'title': 'Test',

                     'this': True,

                     'secret_key': '12345',

                     'base_url': 'http://%s' % name,

                     'saml-sp-metadata-url': 'http://%s/saml/metadata' % name,

                     'variables': {

                         'other_variable': 'bar',

                     }

                    },

                    {'slug': 'other',

                     'title': 'Other',

                     'base_url': 'http://other.example.net'},

                    ]}, fd)

        t = Tenant(domain_url=name,

                      schema_name=name.replace('-', '_').replace('.', '_'))

        t.create_schema()

        return t

    tenants = [make_tenant('tenant1.example.net'), make_tenant('tenant2.example.net')]

    def fin():

        from django.db import connection

        connection.set_schema_to_public()

        for t in tenants:

            t.delete(True)

        shutil.rmtree(base)

    request.addfinalizer(fin)

    return tenants

import os.path

import __builtin__ as builtin

from mock import mock_open, patch

LANGUAGE_CODE = 'en-us'

INSTALLED_APPS = ('django.contrib.auth', 'django.contrib.sessions', 'django.contrib.contenttypes')

PROJECT_NAME = 'fake-agent'

with patch.object(builtin, 'file', mock_open(read_data='xxx')):

    execfile(os.path.join(os.path.dirname(__file__), '../debian/debian_config_common.py'))

TENANT_APPS = ('django.contrib.auth','django.contrib.sessions', 'django.contrib.contenttypes', 'hobo.agent.common')

ROOT_URLCONF = 'hobo.agent.test_urls'

CACHES = {

    'default': {

            'BACKEND': 'hobo.multitenant.cache.TenantCache',

            'REAL_BACKEND': 'django.core.cache.backends.locmem.LocMemCache',

    }

}

# -*- coding: utf-8 -*-

import pytest

pytestmark = pytest.mark.django_db

def test_hobo_notify_roles(tenants):

    from hobo.agent.common.management.commands.hobo_notify import Command

    from tenant_schemas.utils import tenant_context

    from django.contrib.auth.models import Group

    from hobo.agent.common.models import Role

    # test wrong audience  

    for tenant in tenants:

        with tenant_context(tenant):

            notification = {

                u'@type': u'provision',

                u'audience': [u'http://coin.com/saml/metadata'],

                u'objects': [

                    {

                        u'@type': 'role',

                        u'uuid': u'12345',

                        u'name': u'Service petite enfance',

                        u'slug': u'service-petite-enfance',

                        u'description': u'Role du service petite enfance %s' % tenant.domain_url,

                    }

                ]

            }

            Command.process_notification(tenant, notification)

            assert Group.objects.count() == 0

            assert Role.objects.count() == 0

    # test provision

    for tenant in tenants:

        with tenant_context(tenant):

            notification = {

                u'@type': u'provision',

                u'audience': [u'%s/saml/metadata' % tenant.get_base_url()],

                u'objects': [

                    {

                        u'@type': 'role',

                        u'uuid': u'12345',

                        u'name': u'Service petite enfance',

                        u'slug': u'service-petite-enfance',

                        u'description': u'Role du service petite enfance %s' % tenant.domain_url,

                    }

                ]

            }

            Command.process_notification(tenant, notification)

            assert Group.objects.count() == 1

            assert Role.objects.count() == 1

            role = Role.objects.get()

            assert role.uuid == u'12345'

            assert role.name == u'Service petite enfance'

            assert role.description == u'Role du service petite enfance %s' % tenant.domain_url

    # test full provisionning

    for tenant in tenants:

        with tenant_context(tenant):

            notification = {

                u'@type': u'provision',

                u'full': True,

                u'audience': [u'%s/saml/metadata' % tenant.get_base_url()],

                u'objects': [

                    {

                        u'@type': 'role',

                        u'uuid': u'xyz',

                        u'name': u'Service état civil',

                        u'slug': u'service-etat-civil',

                        u'description': u'Role du service état civil %s' % tenant.domain_url,

                    }

                ]

            }

            Command.process_notification(tenant, notification)

            assert Group.objects.count() == 1

            assert Role.objects.count() == 1

            role = Role.objects.get()

            assert role.uuid == u'xyz'

            assert role.name == u'Service état civil'

            assert role.description == u'Role du service état civil %s' % tenant.domain_url

    # test deprovision

    for tenant in tenants:

        with tenant_context(tenant):

            notification = {

                u'@type': u'deprovision',

                u'audience': [u'%s/saml/metadata' % tenant.get_base_url()],

                u'objects': [

                    {

                        u'@type': 'role',

                        u'uuid': u'xyz',

                        u'name': u'Service état civil',

                        u'slug': u'service-etat-civil',

                        u'description': u'Role du service état civil %s' % tenant.domain_url,

                    }

                ]

            }

            Command.process_notification(tenant, notification)

            assert Group.objects.count() == 0

            assert Role.objects.count() == 0

import pytest

pytestmark = pytest.mark.django_db

def test_user_creation(tenants):

    from tenant_schemas.utils import tenant_context

    from django.contrib.auth import models

    for tenant in tenants:

        with tenant_context(tenant):

            models.User.objects.create(username=tenant.domain_url)

            assert models.User.objects.count() == 1

    for tenant in tenants:

        with tenant_context(tenant):

            assert models.User.objects.get().username == tenant.domain_url

import threading

import random

import pytest

from django.apps import apps

import django.conf

from django.core.handlers.base import BaseHandler

from django.core.wsgi import get_wsgi_application

import django.db

from django.core.cache import cache, caches

from tenant_schemas.utils import tenant_context

def test_tenant_middleware(tenants, client):

    res = client.get('/', SERVER_NAME='invalid.example.net')

    assert res.status_code == 404

    for tenant in tenants:

        res = client.get('/', SERVER_NAME=tenant.domain_url)

        assert res.status_code != 404

        assert res.wsgi_request.tenant.schema_name == tenant.schema_name

def test_tenant_json_settings(tenants, settings):

    settings.clear_tenants_settings()

    settings.default_settings.TENANT_SETTINGS_LOADERS = ('hobo.multitenant.settings_loaders.SettingsJSON', )

    # check the setting is not defined

    with pytest.raises(AttributeError):

        settings.HOBO_TEST_VARIABLE

    # check that for each tenant it contains the tenant domain

    # it's set by the tenants fixture in conftest.py

    for tenant in tenants:

        with tenant_context(tenant):

            assert django.conf.settings.HOBO_TEST_VARIABLE == tenant.domain_url

    # check it's no longer defined after going back to the public schema

    with pytest.raises(AttributeError):

        settings.HOBO_TEST_VARIABLE

def test_tenant_template_vars(tenants, settings, client):

    from hobo.multitenant.models import Tenant

    django.conf.settings.clear_tenants_settings()

    settings.default_settings.TENANT_SETTINGS_LOADERS = ('hobo.multitenant.settings_loaders.TemplateVars', )

    # check the setting is not defined

    with pytest.raises(AttributeError):

        django.conf.settings.TEMPLATE_VARS

    for tenant in tenants:

        with tenant_context(tenant):

            # check it's defined when moving into the schema

            assert django.conf.settings.TEMPLATE_VARS

            assert django.conf.settings.TEMPLATE_VARS['hobo_test_variable'] is True

            assert django.conf.settings.TEMPLATE_VARS['test_url'] == tenant.get_base_url()

            assert django.conf.settings.TEMPLATE_VARS['other_url'] == 'http://other.example.net'

            assert django.conf.settings.TEMPLATE_VARS['site_title'] == 'Test'

            assert django.conf.settings.TEMPLATE_VARS['other_variable'] == 'bar'

    # check it's no longer defined after going back to the public schema

    with pytest.raises(AttributeError):

        django.conf.settings.TEMPLATE_VARS

def test_tenant_cors_settings(tenants, settings, client):

    settings.clear_tenants_settings()

    settings.default_settings.TENANT_SETTINGS_LOADERS = ('hobo.multitenant.settings_loaders.CORSSettings', )

    # check the setting is not defined

    with pytest.raises(AttributeError):

        settings.CORS_ORIGIN_WHITELIST

    for tenant in tenants:

        with tenant_context(tenant):

            # check it's defined when moving into the schema

            assert django.conf.settings.CORS_ORIGIN_WHITELIST

            assert tenant.get_base_url() in django.conf.settings.CORS_ORIGIN_WHITELIST

            assert 'http://other.example.net' in django.conf.settings.CORS_ORIGIN_WHITELIST

    with pytest.raises(AttributeError):

        django.conf.settings.CORS_ORIGIN_WHITELIST

def test_multithreading(tenants, settings, client):

    settings.clear_tenants_settings()

    settings.default_settings.TENANT_SETTINGS_LOADERS = ('hobo.multitenant.settings_loaders.TemplateVars', )

    def f(tenant=None):

        if not tenant is None:

            assert hasattr(settings, 'TEMPLATE_VARS')

            assert settings.TEMPLATE_VARS['test_url'] == tenant.get_base_url()

        else:

            assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')

    assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')

    t1 = threading.Thread(target=f)

    t1.start()

    t1.join()

    for tenant in tenants:

        with tenant_context(tenant):

            assert hasattr(settings, 'TEMPLATE_VARS')

            t2 = threading.Thread(target=f, args=(tenant,))

            t2.start()

            t2.join()

    assert not hasattr(django.conf.settings, 'TEMPLATE_VARS')

    t3 = threading.Thread(target=f)

    t3.start()

    t3.join()

def test_cache(tenants, client):

    # Clear caches

    caches._caches.caches = {}

    cache.set('coin', 1)

    for tenant in tenants:

        with tenant_context(tenant):

            assert cache.get('coin') is None

            cache.set('coin', tenant.domain_url)

    assert cache.get('coin') == 1

    for tenant in tenants:

        with tenant_context(tenant):

            def f():

                assert cache.get('coin') == tenant.domain_url

            t1 = threading.Thread(target=f)

            t1.start()

            t1.join()

    def g():

        assert cache.get('coin') == 1

    t2 = threading.Thread(target=g)

    t2.start()

    t2.join()