2 |
2 |
import mock
|
3 |
3 |
import os
|
4 |
4 |
|
|
5 |
from django.core.management import execute_from_command_line
|
|
6 |
|
|
7 |
from hobo.multitenant.middleware import TenantMiddleware
|
|
8 |
|
5 |
9 |
pytestmark = pytest.mark.django_db
|
6 |
10 |
|
|
11 |
|
7 |
12 |
class RecordTenant(object):
|
8 |
13 |
def __init__(self):
|
9 |
14 |
self.tenants = []
|
... | ... | |
12 |
17 |
from django.db import connection
|
13 |
18 |
self.tenants.append(connection.tenant)
|
14 |
19 |
|
|
20 |
|
15 |
21 |
@mock.patch('django.contrib.sessions.management.commands.clearsessions.Command.handle')
|
16 |
22 |
def test_all_tenants(handle, tenants):
|
17 |
|
from django.core.management import execute_from_command_line
|
18 |
23 |
handle.side_effect = RecordTenant()
|
19 |
24 |
execute_from_command_line(['manage.py', 'tenant_command', 'clearsessions', '--all-tenants'])
|
20 |
25 |
assert handle.call_count == 2
|
... | ... | |
22 |
27 |
assert set(tenant.domain_url for tenant in handle.side_effect.tenants) == \
|
23 |
28 |
set(['tenant1.example.net', 'tenant2.example.net'])
|
24 |
29 |
|
|
30 |
|
25 |
31 |
@mock.patch('django.contrib.sessions.management.commands.clearsessions.Command.handle')
|
26 |
32 |
def test_one_tenant(handle, tenants, tenant_in_call=None):
|
27 |
|
from django.core.management import execute_from_command_line
|
28 |
33 |
handle.side_effect = RecordTenant()
|
29 |
34 |
execute_from_command_line(['manage.py', 'tenant_command', 'clearsessions', '-d', 'tenant2.example.net'])
|
30 |
35 |
assert handle.call_count == 1
|
31 |
36 |
assert len(handle.side_effect.tenants) == 1
|
32 |
37 |
assert handle.side_effect.tenants[0].domain_url == 'tenant2.example.net'
|
33 |
38 |
|
|
39 |
|
34 |
40 |
def test_delete_tenant(tenants):
|
35 |
|
from django.core.management import execute_from_command_line
|
36 |
|
from hobo.multitenant.middleware import TenantMiddleware
|
37 |
41 |
base = os.path.dirname(tenants[0].get_directory())
|
38 |
42 |
if any('removed' in d for d in os.listdir(base)):
|
39 |
43 |
assert False
|
|
44 |
|
40 |
45 |
def get_schemas():
|
41 |
46 |
from django.db import connection
|
42 |
47 |
cursor = connection.cursor()
|
... | ... | |
47 |
52 |
all_tenants = list(TenantMiddleware.get_tenants())
|
48 |
53 |
assert len(all_tenants) == 2
|
49 |
54 |
execute_from_command_line(['manage.py', 'delete_tenant', 'tenant2.example.net'])
|
50 |
|
if not any('removed' in d for d in os.listdir(base)):
|
|
55 |
if not any('removed' in d for d in os.listdir(base)):
|
51 |
56 |
assert False
|
52 |
57 |
if not any('removed' in x for x in get_schemas()):
|
53 |
|
assert False
|
|
58 |
assert False
|
54 |
59 |
all_tenants = list(TenantMiddleware.get_tenants())
|
55 |
60 |
assert len(all_tenants) == 1
|
56 |
|
-
|