1
1
import warnings
2
2
from threading import Lock , Thread
3
+
3
4
from django .http import HttpRequest , HttpResponse
4
5
from django .test import TestCase
5
6
from django .test .utils import override_settings
6
7
8
+
7
9
try :
8
10
from unittest import mock
9
11
except ImportError :
10
12
import mock
11
- from nose .tools import eq_
12
13
13
14
# For deprecation tests
14
15
import multidb
15
16
import multidb .pinning
16
-
17
- from multidb import (DEFAULT_DB_ALIAS , ReplicaRouter , PinningReplicaRouter ,
18
- get_replica )
19
- from multidb .middleware import (pinning_cookie , pinning_cookie_httponly ,
20
- pinning_cookie_samesite , pinning_cookie_secure ,
21
- pinning_seconds , PinningRouterMiddleware )
22
- from multidb .pinning import (this_thread_is_pinned , pin_this_thread ,
23
- unpin_this_thread , use_primary_db , db_write )
17
+ from multidb import DEFAULT_DB_ALIAS , PinningReplicaRouter , ReplicaRouter , get_replica
18
+ from multidb .middleware import (
19
+ PinningRouterMiddleware ,
20
+ pinning_cookie ,
21
+ pinning_cookie_httponly ,
22
+ pinning_cookie_samesite ,
23
+ pinning_cookie_secure ,
24
+ pinning_seconds ,
25
+ )
26
+ from multidb .pinning import (
27
+ db_write ,
28
+ pin_this_thread ,
29
+ this_thread_is_pinned ,
30
+ unpin_this_thread ,
31
+ use_primary_db ,
32
+ )
24
33
25
34
26
35
class UnpinningTestCase (TestCase ):
@@ -31,13 +40,12 @@ def tearDown(self):
31
40
32
41
33
42
class ReplicaRouterTests (TestCase ):
34
-
35
43
def test_db_for_read (self ):
36
- eq_ (ReplicaRouter ().db_for_read (None ), get_replica ())
44
+ self . assertEqual (ReplicaRouter ().db_for_read (None ), get_replica ())
37
45
# TODO: Test the round-robin functionality.
38
46
39
47
def test_db_for_write (self ):
40
- eq_ (ReplicaRouter ().db_for_write (None ), DEFAULT_DB_ALIAS )
48
+ self . assertEqual (ReplicaRouter ().db_for_write (None ), DEFAULT_DB_ALIAS )
41
49
42
50
def test_allow_syncdb (self ):
43
51
router = ReplicaRouter ()
@@ -46,32 +54,32 @@ def test_allow_syncdb(self):
46
54
47
55
def test_allow_migrate (self ):
48
56
router = ReplicaRouter ()
49
- assert router .allow_migrate (DEFAULT_DB_ALIAS , ' dummy' )
50
- assert not router .allow_migrate (get_replica (), ' dummy' )
57
+ assert router .allow_migrate (DEFAULT_DB_ALIAS , " dummy" )
58
+ assert not router .allow_migrate (get_replica (), " dummy" )
51
59
52
60
53
61
class SettingsTests (TestCase ):
54
62
"""Tests for default settings."""
55
63
56
64
def test_defaults (self ):
57
65
"""Check that the cookie name has the right default."""
58
- eq_ (pinning_cookie (), ' multidb_pin_writes' )
59
- eq_ (pinning_seconds (), 15 )
60
- eq_ (pinning_cookie_secure (), False )
61
- eq_ (pinning_cookie_httponly (), False )
62
- eq_ (pinning_cookie_samesite (), ' Lax' )
66
+ self . assertEqual (pinning_cookie (), " multidb_pin_writes" )
67
+ self . assertEqual (pinning_seconds (), 15 )
68
+ self . assertEqual (pinning_cookie_secure (), False )
69
+ self . assertEqual (pinning_cookie_httponly (), False )
70
+ self . assertEqual (pinning_cookie_samesite (), " Lax" )
63
71
64
72
@override_settings (MULTIDB_PINNING_COOKIE = "override_pin_writes" )
65
73
@override_settings (MULTIDB_PINNING_SECONDS = 60 )
66
74
@override_settings (MULTIDB_PINNING_COOKIE_SECURE = True )
67
75
@override_settings (MULTIDB_PINNING_COOKIE_HTTPONLY = True )
68
76
@override_settings (MULTIDB_PINNING_COOKIE_SAMESITE = "Strict" )
69
77
def test_overrides (self ):
70
- eq_ (pinning_cookie (), "override_pin_writes" )
71
- eq_ (pinning_seconds (), 60 )
72
- eq_ (pinning_cookie_secure (), True )
73
- eq_ (pinning_cookie_httponly (), True )
74
- eq_ (pinning_cookie_samesite (), "Strict" )
78
+ self . assertEqual (pinning_cookie (), "override_pin_writes" )
79
+ self . assertEqual (pinning_seconds (), 60 )
80
+ self . assertEqual (pinning_cookie_secure (), True )
81
+ self . assertEqual (pinning_cookie_httponly (), True )
82
+ self . assertEqual (pinning_cookie_samesite (), "Strict" )
75
83
76
84
77
85
class PinningTests (UnpinningTestCase ):
@@ -80,40 +88,40 @@ class PinningTests(UnpinningTestCase):
80
88
81
89
def test_pinning_encapsulation (self ):
82
90
"""Check the pinning getters and setters."""
83
- assert not this_thread_is_pinned (), \
84
- "Thread started out pinned or this_thread_is_pinned() is broken."
91
+ assert (
92
+ not this_thread_is_pinned ()
93
+ ), "Thread started out pinned or this_thread_is_pinned() is broken."
85
94
86
95
pin_this_thread ()
87
- assert this_thread_is_pinned (), \
88
- "pin_this_thread() didn't pin the thread."
96
+ assert this_thread_is_pinned (), "pin_this_thread() didn't pin the thread."
89
97
90
98
unpin_this_thread ()
91
- assert not this_thread_is_pinned (), \
92
- "Thread remained pinned after unpin_this_thread()."
99
+ assert (
100
+ not this_thread_is_pinned ()
101
+ ), "Thread remained pinned after unpin_this_thread()."
93
102
94
103
def test_pinned_reads (self ):
95
104
"""Test PinningReplicaRouter.db_for_read() when pinned and when
96
105
not."""
97
106
router = PinningReplicaRouter ()
98
107
99
- eq_ (router .db_for_read (None ), get_replica ())
108
+ self . assertEqual (router .db_for_read (None ), get_replica ())
100
109
101
110
pin_this_thread ()
102
- eq_ (router .db_for_read (None ), DEFAULT_DB_ALIAS )
111
+ self . assertEqual (router .db_for_read (None ), DEFAULT_DB_ALIAS )
103
112
104
113
def test_db_write_decorator (self ):
105
-
106
114
def read_view (req ):
107
- eq_ (router .db_for_read (None ), get_replica ())
115
+ self . assertEqual (router .db_for_read (None ), get_replica ())
108
116
return HttpResponse ()
109
117
110
118
@db_write
111
119
def write_view (req ):
112
- eq_ (router .db_for_read (None ), DEFAULT_DB_ALIAS )
120
+ self . assertEqual (router .db_for_read (None ), DEFAULT_DB_ALIAS )
113
121
return HttpResponse ()
114
122
115
123
router = PinningReplicaRouter ()
116
- eq_ (router .db_for_read (None ), get_replica ())
124
+ self . assertEqual (router .db_for_read (None ), get_replica ())
117
125
write_view (HttpRequest ())
118
126
read_view (HttpRequest ())
119
127
@@ -134,43 +142,46 @@ def setUp(self):
134
142
135
143
def test_pin_on_cookie (self ):
136
144
"""Thread should pin when the cookie is set."""
137
- self .request .COOKIES [pinning_cookie ()] = 'y'
145
+ self .request .COOKIES [pinning_cookie ()] = "y"
138
146
self .middleware .process_request (self .request )
139
147
assert this_thread_is_pinned ()
140
148
141
149
def test_unpin_on_no_cookie (self ):
142
150
"""Thread should unpin when cookie is absent and method is GET."""
143
151
pin_this_thread ()
144
- self .request .method = ' GET'
152
+ self .request .method = " GET"
145
153
self .middleware .process_request (self .request )
146
154
assert not this_thread_is_pinned ()
147
155
148
156
def test_pin_on_post (self ):
149
157
"""Thread should pin when method is POST."""
150
- self .request .method = ' POST'
158
+ self .request .method = " POST"
151
159
self .middleware .process_request (self .request )
152
160
assert this_thread_is_pinned ()
153
161
154
162
def test_process_response (self ):
155
163
"""Make sure the cookie gets set on POSTs but not GETs."""
156
164
157
- self .request .method = 'GET'
158
- response = self .middleware .process_response (
159
- self .request , HttpResponse ())
165
+ self .request .method = "GET"
166
+ response = self .middleware .process_response (self .request , HttpResponse ())
160
167
assert pinning_cookie () not in response .cookies
161
168
162
- self .request .method = 'POST'
163
- response = self .middleware .process_response (
164
- self .request , HttpResponse ())
169
+ self .request .method = "POST"
170
+ response = self .middleware .process_response (self .request , HttpResponse ())
165
171
assert pinning_cookie () in response .cookies
166
- eq_ (response .cookies [pinning_cookie ()]['max-age' ],
167
- pinning_seconds ())
168
- eq_ (response .cookies [pinning_cookie ()]['samesite' ],
169
- pinning_cookie_samesite ())
170
- eq_ (response .cookies [pinning_cookie ()]['httponly' ],
171
- pinning_cookie_httponly () or '' )
172
- eq_ (response .cookies [pinning_cookie ()]['secure' ],
173
- pinning_cookie_secure () or '' )
172
+ self .assertEqual (
173
+ response .cookies [pinning_cookie ()]["max-age" ], pinning_seconds ()
174
+ )
175
+ self .assertEqual (
176
+ response .cookies [pinning_cookie ()]["samesite" ], pinning_cookie_samesite ()
177
+ )
178
+ self .assertEqual (
179
+ response .cookies [pinning_cookie ()]["httponly" ],
180
+ pinning_cookie_httponly () or "" ,
181
+ )
182
+ self .assertEqual (
183
+ response .cookies [pinning_cookie ()]["secure" ], pinning_cookie_secure () or ""
184
+ )
174
185
175
186
def test_attribute (self ):
176
187
"""The cookie should get set if the _db_write attribute is True."""
@@ -182,16 +193,18 @@ def test_attribute(self):
182
193
def test_db_write_decorator (self ):
183
194
"""The @db_write decorator should make any view set the cookie."""
184
195
req = self .request
185
- req .method = ' GET'
196
+ req .method = " GET"
186
197
187
198
def view (req ):
188
199
return HttpResponse ()
200
+
189
201
response = self .middleware .process_response (req , view (req ))
190
202
assert pinning_cookie () not in response .cookies
191
203
192
204
@db_write
193
205
def write_view (req ):
194
206
return HttpResponse ()
207
+
195
208
response = self .middleware .process_response (req , write_view (req ))
196
209
assert pinning_cookie () in response .cookies
197
210
@@ -201,6 +214,7 @@ def test_decorator(self):
201
214
@use_primary_db
202
215
def check ():
203
216
assert this_thread_is_pinned ()
217
+
204
218
unpin_this_thread ()
205
219
assert not this_thread_is_pinned ()
206
220
check ()
@@ -210,6 +224,7 @@ def test_decorator_resets(self):
210
224
@use_primary_db
211
225
def check ():
212
226
assert this_thread_is_pinned ()
227
+
213
228
pin_this_thread ()
214
229
assert this_thread_is_pinned ()
215
230
check ()
@@ -289,34 +304,33 @@ def thread2_worker():
289
304
class DeprecationTestCase (TestCase ):
290
305
def test_masterslaverouter (self ):
291
306
with warnings .catch_warnings (record = True ) as w :
292
- warnings .simplefilter (' always' )
307
+ warnings .simplefilter (" always" )
293
308
router = multidb .MasterSlaveRouter ()
294
309
assert isinstance (router , ReplicaRouter )
295
310
assert len (w ) == 1
296
311
assert issubclass (w [- 1 ].category , DeprecationWarning )
297
312
298
313
def test_pinningmasterslaverouter (self ):
299
314
with warnings .catch_warnings (record = True ) as w :
300
- warnings .simplefilter (' always' )
315
+ warnings .simplefilter (" always" )
301
316
router = multidb .PinningMasterSlaveRouter ()
302
317
assert isinstance (router , PinningReplicaRouter )
303
318
assert len (w ) == 1
304
319
assert issubclass (w [- 1 ].category , DeprecationWarning )
305
320
306
- @mock .patch .object (multidb , ' get_replica' )
321
+ @mock .patch .object (multidb , " get_replica" )
307
322
def test_get_slave (self , mock_get_replica ):
308
323
with warnings .catch_warnings (record = True ) as w :
309
- warnings .simplefilter (' always' )
324
+ warnings .simplefilter (" always" )
310
325
multidb .get_slave ()
311
326
assert mock_get_replica .called
312
327
assert len (w ) == 1
313
328
assert issubclass (w [- 1 ].category , DeprecationWarning )
314
329
315
330
def test_use_master (self ):
316
- assert isinstance (multidb .pinning .use_master ,
317
- use_primary_db .__class__ )
331
+ assert isinstance (multidb .pinning .use_master , use_primary_db .__class__ )
318
332
with warnings .catch_warnings (record = True ) as w :
319
- warnings .simplefilter (' always' )
333
+ warnings .simplefilter (" always" )
320
334
with multidb .pinning .use_master :
321
335
pass
322
336
assert len (w ) == 1
0 commit comments