« back to all changes in this revision
Viewing changes to tests/django/middleware.py
- browse files at revision 480
- compare with another revision
- download diff
- download tarball
- view history from revision 480
- Committer: Benji York
- Date: 2016-03-01 14:19:44 UTC
- mfrom: (478.2.2 kill-django-storm)
- Revision ID: benji.york@canonical.com-20160301141944-rk34atio78h7n1h3
Merge lp:~adam-collard/storm/kill-django-storm [r=benji]
Remove Django support in Storm.
The Django support has long since bit-rotted in Storm, and searching
on http://searchcode.com/ shows it isn't being used. Rather than try
to keep it alive this branch does the honourable thing and shoots it
in the head.
With this branch merged, tests should pass on Xenial.
Remove Django support in Storm.
The Django support has long since bit-rotted in Storm, and searching
on http://searchcode.com/ shows it isn't being used. Rather than try
to keep it alive this branch does the honourable thing and shoots it
in the head.
With this branch merged, tests should pass on Xenial.
added
removed
tests/django/middleware.py
1
#
2
# Copyright (c) 2008 Canonical
3
#
4
# Written by James Henstridge <jamesh@canonical.com>
5
#
6
# This file is part of Storm Object Relational Mapper.
7
#
8
# Storm is free software; you can redistribute it and/or modify
9
# it under the terms of the GNU Lesser General Public License as
10
# published by the Free Software Foundation; either version 2.1 of
11
# the License, or (at your option) any later version.
12
#
13
# Storm is distributed in the hope that it will be useful,
14
# but WITHOUT ANY WARRANTY; without even the implied warranty of
15
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16
# GNU Lesser General Public License for more details.
17
#
18
# You should have received a copy of the GNU Lesser General Public License
19
# along with this program. If not, see <http://www.gnu.org/licenses/>.
20
#
21
22
from storm.exceptions import DisconnectionError
23
24
try:
25
import django
26
import transaction
27
except ImportError:
28
have_django_and_transaction = False
29
else:
30
have_django_and_transaction = True
31
from django import conf
32
from django.http import HttpRequest, HttpResponse
33
from storm.django.middleware import ZopeTransactionMiddleware
34
35
from tests.helper import TestHelper
36
37
38
class TransactionMiddlewareTests(TestHelper):
39
40
def is_supported(self):
41
return have_django_and_transaction
42
43
def setUp(self):
44
super(TransactionMiddlewareTests, self).setUp()
45
conf.settings.configure(STORM_COMMIT_SAFE_METHODS=False)
46
from django.db import transaction as django_transaction
47
self.django_transaction = django_transaction
48
49
def tearDown(self):
50
if django.VERSION >= (1, 1):
51
if django.VERSION >= (1, 6):
52
from django.utils.functional import empty
53
else:
54
empty = None
55
conf.settings._wrapped = empty
56
else:
57
conf.settings._target = None
58
super(TransactionMiddlewareTests, self).tearDown()
59
60
def test_process_request_begins_transaction(self):
61
begin = self.mocker.replace("transaction.begin")
62
enter_transaction_management = self.mocker.replace(
63
"django.db.transaction.enter_transaction_management")
64
managed = self.mocker.replace(
65
"django.db.transaction.managed")
66
enter_transaction_management()
67
managed(True)
68
begin()
69
self.mocker.replay()
70
71
zope_middleware = ZopeTransactionMiddleware()
72
request = HttpRequest()
73
request.method = "GET"
74
zope_middleware.process_request(request)
75
76
def test_process_exception_aborts_transaction(self):
77
abort = self.mocker.replace("transaction.abort")
78
leave_transaction_management = self.mocker.replace(
79
"django.db.transaction.leave_transaction_management")
80
set_clean = self.mocker.replace(
81
"django.db.transaction.set_clean")
82
abort()
83
set_clean()
84
leave_transaction_management()
85
self.mocker.replay()
86
87
zope_middleware = ZopeTransactionMiddleware()
88
request = HttpRequest()
89
request.method = "GET"
90
exception = RuntimeError("some error")
91
zope_middleware.process_exception(request, exception)
92
93
def test_process_response_commits_transaction_if_managed(self):
94
commit = self.mocker.replace("transaction.commit")
95
leave_transaction_management = self.mocker.replace(
96
"django.db.transaction.leave_transaction_management")
97
is_managed = self.mocker.replace(
98
"django.db.transaction.is_managed")
99
set_clean = self.mocker.replace(
100
"django.db.transaction.set_clean")
101
# We test three request methods
102
self.expect(is_managed()).result(True).count(3)
103
self.expect(commit()).count(3)
104
self.expect(set_clean()).count(3)
105
self.expect(leave_transaction_management()).count(3)
106
self.mocker.replay()
107
108
# Commit on all methods
109
conf.settings.STORM_COMMIT_SAFE_METHODS = True
110
111
zope_middleware = ZopeTransactionMiddleware()
112
request = HttpRequest()
113
response = HttpResponse()
114
115
request.method = "GET"
116
zope_middleware.process_response(request, response)
117
request.method = "HEAD"
118
zope_middleware.process_response(request, response)
119
request.method = "POST"
120
zope_middleware.process_response(request, response)
121
122
def test_process_response_aborts_transaction_for_safe_methods(self):
123
abort = self.mocker.replace("transaction.abort")
124
commit = self.mocker.replace("transaction.commit")
125
leave_transaction_management = self.mocker.replace(
126
"django.db.transaction.leave_transaction_management")
127
is_managed = self.mocker.replace(
128
"django.db.transaction.is_managed")
129
set_clean = self.mocker.replace(
130
"django.db.transaction.set_clean")
131
# We test three request methods
132
self.expect(is_managed()).result(True).count(3)
133
self.expect(abort()).count(2)
134
commit()
135
self.expect(set_clean()).count(3)
136
self.expect(leave_transaction_management()).count(3)
137
self.mocker.replay()
138
139
# Don't commit on safe methods
140
conf.settings.STORM_COMMIT_SAFE_METHODS = False
141
142
zope_middleware = ZopeTransactionMiddleware()
143
request = HttpRequest()
144
response = HttpResponse()
145
146
request.method = "GET"
147
zope_middleware.process_response(request, response)
148
request.method = "HEAD"
149
zope_middleware.process_response(request, response)
150
request.method = "POST"
151
zope_middleware.process_response(request, response)
152
153
def test_process_response_aborts_transaction_not_managed(self):
154
abort = self.mocker.replace("transaction.abort")
155
commit = self.mocker.replace("transaction.commit")
156
leave_transaction_management = self.mocker.replace(
157
"django.db.transaction.leave_transaction_management")
158
is_managed = self.mocker.replace(
159
"django.db.transaction.is_managed")
160
set_clean = self.mocker.replace(
161
"django.db.transaction.set_clean")
162
163
self.expect(is_managed()).result(False).count(2)
164
# None of these methods should be called
165
self.expect(commit()).count(0)
166
self.expect(abort()).count(0)
167
self.expect(set_clean()).count(0)
168
self.expect(leave_transaction_management()).count(0)
169
self.mocker.replay()
170
171
zope_middleware = ZopeTransactionMiddleware()
172
request = HttpRequest()
173
response = HttpResponse()
174
175
request.method = "GET"
176
zope_middleware.process_response(request, response)
177
178
# Try the same with a safe method.
179
conf.settings.STORM_COMMIT_SAFE_METHODS = False
180
zope_middleware = ZopeTransactionMiddleware()
181
zope_middleware.process_response(request, response)
182
183
def test_process_response_aborts_transaction_on_failed_commit(self):
184
_transaction = transaction.get()
185
resource1 = self.mocker.mock()
186
self.expect(resource1.prepare).throw(AttributeError).count(0)
187
resource2 = self.mocker.mock()
188
self.expect(resource2.prepare).throw(AttributeError).count(0)
189
leave_transaction_management = self.mocker.replace(
190
"django.db.transaction.leave_transaction_management")
191
is_managed = self.mocker.replace(
192
"django.db.transaction.is_managed")
193
set_clean = self.mocker.replace(
194
"django.db.transaction.set_clean")
195
196
self.expect(is_managed()).result(True).count(1)
197
self.expect(resource1.tpc_begin(_transaction)).throw(DisconnectionError)
198
self.expect(resource1.abort(_transaction)).count(2)
199
# None of these methods should be called
200
self.expect(set_clean()).count(0)
201
self.expect(leave_transaction_management()).count(0)
202
self.mocker.replay()
203
204
_transaction.join(resource1)
205
zope_middleware = ZopeTransactionMiddleware()
206
request = HttpRequest()
207
response = HttpResponse()
208
209
# Processing this response should fail on "commit()" with a
210
# DisconnectionError. The error is saved by
211
# "_saveAndRaiseCommitishError()", and re-raised, so we need to catch
212
# it and abort the transaction.
213
request.method = "POST"
214
self.assertRaises(DisconnectionError, zope_middleware.process_response,
215
request, response)
216
217
# Now the transaction should have been cleared by "abort()". A resource
218
# joining the transaction manager should not fail with a
219
# "TransactionFailedError".
220
transaction.get().join(resource2)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1