259 lines
10 KiB
Python
259 lines
10 KiB
Python
import os
|
|
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 django.test import override_settings
|
|
|
|
from tenant_schemas.utils import tenant_context
|
|
|
|
import utilities
|
|
from hobo.theme.utils import set_theme, get_themes
|
|
|
|
def test_tenant_middleware(tenants, client, settings):
|
|
settings.ALLOWED_HOSTS.append('invalid.example.net')
|
|
settings.SETTINGS_MODULE = 'fake.settings'
|
|
res = client.get('/', SERVER_NAME='invalid.example.net')
|
|
assert res.status_code == 404
|
|
for tenant in tenants:
|
|
settings.ALLOWED_HOSTS.append(tenant.domain_url)
|
|
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()
|
|
|
|
with utilities.patch_default_settings(settings,
|
|
TENANT_SETTINGS_LOADERS=('hobo.multitenant.settings_loaders.SettingsJSON',)):
|
|
|
|
# check the setting is not defined
|
|
with pytest.raises(AttributeError):
|
|
settings.HOBO_TEST_VARIABLE
|
|
assert django.conf.settings.UPDATE_ME == {'x': 1}
|
|
assert django.conf.settings.EXTEND_ME == [1]
|
|
|
|
|
|
# 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
|
|
assert django.conf.settings.UPDATE_ME == {'x': 1, 'y': 2}
|
|
assert django.conf.settings.EXTEND_ME == [1, 2]
|
|
|
|
# check it's no longer defined after going back to the public schema
|
|
with pytest.raises(AttributeError):
|
|
settings.HOBO_TEST_VARIABLE
|
|
assert django.conf.settings.UPDATE_ME == {'x': 1}
|
|
assert django.conf.settings.EXTEND_ME == [1]
|
|
|
|
|
|
def test_tenant_template_vars(tenants, settings, client):
|
|
from hobo.multitenant.models import Tenant
|
|
|
|
django.conf.settings.clear_tenants_settings()
|
|
|
|
with utilities.patch_default_settings(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_settings_vars(tenants, settings, client):
|
|
from hobo.multitenant.models import Tenant
|
|
|
|
django.conf.settings.clear_tenants_settings()
|
|
|
|
with utilities.patch_default_settings(settings,
|
|
TENANT_SETTINGS_LOADERS=('hobo.multitenant.settings_loaders.SettingsVars',)):
|
|
# check the setting is not defined
|
|
with pytest.raises(AttributeError):
|
|
django.conf.settings.LOCAL1
|
|
with pytest.raises(AttributeError):
|
|
django.conf.settings.LOCAL2
|
|
with pytest.raises(AttributeError):
|
|
django.conf.settings.LOCAL3
|
|
|
|
for tenant in tenants:
|
|
with tenant_context(tenant):
|
|
# check it's defined when moving into the schema
|
|
assert django.conf.settings.GLOBAL1 is True
|
|
assert django.conf.settings.GLOBAL2 == [1, 2, 3, 2, 3, 4]
|
|
assert django.conf.settings.GLOBAL3 == {'x': 1, 'y': 2, 'z': 1}
|
|
assert django.conf.settings.OVERRIDE1 is True
|
|
assert django.conf.settings.OVERRIDE2 == [1, 2, 3, tenant.domain_url, 7, 8]
|
|
assert django.conf.settings.OVERRIDE3 == {'a': tenant.domain_url, 'b': 2, 'z': 1}
|
|
assert django.conf.settings.LOCAL1 is False
|
|
assert django.conf.settings.LOCAL2 == [tenant.domain_url, 7, 8]
|
|
assert django.conf.settings.LOCAL3 == {'a': tenant.domain_url, 'b': 2}
|
|
|
|
def test_tenant_cors_settings(tenants, settings, client):
|
|
settings.clear_tenants_settings()
|
|
|
|
with utilities.patch_default_settings(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_tenant_theme_settings(tenants, settings, client):
|
|
from hobo.multitenant.models import Tenant
|
|
|
|
django.conf.settings.clear_tenants_settings()
|
|
|
|
with utilities.patch_default_settings(settings,
|
|
TENANT_SETTINGS_LOADERS=('hobo.multitenant.settings_loaders.TemplateVars',
|
|
'hobo.multitenant.settings_loaders.ThemeSettings')):
|
|
old_value = os.environ['DJANGO_SETTINGS_MODULE']
|
|
os.environ['DJANGO_SETTINGS_MODULE'] = 'fake.settings'
|
|
with tenant_context(tenants[0]):
|
|
# check it's defined when moving into the schema
|
|
assert django.conf.settings.TEMPLATE_VARS
|
|
assert django.conf.settings.TEMPLATE_VARS['theme'] == 'publik'
|
|
assert django.conf.settings.HELLO == 'world'
|
|
os.environ['DJANGO_SETTINGS_MODULE'] = old_value
|
|
|
|
def test_multithreading(tenants, settings, client):
|
|
|
|
settings.clear_tenants_settings()
|
|
with utilities.patch_default_settings(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()
|
|
|
|
|
|
def test_shared_secret():
|
|
from hobo.multitenant.settings_loaders import KnownServices
|
|
|
|
secrets = set()
|
|
for i in range(100):
|
|
a = str(random.getrandbits(160))
|
|
b = str(random.getrandbits(160))
|
|
assert KnownServices.shared_secret(a, b) == KnownServices.shared_secret(b, a)
|
|
secrets.add(KnownServices.shared_secret(a, b))
|
|
# Verify minimum entropy
|
|
assert len(secrets) == 100
|
|
|
|
|
|
def test_known_services(tenants, settings):
|
|
from hobo.multitenant.settings_loaders import KnownServices
|
|
|
|
settings.clear_tenants_settings()
|
|
settings.SETTINGS_MODULE = 'fake.settings'
|
|
|
|
for tenant in tenants:
|
|
with tenant_context(tenant):
|
|
hobo_json = tenant.get_hobo_json()
|
|
assert hasattr(settings, 'KNOWN_SERVICES')
|
|
assert 'authentic' in settings.KNOWN_SERVICES
|
|
assert 'other' in settings.KNOWN_SERVICES['authentic']
|
|
assert (set(['url', 'backoffice-menu-url', 'title', 'orig', 'verif_orig', 'secret', 'template_name', 'variables', 'secondary'])
|
|
== set(settings.KNOWN_SERVICES['authentic']['other'].keys()))
|
|
assert (settings.KNOWN_SERVICES['authentic']['other']['url']
|
|
== hobo_json['services'][2]['base_url'])
|
|
assert (settings.KNOWN_SERVICES['authentic']['other']['variables']
|
|
== hobo_json['services'][2].get('variables'))
|
|
assert (settings.KNOWN_SERVICES['authentic']['other']['title']
|
|
== hobo_json['services'][2]['title'])
|
|
assert settings.KNOWN_SERVICES['authentic']['other']['orig'] == tenant.domain_url
|
|
assert (settings.KNOWN_SERVICES['authentic']['other']['verif_orig'] ==
|
|
'other.example.net')
|
|
key1 = hobo_json['services'][0]['secret_key']
|
|
key2 = hobo_json['services'][2]['secret_key']
|
|
assert (settings.KNOWN_SERVICES['authentic']['other']['secret'] ==
|
|
KnownServices.shared_secret(key1, key2))
|
|
|
|
|
|
def test_unique_cookies(tenants, settings):
|
|
from hobo.multitenant.settings_loaders import KnownServices
|
|
|
|
settings.clear_tenants_settings()
|
|
|
|
cookie_names = set()
|
|
for tenant in tenants:
|
|
with tenant_context(tenant):
|
|
cookie_names.add(settings.CSRF_COOKIE_NAME)
|
|
cookie_names.add(settings.SESSION_COOKIE_NAME)
|
|
|
|
assert len(cookie_names) == len(tenants) * 2
|