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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
|
import pyramid.request
import pyramid.response
import pyramid.testing
import pytest
import webtest
def test_raising_RetryableException_is_caught(config):
from pyramid_retry import RetryableException
calls = []
def final_view(request):
calls.append('ok')
return 'ok'
def bad_view(request):
calls.append('fail')
raise RetryableException
config.add_view(bad_view, last_retry_attempt=False)
config.add_view(final_view, last_retry_attempt=True, renderer='string')
app = config.make_wsgi_app()
app = webtest.TestApp(app)
response = app.get('/')
assert response.body == b'ok'
assert calls == ['fail', 'fail', 'ok']
def test_raising_IRetryableError_instance_is_caught(config):
from pyramid_retry import mark_error_retryable
calls = []
def final_view(request):
calls.append('ok')
return 'ok'
def bad_view(request):
calls.append('fail')
ex = Exception()
mark_error_retryable(ex)
raise ex
config.add_view(bad_view, last_retry_attempt=False)
config.add_view(final_view, last_retry_attempt=True, renderer='string')
app = config.make_wsgi_app()
app = webtest.TestApp(app)
response = app.get('/')
assert response.body == b'ok'
assert calls == ['fail', 'fail', 'ok']
def test_raising_IRetryableError_type_is_caught(config):
from pyramid_retry import mark_error_retryable
class MyRetryableError(Exception):
pass
mark_error_retryable(MyRetryableError)
calls = []
def final_view(request):
calls.append('ok')
return 'ok'
def bad_view(request):
calls.append('fail')
raise MyRetryableError
config.add_view(bad_view, last_retry_attempt=False)
config.add_view(final_view, last_retry_attempt=True, renderer='string')
app = config.make_wsgi_app()
app = webtest.TestApp(app)
response = app.get('/')
assert response.body == b'ok'
assert calls == ['fail', 'fail', 'ok']
def test_raising_nonretryable_is_not_caught(config):
calls = []
def bad_view(request):
calls.append('fail')
raise Exception
config.add_view(bad_view)
app = config.make_wsgi_app()
app = webtest.TestApp(app)
with pytest.raises(Exception):
app.get('/')
assert calls == ['fail']
def test_handled_error_is_retried(config):
from pyramid_retry import RetryableException
calls = []
def bad_view(request):
calls.append('fail')
raise RetryableException
def retryable_exc_view(request):
calls.append('caught')
return 'caught'
def default_exc_view(request):
calls.append('default')
return 'default'
config.add_view(bad_view)
config.add_exception_view(default_exc_view, renderer='string')
config.add_exception_view(
retryable_exc_view, retryable_error=True, renderer='string')
app = config.make_wsgi_app()
app = webtest.TestApp(app)
response = app.get('/')
assert response.body == b'default'
assert calls == ['fail', 'caught', 'fail', 'caught', 'fail', 'default']
def test_retryable_exception_is_ignored_on_last_attempt(config):
from pyramid_retry import RetryableException
calls = []
def bad_view(request):
calls.append('fail')
raise RetryableException
config.add_view(bad_view)
app = config.make_wsgi_app()
app = webtest.TestApp(app)
with pytest.raises(Exception):
app.get('/')
assert calls == ['fail', 'fail', 'fail']
def test_BeforeRetry_event_is_raised(config):
from pyramid_retry import RetryableException
from pyramid_retry import IBeforeRetry
calls = []
retries = []
first_exception = RetryableException()
second_exception = RetryableException()
exceptions_to_be_raised = [first_exception, second_exception]
def retry_subscriber(event):
retries.append(event)
def final_view(request):
calls.append('ok')
return 'ok'
def bad_view(request):
calls.append('fail')
raise exceptions_to_be_raised.pop(0)
config.add_subscriber(retry_subscriber, IBeforeRetry)
config.add_view(bad_view, last_retry_attempt=False)
config.add_view(final_view, last_retry_attempt=True, renderer='string')
app = config.make_wsgi_app()
app = webtest.TestApp(app)
response = app.get('/')
assert response.body == b'ok'
assert calls == ['fail', 'fail', 'ok']
assert len(retries) == 2
assert retries[0].exception == first_exception
assert retries[0].response is None
assert retries[1].exception == second_exception
assert retries[1].response is None
def test_BeforeRetry_event_is_raised_from_squashed_exception(config):
from pyramid_retry import IBeforeRetry
from pyramid_retry import RetryableException
calls = []
retries = []
first_exception = RetryableException()
second_exception = RetryableException()
exceptions_to_be_raised = [first_exception, second_exception]
def retry_subscriber(event):
retries.append(event)
def final_view(request):
calls.append('ok')
return 'ok'
def bad_view(request):
raise exceptions_to_be_raised.pop(0)
def exc_view(request):
calls.append('squash')
return 'squash'
config.add_subscriber(retry_subscriber, IBeforeRetry)
config.add_view(bad_view, last_retry_attempt=False)
config.add_view(exc_view, context=RetryableException, exception_only=True,
renderer='string')
config.add_view(final_view, last_retry_attempt=True, renderer='string')
app = config.make_wsgi_app()
app = webtest.TestApp(app)
response = app.get('/')
assert response.body == b'ok'
assert calls == ['squash', 'squash', 'ok']
assert len(retries) == 2
assert retries[0].exception == first_exception
assert isinstance(retries[0].response, pyramid.response.Response)
assert retries[1].exception == second_exception
assert isinstance(retries[1].response, pyramid.response.Response)
def test_activate_hook_overrides_default_attempts(config):
from pyramid_retry import RetryableException
calls = []
def activate_hook(request):
return 1
def bad_view(request):
calls.append('fail')
raise RetryableException
config.add_settings({
'retry.attempts': 3,
'retry.activate_hook': activate_hook,
})
config.add_view(bad_view)
app = config.make_wsgi_app()
app = webtest.TestApp(app)
with pytest.raises(Exception):
app.get('/')
assert calls == ['fail']
def test_activate_hook_falls_back_to_default_attempts(config):
from pyramid_retry import RetryableException
calls = []
def activate_hook(request):
return None
def bad_view(request):
calls.append('fail')
raise RetryableException
config.add_settings({
'retry.attempts': 3,
'retry.activate_hook': activate_hook,
})
config.add_view(bad_view)
app = config.make_wsgi_app()
app = webtest.TestApp(app)
with pytest.raises(Exception):
app.get('/')
assert calls == ['fail', 'fail', 'fail']
def test_request_make_body_seekable_cleans_up_threadmanger_on_exception(config):
from pyramid.threadlocal import manager
# Clear defaults.
manager.pop()
assert len(manager.stack) == 0
app = config.make_wsgi_app()
app = webtest.TestApp(app)
with pytest.raises(Exception):
# Content-length=1 and empty body causes
# webob.request.DisconnectionError:
# The client disconnected while sending the body
# (1 more bytes were expected) when you call
# request.make_body_seekable().
app.get('/', headers={'Content-Length': '1'})
# len(manager.stack) == 1 when you don't catch exception
# from request.make_body_seekable() and clean up.
assert len(manager.stack) == 0
def test_activate_hook_cleans_up_threadmanager_on_exception(config):
from pyramid.threadlocal import manager
# Clear defaults.
manager.pop()
assert len(manager.stack) == 0
def activate_hook(request):
raise Exception
config.add_settings({
'retry.attempts': 3,
'retry.activate_hook': activate_hook,
})
app = config.make_wsgi_app()
app = webtest.TestApp(app)
with pytest.raises(Exception):
app.get('/')
# len(manager.stack) == 1 when you don't catch exception
# from activate_hook and clean up.
assert len(manager.stack) == 0
def test_activate_hook_cleans_up_threadmanager_on_generator_exit(config):
from pyramid.threadlocal import manager
# Clear defaults.
manager.pop()
assert len(manager.stack) == 0
def activate_hook(request):
raise GeneratorExit
config.add_settings({
'retry.attempts': 3,
'retry.activate_hook': activate_hook,
})
app = config.make_wsgi_app()
app = webtest.TestApp(app)
with pytest.raises(GeneratorExit):
app.get('/')
# len(manager.stack) == 1 when you don't catch GeneratorExit
# from activate_hook and clean up.
assert len(manager.stack) == 0
def test_is_last_attempt_True_when_inactive():
from pyramid_retry import is_last_attempt
request = pyramid.request.Request.blank('/')
assert is_last_attempt(request)
def test_retryable_error_predicate_is_bool(config):
from pyramid.exceptions import ConfigurationError
view = lambda r: 'ok'
with pytest.raises(ConfigurationError):
config.add_view(view, retryable_error='yes', renderer='string')
config.commit()
def test_last_retry_attempt_predicate_is_bool(config):
from pyramid.exceptions import ConfigurationError
view = lambda r: 'ok'
with pytest.raises(ConfigurationError):
config.add_view(view, last_retry_attempt='yes', renderer='string')
config.commit()
def test_mark_error_retryable_on_non_error():
from pyramid_retry import mark_error_retryable
with pytest.raises(ValueError):
mark_error_retryable('some string')
|
