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.
This commit is contained in:
parent
b9e8f4ee3f
commit
0b0ef397b8
17
README
17
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 .
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
)
|
||||
|
|
@ -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()
|
||||
|
|
@ -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
|
||||
|
|
@ -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',
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
@ -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()
|
||||
Loading…
Reference in New Issue