From 90747feeb680d94c0ae9f3bc7c71125ee545d562 Mon Sep 17 00:00:00 2001 From: Benjamin Dauvergne Date: Tue, 29 Sep 2015 19:49:14 +0200 Subject: [PATCH 3/5] tests: add tests for the multitenant framework (#8425) tests/test_settings.py is moved in this new test suite. Tested are the hobo_notify script and the simple creation of user objects. --- README | 17 +++- hobo/agent/test_urls.py | 9 ++ tests/test_multitenant.py | 187 ---------------------------------- tests_multitenant/conftest.py | 54 ++++++++++ tests_multitenant/settings.py | 23 +++++ tests_multitenant/test_hobo_notify.py | 101 ++++++++++++++++++ tests_multitenant/test_objects.py | 16 +++ tests_multitenant/test_settings.py | 139 +++++++++++++++++++++++++ 8 files changed, 358 insertions(+), 188 deletions(-) create mode 100644 hobo/agent/test_urls.py delete mode 100644 tests/test_multitenant.py create mode 100644 tests_multitenant/conftest.py create mode 100644 tests_multitenant/settings.py create mode 100644 tests_multitenant/test_hobo_notify.py create mode 100644 tests_multitenant/test_objects.py create mode 100644 tests_multitenant/test_settings.py diff --git a/README b/README index 4358f13..5e2dca4 100644 --- a/README +++ b/README @@ -161,4 +161,19 @@ import-wcs-roles. It computes the web-service credentials from the hobo.json and use the email of the oldest superuser. Cron job can be created for calling this command when regular synchronization of roles with your w.c.s. instances is needed. The sole option named "--delete" indicate if you want to delete -stale roles, default is to not delete them. +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 . diff --git a/hobo/agent/test_urls.py b/hobo/agent/test_urls.py new file mode 100644 index 0000000..d601d84 --- /dev/null +++ b/hobo/agent/test_urls.py @@ -0,0 +1,9 @@ +from django.conf.urls import patterns, url +from django.http import HttpResponse + +def helloworld(request): + return HttpResponse('Hello world') + +urlpatterns = patterns('', + url(r'^$', helloworld), +) diff --git a/tests/test_multitenant.py b/tests/test_multitenant.py deleted file mode 100644 index f388c8a..0000000 --- a/tests/test_multitenant.py +++ /dev/null @@ -1,187 +0,0 @@ -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 - -@pytest.fixture -def multitenant_settings(settings): - settings.MIDDLEWARE_CLASSES = ( - 'hobo.multitenant.middleware.TenantMiddleware', - ) + settings.MIDDLEWARE_CLASSES - - settings.TENANT_APPS = settings.INSTALLED_APPS - settings.TENANT_MODEL = 'multitenant.Tenant' - settings.DATABASE_ROUTERS = ('tenant_schemas.routers.TenantSyncRouter',) - settings.DATABASES = { - 'default': { - 'ENGINE': 'tenant_schemas.postgresql_backend', - 'NAME': 'test%s' % str(random.random())[2:] - } - } - settings.CACHES = { - 'default': { - 'BACKEND': 'hobo.multitenant.cache.TenantCache', - # add a real Django cache backend, with its parameters if needed - 'REAL_BACKEND': 'django.core.cache.backends.locmem.LocMemCache', - 'LOCATION': '127.0.0.1:11211', - } - } - - django.db.connections = django.db.ConnectionHandler(settings.DATABASES) - apps.set_installed_apps(('hobo.multitenant', 'hobo.agent.common',) + settings.INSTALLED_APPS) - return settings - -def test_tenant_middleware(multitenant_settings, client): - res = client.get('/', SERVER_NAME='invalid.example.net') - assert res.status_code == 404 - res = client.get('/', SERVER_NAME='test.example.net') - assert res.status_code != 404 - assert res.wsgi_request.tenant.schema_name == 'test_example_net' - -def test_tenant_json_settings(multitenant_settings, client): - from hobo.multitenant.models import Tenant - - django.conf.settings.clear_tenants_settings() - - multitenant_settings.default_settings.TENANT_SETTINGS_LOADERS = ('hobo.multitenant.settings_loaders.SettingsJSON', ) - - # check the setting is not defined - with pytest.raises(AttributeError): - django.conf.settings.HOBO_TEST_VARIABLE - - django.db.connection.set_tenant(Tenant(domain_url='test-settings.example.net', - schema_name='test_settings_example_net')) - - # check it's defined when moving into the schema - assert django.conf.settings.HOBO_TEST_VARIABLE is True - - django.db.connection.set_schema_to_public() - - # check it's no longer defined after going back to the public schema - with pytest.raises(AttributeError): - django.conf.settings.HOBO_TEST_VARIABLE - -def test_tenant_hobo_settings(multitenant_settings, client): - from hobo.multitenant.models import Tenant - - django.conf.settings.clear_tenants_settings() - - multitenant_settings.TENANT_SETTINGS_LOADERS = ('hobo.multitenant.settings_loaders.TemplateVars', ) - - # check the setting is not defined - with pytest.raises(AttributeError): - django.conf.settings.TEMPLATE_VARS - - django.db.connection.set_tenant(Tenant(domain_url='test-templatevars.example.net', - schema_name='test_templatevars_example_net')) - - # 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'] == 'http://test-templatevars.example.net' - 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' - - django.db.connection.set_schema_to_public() - - # 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(multitenant_settings, client): - from hobo.multitenant.models import Tenant - - django.conf.settings.clear_tenants_settings() - - multitenant_settings.TENANT_SETTINGS_LOADERS = ('hobo.multitenant.settings_loaders.CORSSettings', ) - # check the setting is not defined - with pytest.raises(AttributeError): - django.conf.settings.CORS_ORIGIN_WHITELIST - - django.db.connection.set_tenant(Tenant(domain_url='test-templatevars.example.net', - schema_name='test_templatevars_example_net')) - - # check it's defined when moving into the schema - assert django.conf.settings.CORS_ORIGIN_WHITELIST - assert 'http://test-templatevars.example.net' in django.conf.settings.CORS_ORIGIN_WHITELIST - assert 'http://other.example.net' in django.conf.settings.CORS_ORIGIN_WHITELIST - - # check it's no longer defined after process_response() - django.db.connection.set_schema_to_public() - - with pytest.raises(AttributeError): - django.conf.settings.CORS_ORIGIN_WHITELIST - -def test_multithreading(multitenant_settings, client): - from hobo.multitenant.models import Tenant - - django.conf.settings.clear_tenants_settings() - multitenant_settings.TENANT_SETTINGS_LOADERS = ('hobo.multitenant.settings_loaders.TemplateVars', ) - - def f(template_vars=False): - if template_vars: - assert hasattr(django.conf.settings, 'TEMPLATE_VARS') - 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() - - django.db.connection.set_tenant(Tenant(domain_url='test-templatevars.example.net', - schema_name='test_templatevars_example_net')) - - assert hasattr(django.conf.settings, 'TEMPLATE_VARS') - t2 = threading.Thread(target=f, args=(True,)) - t2.start() - t2.join() - - django.db.connection.set_schema_to_public() - - assert not hasattr(django.conf.settings, 'TEMPLATE_VARS') - t3 = threading.Thread(target=f) - t3.start() - t3.join() - -def test_cache(multitenant_settings, client): - from hobo.multitenant.models import Tenant - - # Clear caches - caches._caches.caches = {} - - assert not hasattr(django.db.connection.get_tenant(), 'domain_url') - cache.set('coin', 1) - - django.db.connection.set_tenant(Tenant(domain_url='test-templatevars.example.net', - schema_name='test_templatevars_example_net')) - assert cache.get('coin') is None - cache.set('coin', 2) - - django.db.connection.set_schema_to_public() - - assert cache.get('coin') == 1 - - django.db.connection.set_tenant(Tenant(domain_url='test-templatevars.example.net', - schema_name='test_templatevars_example_net')) - - def f(): - assert cache.get('coin') == 2 - t1 = threading.Thread(target=f) - t1.start() - t1.join() - - django.db.connection.set_schema_to_public() - - def g(): - assert cache.get('coin') == 1 - t2 = threading.Thread(target=f) - t2.start() - t2.join() diff --git a/tests_multitenant/conftest.py b/tests_multitenant/conftest.py new file mode 100644 index 0000000..77dff80 --- /dev/null +++ b/tests_multitenant/conftest.py @@ -0,0 +1,54 @@ +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 diff --git a/tests_multitenant/settings.py b/tests_multitenant/settings.py new file mode 100644 index 0000000..a1b9106 --- /dev/null +++ b/tests_multitenant/settings.py @@ -0,0 +1,23 @@ +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', + } +} + diff --git a/tests_multitenant/test_hobo_notify.py b/tests_multitenant/test_hobo_notify.py new file mode 100644 index 0000000..be57621 --- /dev/null +++ b/tests_multitenant/test_hobo_notify.py @@ -0,0 +1,101 @@ +# -*- 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 diff --git a/tests_multitenant/test_objects.py b/tests_multitenant/test_objects.py new file mode 100644 index 0000000..561ff06 --- /dev/null +++ b/tests_multitenant/test_objects.py @@ -0,0 +1,16 @@ +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 diff --git a/tests_multitenant/test_settings.py b/tests_multitenant/test_settings.py new file mode 100644 index 0000000..762ac01 --- /dev/null +++ b/tests_multitenant/test_settings.py @@ -0,0 +1,139 @@ +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() -- 2.1.4