1 |
|
import json
|
2 |
|
import os
|
3 |
|
import shutil
|
4 |
|
import tempfile
|
5 |
|
|
6 |
|
import django_webtest
|
7 |
1 |
import pytest
|
8 |
2 |
|
9 |
3 |
|
10 |
|
@pytest.fixture(scope='function')
|
11 |
|
def tenants(transactional_db, request, settings):
|
|
4 |
@pytest.fixture
|
|
5 |
def make_tenant(tmp_path, transactional_db, settings, request):
|
|
6 |
import json
|
|
7 |
|
12 |
8 |
from hobo.multitenant.models import Tenant
|
13 |
9 |
|
14 |
|
base = tempfile.mkdtemp('combo-tenant-base')
|
15 |
|
settings.TENANT_BASE = base
|
|
10 |
base = tmp_path / 'combo-tenant-base'
|
|
11 |
base.mkdir()
|
|
12 |
settings.TENANT_BASE = str(base)
|
16 |
13 |
|
17 |
|
@pytest.mark.django_db
|
18 |
14 |
def make_tenant(name):
|
19 |
|
tenant_dir = os.path.join(base, name)
|
20 |
|
os.mkdir(tenant_dir)
|
21 |
|
with open(os.path.join(tenant_dir, 'unsecure'), 'w') as fd:
|
|
15 |
tenant_dir = base / name
|
|
16 |
tenant_dir.mkdir()
|
|
17 |
with (tenant_dir / 'unsecure').open(mode='w') as fd:
|
22 |
18 |
fd.write('1')
|
23 |
|
with open(os.path.join(tenant_dir, 'settings.json'), 'w') as fd:
|
|
19 |
with (tenant_dir / 'settings.json').open(mode='w') as fd:
|
24 |
20 |
json.dump(
|
25 |
21 |
{
|
26 |
22 |
'HOBO_TEST_VARIABLE': name,
|
... | ... | |
29 |
25 |
},
|
30 |
26 |
fd,
|
31 |
27 |
)
|
32 |
|
with open(os.path.join(tenant_dir, 'hobo.json'), 'w') as fd:
|
|
28 |
with (tenant_dir / 'hobo.json').open(mode='w') as fd:
|
33 |
29 |
json.dump(
|
34 |
30 |
{
|
35 |
31 |
'variables': {
|
... | ... | |
94 |
90 |
)
|
95 |
91 |
t = Tenant(domain_url=name, schema_name=name.replace('-', '_').replace('.', '_'))
|
96 |
92 |
t.create_schema()
|
|
93 |
|
|
94 |
def drop_tenant():
|
|
95 |
from django.db import connection
|
|
96 |
|
|
97 |
from hobo.multitenant.middleware import TenantMiddleware, TenantNotFound
|
|
98 |
|
|
99 |
connection.set_schema_to_public()
|
|
100 |
try:
|
|
101 |
TenantMiddleware.get_tenant_by_hostname(t.domain_url)
|
|
102 |
except TenantNotFound:
|
|
103 |
pass
|
|
104 |
else:
|
|
105 |
t.delete(True)
|
|
106 |
|
|
107 |
request.addfinalizer(drop_tenant)
|
97 |
108 |
return t
|
98 |
109 |
|
99 |
|
tenants = [make_tenant('tenant1.example.net'), make_tenant('tenant2.example.net')]
|
|
110 |
return make_tenant
|
100 |
111 |
|
101 |
|
def fin():
|
102 |
|
from django.db import connection
|
103 |
112 |
|
104 |
|
from hobo.multitenant.middleware import TenantMiddleware
|
|
113 |
@pytest.fixture
|
|
114 |
def tenants(make_tenant):
|
|
115 |
return [make_tenant('tenant1.example.net'), make_tenant('tenant2.example.net')]
|
105 |
116 |
|
106 |
|
connection.set_schema_to_public()
|
107 |
|
for t in TenantMiddleware.get_tenants():
|
108 |
|
t.delete(True)
|
109 |
|
shutil.rmtree(base)
|
110 |
117 |
|
111 |
|
request.addfinalizer(fin)
|
112 |
|
return tenants
|
|
118 |
@pytest.fixture
|
|
119 |
def tenant(make_tenant):
|
|
120 |
return make_tenant('tenant.example.net')
|
113 |
121 |
|
114 |
122 |
|
115 |
123 |
@pytest.fixture
|
116 |
124 |
def app(request):
|
|
125 |
import django_webtest
|
|
126 |
|
117 |
127 |
wtm = django_webtest.WebTestMixin()
|
118 |
128 |
wtm._patch_settings()
|
119 |
129 |
yield django_webtest.DjangoTestApp()
|
120 |
|
-
|