0
# Configuration and Testing
1
2
Global configuration functions for controlling retry behavior system-wide and enabling test mode with modified retry parameters for faster test execution and predictable behavior.
3
4
## Capabilities
5
6
### Global Activation Control
7
8
Control whether retry behavior is active across the entire application. When deactivated, all retry decorators and contexts will execute only once without retries.
9
10
```python { .api }
11
def is_active() -> bool:
12
"""
13
Check whether retrying is globally active.
14
15
Returns:
16
bool: True if retrying is active, False otherwise
17
"""
18
19
def set_active(active: bool) -> None:
20
"""
21
Globally activate or deactivate retrying.
22
23
When deactivated, all retry decorators and retry contexts will
24
execute only once regardless of configured attempts.
25
26
Parameters:
27
- active: bool - Whether to activate retrying
28
29
This function is idempotent and can be called repeatedly.
30
"""
31
```
32
33
**Usage Examples:**
34
35
```python
36
import stamina
37
38
# Check current state
39
print(f"Retries active: {stamina.is_active()}") # True by default
40
41
# Temporarily disable retries
42
stamina.set_active(False)
43
44
@stamina.retry(on=Exception, attempts=5)
45
def unreliable_function():
46
raise ValueError("This will not retry")
47
48
try:
49
unreliable_function() # Fails immediately, no retries
50
except ValueError:
51
print("Function failed without retries")
52
53
# Re-enable retries
54
stamina.set_active(True)
55
56
# Emergency circuit breaker pattern
57
def emergency_disable_retries():
58
"""Disable retries when system is under stress."""
59
stamina.set_active(False)
60
logger.warning("Retries disabled due to system overload")
61
62
def restore_retries():
63
"""Re-enable retries when system recovers."""
64
stamina.set_active(True)
65
logger.info("Retries re-enabled")
66
```
67
68
### Test Mode Configuration
69
70
Test mode provides predictable retry behavior for testing scenarios by disabling backoff delays and controlling attempt counts.
71
72
```python { .api }
73
def is_testing() -> bool:
74
"""
75
Check whether test mode is enabled.
76
77
Returns:
78
bool: True if test mode is active, False otherwise
79
"""
80
81
def set_testing(
82
testing: bool,
83
*,
84
attempts: int = 1,
85
cap: bool = False
86
) -> _RestoreTestingCM:
87
"""
88
Activate or deactivate test mode.
89
90
In test mode:
91
- All backoff delays are set to 0 (no waiting)
92
- Attempt counts are controlled by the attempts parameter
93
94
Parameters:
95
- testing: bool - Whether to enable test mode
96
- attempts: int - Number of attempts in test mode (default: 1)
97
- cap: bool - If True, cap attempts rather than override (default: False)
98
99
Returns:
100
Context manager that restores previous testing state when exited
101
102
When cap=True:
103
- If attempts=3 and user specifies 5 attempts, use 3 (capped)
104
- If attempts=3 and user specifies 2 attempts, use 2 (user's value)
105
106
This function can be used as a context manager and is idempotent.
107
"""
108
```
109
110
**Usage Examples:**
111
112
```python
113
import stamina
114
115
# Basic test mode usage
116
def test_retry_behavior():
117
# Enable test mode with single attempt
118
stamina.set_testing(True, attempts=1)
119
120
@stamina.retry(on=ValueError, attempts=5) # Will only try once
121
def failing_function():
122
raise ValueError("Test failure")
123
124
try:
125
failing_function()
126
assert False, "Should have failed"
127
except ValueError:
128
pass # Expected
129
130
# Disable test mode
131
stamina.set_testing(False)
132
133
# Test mode with multiple attempts
134
def test_eventual_success():
135
stamina.set_testing(True, attempts=3) # Allow 3 attempts in test
136
137
call_count = 0
138
139
@stamina.retry(on=ValueError, attempts=10) # Normally 10, but test caps at 3
140
def eventually_succeeds():
141
nonlocal call_count
142
call_count += 1
143
if call_count < 3:
144
raise ValueError("Not yet")
145
return "success"
146
147
result = eventually_succeeds()
148
assert result == "success"
149
assert call_count == 3
150
151
stamina.set_testing(False)
152
153
# Context manager usage
154
def test_with_context_manager():
155
with stamina.set_testing(True, attempts=2):
156
# Test code here runs with test mode
157
@stamina.retry(on=Exception, attempts=5)
158
def test_function():
159
raise Exception("Test")
160
161
try:
162
test_function() # Will try 2 times, not 5
163
except Exception:
164
pass
165
166
# Test mode automatically restored after context
167
168
# Capped attempts mode
169
def test_capped_attempts():
170
with stamina.set_testing(True, attempts=3, cap=True):
171
# User specifies 2 attempts, test allows up to 3, so use 2
172
@stamina.retry(on=ValueError, attempts=2)
173
def function_a():
174
raise ValueError("Test")
175
176
# User specifies 5 attempts, test caps at 3, so use 3
177
@stamina.retry(on=ValueError, attempts=5)
178
def function_b():
179
raise ValueError("Test")
180
```
181
182
### Testing Patterns
183
184
Common patterns for testing retry behavior:
185
186
```python
187
import pytest
188
import stamina
189
190
class TestRetryBehavior:
191
def setup_method(self):
192
"""Reset retry state before each test."""
193
stamina.set_active(True)
194
stamina.set_testing(False)
195
196
def test_successful_retry(self):
197
"""Test that retries eventually succeed."""
198
with stamina.set_testing(True, attempts=3):
199
call_count = 0
200
201
@stamina.retry(on=ValueError, attempts=5)
202
def flaky_function():
203
nonlocal call_count
204
call_count += 1
205
if call_count < 3:
206
raise ValueError("Not ready")
207
return "success"
208
209
result = flaky_function()
210
assert result == "success"
211
assert call_count == 3
212
213
def test_exhausted_retries(self):
214
"""Test behavior when all retries are exhausted."""
215
with stamina.set_testing(True, attempts=2):
216
call_count = 0
217
218
@stamina.retry(on=ValueError, attempts=3) # Will be capped at 2
219
def always_fails():
220
nonlocal call_count
221
call_count += 1
222
raise ValueError(f"Failure {call_count}")
223
224
with pytest.raises(ValueError, match="Failure 2"):
225
always_fails()
226
227
assert call_count == 2
228
229
def test_no_retries_when_inactive(self):
230
"""Test that inactive mode prevents retries."""
231
stamina.set_active(False)
232
233
call_count = 0
234
235
@stamina.retry(on=ValueError, attempts=5)
236
def should_not_retry():
237
nonlocal call_count
238
call_count += 1
239
raise ValueError("Single failure")
240
241
with pytest.raises(ValueError):
242
should_not_retry()
243
244
assert call_count == 1 # Called only once
245
246
def test_context_manager_cleanup(self):
247
"""Test that context managers properly restore state."""
248
original_testing = stamina.is_testing()
249
250
try:
251
with stamina.set_testing(True, attempts=2):
252
assert stamina.is_testing() == True
253
# Test code here
254
255
# State should be restored
256
assert stamina.is_testing() == original_testing
257
finally:
258
stamina.set_testing(original_testing)
259
260
# Pytest fixtures for common test setups
261
@pytest.fixture
262
def no_retries():
263
"""Fixture to disable retries for fast tests."""
264
with stamina.set_testing(True, attempts=1):
265
yield
266
267
@pytest.fixture
268
def fast_retries():
269
"""Fixture for tests that need some retries but fast execution."""
270
with stamina.set_testing(True, attempts=3):
271
yield
272
273
def test_with_no_retries(no_retries):
274
"""Test using the no_retries fixture."""
275
@stamina.retry(on=ValueError, attempts=10)
276
def fails_once():
277
raise ValueError("Test failure")
278
279
with pytest.raises(ValueError):
280
fails_once() # Fails immediately
281
282
def test_with_fast_retries(fast_retries):
283
"""Test using the fast_retries fixture."""
284
call_count = 0
285
286
@stamina.retry(on=ValueError, attempts=10)
287
def succeeds_on_third():
288
nonlocal call_count
289
call_count += 1
290
if call_count < 3:
291
raise ValueError("Not yet")
292
return "success"
293
294
result = succeeds_on_third()
295
assert result == "success"
296
```
297
298
## Configuration Best Practices
299
300
### Application Startup
301
302
Configure retry behavior during application initialization:
303
304
```python
305
import stamina
306
import os
307
308
def configure_retries():
309
"""Configure retry behavior based on environment."""
310
env = os.getenv("ENVIRONMENT", "production")
311
312
if env == "test":
313
# Fast, predictable retries for tests
314
stamina.set_testing(True, attempts=2)
315
elif env == "development":
316
# Shorter timeouts for development
317
# Use default active behavior but configure individual retries
318
pass
319
elif env == "production":
320
# Full retry behavior (default)
321
stamina.set_active(True)
322
else:
323
raise ValueError(f"Unknown environment: {env}")
324
325
# Call during app startup
326
configure_retries()
327
```
328
329
### Health Check Integration
330
331
Integrate with application health checks:
332
333
```python
334
import stamina
335
from your_app import health_check
336
337
def monitor_system_health():
338
"""Disable retries when system is unhealthy."""
339
if not health_check.is_healthy():
340
stamina.set_active(False)
341
logger.warning("Retries disabled due to system health issues")
342
else:
343
stamina.set_active(True)
344
345
# Run periodically
346
import threading
347
import time
348
349
def health_monitor_loop():
350
while True:
351
monitor_system_health()
352
time.sleep(30) # Check every 30 seconds
353
354
health_thread = threading.Thread(target=health_monitor_loop, daemon=True)
355
health_thread.start()
356
```
357
358
## Types
359
360
```python { .api }
361
class _RestoreTestingCM:
362
"""Context manager that restores previous testing state when exited."""
363
def __enter__(self) -> None: ...
364
def __exit__(self, exc_type, exc_val, exc_tb) -> None: ...
365
```