1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
|
# tests that db settings can change between requests
import copy
import os
from django.conf import settings
from django.core.handlers.wsgi import WSGIHandler
from django.db import models, model_connection_name, _default, connection
from django.http import HttpResponse
# state holder
S = {}
# models
class MX(models.Model):
val = models.CharField(maxlength=10)
class Meta:
app_label = 'ri'
class MY(models.Model):
val = models.CharField(maxlength=10)
class Meta:
app_label = 'ri'
# tests
def test_one(path, request):
"""Start out with settings as originally configured"""
assert model_connection_name(MX) == 'django_test_db_a'
assert MX._default_manager.db.connection.settings.DATABASE_NAME == \
settings.OTHER_DATABASES['django_test_db_a']['DATABASE_NAME']
assert model_connection_name(MY) == 'django_test_db_b'
assert MY._default_manager.db.connection.settings.DATABASE_NAME == \
settings.OTHER_DATABASES['django_test_db_b']['DATABASE_NAME'], \
"%s != %s" % \
(MY._default_manager.db.connection.settings.DATABASE_NAME,
settings.OTHER_DATABASES['django_test_db_b']['DATABASE_NAME'])
def test_two(path, request):
"""Between the first and second requests, settings change to assign
model MY to a different connection
"""
assert model_connection_name(MX) == 'django_test_db_a'
assert MX._default_manager.db.connection.settings.DATABASE_NAME == \
settings.OTHER_DATABASES['django_test_db_a']['DATABASE_NAME']
assert model_connection_name(MY) == _default
assert MY._default_manager.db.connection.settings.DATABASE_NAME == \
settings.DATABASE_NAME, "%s != %s" % \
(MY._default_manager.db.connection.settings.DATABASE_NAME,
settings.DATABASE_NAME)
# helpers
class MockHandler(WSGIHandler):
def __init__(self, test):
self.test = test
super(MockHandler, self).__init__()
def get_response(self, path, request):
# debug("mock handler answering %s, %s", path, request)
return HttpResponse(self.test(path, request))
def pr(*arg):
if S['verbosity'] >= 1:
msg, arg = arg[0], arg[1:]
print msg % arg
def debug(*arg):
if S['verbosity'] >= 2:
msg, arg = arg[0], arg[1:]
print msg % arg
def setup():
debug("setup")
S['ODB'] = copy.deepcopy(settings.OTHER_DATABASES)
for sk in [ k for k in dir(settings) if k.startswith('DATABASE') ]:
S[sk] = getattr(settings, sk)
settings.OTHER_DATABASES['django_test_db_a']['MODELS'] = ['ri.MX']
settings.OTHER_DATABASES['django_test_db_b']['MODELS'] = ['ri.MY']
def teardown():
debug("teardown")
settings.OTHER_DATABASES = S['ODB']
for sk in [ k for k in S.keys() if k.startswith('DATABASE') ]:
setattr(settings, sk, S[sk])
def start_response(code, headers):
debug("start response: %s %s", code, headers)
pass
def main():
debug("running tests")
env = os.environ.copy()
env['PATH_INFO'] = '/'
env['REQUEST_METHOD'] = 'GET'
MockHandler(test_one)(env, start_response)
settings.OTHER_DATABASES['django_test_db_b']['MODELS'] = []
MockHandler(test_two)(env, start_response)
def run_tests(verbosity=0):
S['verbosity'] = verbosity
setup()
try:
main()
finally:
teardown()
if __name__ == '__main__':
run_tests(2)
|
