From ec483bc2edf9f829dbc5ec0898584f086f7ad7bb Mon Sep 17 00:00:00 2001 From: Benjamin Dauvergne Date: Tue, 29 Sep 2015 19:49:14 +0200 Subject: [PATCH 3/3] tests: add tests for the multitenant framework --- 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 | 22 ++++ tests_multitenant/test_settings.py | 139 +++++++++++++++++++++++++ 7 files changed, 263 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_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..aa97203 --- /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') + +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..71b6938 --- /dev/null +++ b/tests_multitenant/test_hobo_notify.py @@ -0,0 +1,22 @@ +import random +import pytest + +from django.apps import apps +import django.conf +import django.db + + + + +def test_hobo_notify(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