summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorBenjamin Dauvergne <bdauvergne@entrouvert.com>2015-09-29 17:49:14 (GMT)
committerBenjamin Dauvergne <bdauvergne@entrouvert.com>2015-10-05 17:19:33 (GMT)
commit0b0ef397b8056c9a2034e64fb08a66c1351dd66b (patch)
tree94639789a8142ddb4510019253dd7e075247bf07
parentb9e8f4ee3f2a6e8969fcd04a93da16078e3d972e (diff)
downloadhobo-0b0ef397b8056c9a2034e64fb08a66c1351dd66b.zip
hobo-0b0ef397b8056c9a2034e64fb08a66c1351dd66b.tar.gz
hobo-0b0ef397b8056c9a2034e64fb08a66c1351dd66b.tar.bz2
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.
-rw-r--r--README17
-rw-r--r--hobo/agent/test_urls.py9
-rw-r--r--tests/test_multitenant.py191
-rw-r--r--tests_multitenant/conftest.py54
-rw-r--r--tests_multitenant/settings.py23
-rw-r--r--tests_multitenant/test_hobo_notify.py101
-rw-r--r--tests_multitenant/test_objects.py16
-rw-r--r--tests_multitenant/test_settings.py139
8 files changed, 358 insertions, 192 deletions
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 01a4d40..0000000
--- a/tests/test_multitenant.py
+++ /dev/null
@@ -1,191 +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()
- def h():
- pass
-
- threading.Timer(1, h).start()
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()