0
# Twisted Integration
1
2
Complete integration with Twisted framework for testing asynchronous code using Deferreds, including specialized matchers and test execution classes.
3
4
## Capabilities
5
6
### Deferred Matchers
7
8
Specialized matchers for testing Twisted Deferred objects and asynchronous operations.
9
10
```python { .api }
11
def succeeded():
12
"""
13
Match successfully resolved Twisted Deferred objects.
14
15
Matches Deferred objects that have fired successfully
16
with a result value (not an error).
17
18
Returns:
19
Matcher: Deferred success matcher
20
21
Example:
22
self.assertThat(deferred, succeeded())
23
"""
24
25
def failed():
26
"""
27
Match failed Twisted Deferred objects.
28
29
Matches Deferred objects that have fired with an
30
error/exception rather than a successful result.
31
32
Returns:
33
Matcher: Deferred failure matcher
34
35
Example:
36
self.assertThat(deferred, failed())
37
"""
38
39
def has_no_result():
40
"""
41
Match Deferred objects with no result yet.
42
43
Matches Deferred objects that have not yet fired
44
(neither success nor failure).
45
46
Returns:
47
Matcher: Deferred pending matcher
48
49
Example:
50
self.assertThat(pending_deferred, has_no_result())
51
"""
52
```
53
54
### Asynchronous Test Execution
55
56
Test runner classes for executing tests that return Deferred objects.
57
58
```python { .api }
59
class AsynchronousDeferredRunTest(testtools.RunTest):
60
"""
61
Run Deferred-returning tests asynchronously.
62
63
Executes test methods that return Deferred objects,
64
properly handling the asynchronous completion and
65
result propagation.
66
"""
67
68
def __init__(self, case, handlers=None, reactor=None, timeout=None):
69
"""
70
Create asynchronous Deferred test runner.
71
72
Args:
73
case: TestCase instance
74
handlers: Optional exception handlers
75
reactor: Twisted reactor instance
76
timeout (float): Optional test timeout in seconds
77
"""
78
79
def run(self, result):
80
"""
81
Execute the test asynchronously.
82
83
Runs the test method and waits for Deferred completion,
84
handling both success and failure cases appropriately.
85
86
Args:
87
result: TestResult to record results
88
89
Returns:
90
TestResult: The test result after completion
91
"""
92
93
class AsynchronousDeferredRunTestForBrokenTwisted(AsynchronousDeferredRunTest):
94
"""
95
Workaround for broken Twisted versions.
96
97
Alternative asynchronous test runner that works around
98
known issues in certain Twisted versions or configurations.
99
"""
100
101
def __init__(self, case, handlers=None, reactor=None, timeout=None):
102
"""
103
Create workaround asynchronous test runner.
104
105
Args:
106
case: TestCase instance
107
handlers: Optional exception handlers
108
reactor: Twisted reactor instance
109
timeout (float): Optional test timeout
110
"""
111
112
class SynchronousDeferredRunTest(testtools.RunTest):
113
"""
114
Run Deferred-returning tests synchronously.
115
116
Executes Deferred-returning tests in a synchronous manner
117
by spinning the reactor until completion.
118
"""
119
120
def __init__(self, case, handlers=None, reactor=None, timeout=None):
121
"""
122
Create synchronous Deferred test runner.
123
124
Args:
125
case: TestCase instance
126
handlers: Optional exception handlers
127
reactor: Twisted reactor instance
128
timeout (float): Optional test timeout
129
"""
130
131
def run(self, result):
132
"""
133
Execute the test synchronously.
134
135
Runs the test and spins the reactor until the
136
Deferred completes, then returns the result.
137
138
Args:
139
result: TestResult to record results
140
141
Returns:
142
TestResult: The test result after completion
143
"""
144
```
145
146
### Log Capture and Management
147
148
Classes for capturing and managing Twisted log output during tests.
149
150
```python { .api }
151
class CaptureTwistedLogs(testtools.RunTest):
152
"""
153
Capture Twisted log output during test execution.
154
155
Intercepts Twisted log messages generated during test
156
execution and attaches them to test results for debugging.
157
"""
158
159
def __init__(self, case, handlers=None):
160
"""
161
Create log capturing test runner.
162
163
Args:
164
case: TestCase instance
165
handlers: Optional exception handlers
166
"""
167
168
def run(self, result):
169
"""
170
Execute test with log capture.
171
172
Runs the test while capturing all Twisted log output
173
and attaching it to the test result for analysis.
174
175
Args:
176
result: TestResult to record results and logs
177
178
Returns:
179
TestResult: Test result with captured logs
180
"""
181
```
182
183
### Helper Functions
184
185
Utility functions for common Twisted testing patterns.
186
187
```python { .api }
188
def assert_fails_with(deferred, *expected_failures):
189
"""
190
Assert that Deferred fails with specific error type.
191
192
Verifies that a Deferred object fails and that the
193
failure matches one of the expected error types.
194
195
Args:
196
deferred: Deferred object to check
197
*expected_failures: Expected exception types
198
199
Returns:
200
Deferred: Deferred that fires when assertion completes
201
202
Example:
203
d = failing_operation()
204
return assert_fails_with(d, ValueError, TypeError)
205
"""
206
207
def flush_logged_errors(*error_types):
208
"""
209
Flush logged errors from Twisted logging system.
210
211
Removes accumulated error log entries of specified types
212
from Twisted's global error log, useful for test cleanup.
213
214
Args:
215
*error_types: Error types to flush from logs
216
217
Returns:
218
list: List of flushed error entries
219
220
Example:
221
# Clean up expected errors after test
222
flush_logged_errors(ConnectionLost, TimeoutError)
223
"""
224
```
225
226
## Usage Examples
227
228
### Basic Deferred Testing
229
230
```python
231
import testtools
232
from testtools.twistedsupport import *
233
from twisted.internet import defer, reactor
234
from twisted.internet.defer import inlineCallbacks
235
236
class AsyncTest(testtools.TestCase):
237
238
def test_successful_deferred(self):
239
"""Test Deferred that completes successfully."""
240
d = defer.succeed("test result")
241
self.assertThat(d, succeeded())
242
243
def test_failed_deferred(self):
244
"""Test Deferred that fails."""
245
d = defer.fail(ValueError("test error"))
246
self.assertThat(d, failed())
247
248
def test_pending_deferred(self):
249
"""Test Deferred that hasn't fired yet."""
250
d = defer.Deferred()
251
self.assertThat(d, has_no_result())
252
253
# Later, fire the deferred
254
d.callback("result")
255
self.assertThat(d, succeeded())
256
```
257
258
### Asynchronous Test Methods
259
260
```python
261
import testtools
262
from testtools.twistedsupport import AsynchronousDeferredRunTest
263
from twisted.internet import defer, reactor
264
from twisted.internet.defer import inlineCallbacks
265
import time
266
267
class AsyncOperationTest(testtools.TestCase):
268
269
@testtools.run_test_with(AsynchronousDeferredRunTest)
270
def test_async_operation(self):
271
"""Test method that returns a Deferred."""
272
def slow_operation():
273
d = defer.Deferred()
274
reactor.callLater(0.1, d.callback, "completed")
275
return d
276
277
# Return Deferred from test method
278
d = slow_operation()
279
d.addCallback(lambda result: self.assertEqual(result, "completed"))
280
return d
281
282
@testtools.run_test_with(AsynchronousDeferredRunTest)
283
@inlineCallbacks
284
def test_with_inline_callbacks(self):
285
"""Test using inlineCallbacks decorator."""
286
def async_fetch(url):
287
d = defer.Deferred()
288
reactor.callLater(0.05, d.callback, f"data from {url}")
289
return d
290
291
# Use yield with inlineCallbacks
292
result1 = yield async_fetch("http://example.com/data1")
293
result2 = yield async_fetch("http://example.com/data2")
294
295
self.assertIn("data1", result1)
296
self.assertIn("data2", result2)
297
```
298
299
### Error Handling and Failures
300
301
```python
302
import testtools
303
from testtools.twistedsupport import *
304
from twisted.internet import defer
305
from twisted.internet.error import ConnectionLost, TimeoutError
306
307
class ErrorHandlingTest(testtools.TestCase):
308
309
@testtools.run_test_with(AsynchronousDeferredRunTest)
310
def test_expected_failure(self):
311
"""Test that expects a specific failure."""
312
def failing_operation():
313
return defer.fail(ConnectionLost("Connection dropped"))
314
315
d = failing_operation()
316
return assert_fails_with(d, ConnectionLost)
317
318
@testtools.run_test_with(AsynchronousDeferredRunTest)
319
def test_multiple_possible_failures(self):
320
"""Test that can fail with different error types."""
321
def unstable_operation():
322
import random
323
if random.random() < 0.5:
324
return defer.fail(TimeoutError("Request timed out"))
325
else:
326
return defer.fail(ConnectionLost("Connection lost"))
327
328
d = unstable_operation()
329
return assert_fails_with(d, TimeoutError, ConnectionLost)
330
331
def test_error_cleanup(self):
332
"""Test with error cleanup."""
333
# Generate some errors that get logged
334
defer.fail(ValueError("test error 1"))
335
defer.fail(TypeError("test error 2"))
336
337
# Clean up the logged errors
338
flushed = flush_logged_errors(ValueError, TypeError)
339
self.assertEqual(len(flushed), 2)
340
```
341
342
### Log Capture Testing
343
344
```python
345
import testtools
346
from testtools.twistedsupport import CaptureTwistedLogs
347
from twisted.python import log
348
from twisted.internet import defer
349
350
class LogCaptureTest(testtools.TestCase):
351
352
@testtools.run_test_with(CaptureTwistedLogs)
353
def test_with_log_capture(self):
354
"""Test that captures Twisted logs."""
355
def operation_with_logging():
356
log.msg("Starting operation")
357
log.msg("Processing data")
358
d = defer.succeed("result")
359
d.addCallback(lambda r: log.msg(f"Operation completed: {r}") or r)
360
return d
361
362
d = operation_with_logging()
363
d.addCallback(lambda result: self.assertEqual(result, "result"))
364
return d
365
# Logs are automatically captured and attached to test result
366
```
367
368
### Complex Asynchronous Testing
369
370
```python
371
import testtools
372
from testtools.twistedsupport import *
373
from twisted.internet import defer, reactor
374
from twisted.internet.defer import inlineCallbacks, gatherResults
375
376
class ComplexAsyncTest(testtools.TestCase):
377
378
@testtools.run_test_with(AsynchronousDeferredRunTest)
379
@inlineCallbacks
380
def test_multiple_async_operations(self):
381
"""Test coordinating multiple async operations."""
382
def fetch_user(user_id):
383
d = defer.Deferred()
384
reactor.callLater(0.1, d.callback, {"id": user_id, "name": f"User{user_id}"})
385
return d
386
387
def fetch_permissions(user_id):
388
d = defer.Deferred()
389
reactor.callLater(0.05, d.callback, ["read", "write"])
390
return d
391
392
# Fetch user data and permissions concurrently
393
user_d = fetch_user(123)
394
perms_d = fetch_permissions(123)
395
396
user, permissions = yield gatherResults([user_d, perms_d])
397
398
self.assertEqual(user["id"], 123)
399
self.assertIn("read", permissions)
400
401
@testtools.run_test_with(AsynchronousDeferredRunTest)
402
def test_chained_operations(self):
403
"""Test chained asynchronous operations."""
404
def step1():
405
return defer.succeed("step1_result")
406
407
def step2(prev_result):
408
self.assertEqual(prev_result, "step1_result")
409
return defer.succeed("step2_result")
410
411
def step3(prev_result):
412
self.assertEqual(prev_result, "step2_result")
413
return defer.succeed("final_result")
414
415
# Chain the operations
416
d = step1()
417
d.addCallback(step2)
418
d.addCallback(step3)
419
d.addCallback(lambda result: self.assertEqual(result, "final_result"))
420
return d
421
422
@testtools.run_test_with(AsynchronousDeferredRunTest)
423
def test_timeout_handling(self):
424
"""Test operations with timeout."""
425
def slow_operation():
426
d = defer.Deferred()
427
# This operation takes too long
428
reactor.callLater(2.0, d.callback, "too slow")
429
return d
430
431
# Set up timeout
432
d = slow_operation()
433
timeout_d = defer.Deferred()
434
reactor.callLater(0.5, timeout_d.errback,
435
TimeoutError("Operation timed out"))
436
437
# Race between operation and timeout
438
racing_d = defer.DeferredList([d, timeout_d],
439
fireOnOneCallback=True,
440
fireOnOneErrback=True,
441
consumeErrors=True)
442
443
return assert_fails_with(racing_d, TimeoutError)
444
```
445
446
### Integration with TestCase Features
447
448
```python
449
import testtools
450
from testtools.twistedsupport import *
451
from testtools.matchers import *
452
from twisted.internet import defer
453
454
class IntegratedAsyncTest(testtools.TestCase):
455
456
@testtools.run_test_with(AsynchronousDeferredRunTest)
457
def test_with_matchers_and_content(self):
458
"""Test combining async testing with matchers and content."""
459
def fetch_data():
460
# Simulate fetching data
461
data = {
462
"users": [{"name": "Alice"}, {"name": "Bob"}],
463
"total": 2,
464
"status": "success"
465
}
466
return defer.succeed(data)
467
468
def verify_data(data):
469
# Attach data to test results for debugging
470
self.addDetail('fetched_data',
471
testtools.content.json_content(data))
472
473
# Use matchers for sophisticated assertions
474
self.assertThat(data, MatchesDict({
475
"users": HasLength(2),
476
"total": Equals(2),
477
"status": Equals("success")
478
}))
479
480
# Verify user structure
481
self.assertThat(data["users"], AllMatch(
482
MatchesDict({"name": IsInstance(str)})
483
))
484
485
return data
486
487
d = fetch_data()
488
d.addCallback(verify_data)
489
return d
490
491
@testtools.run_test_with(AsynchronousDeferredRunTest)
492
@testtools.skipIf(not twisted_available, "Twisted not available")
493
def test_conditional_async(self):
494
"""Test with conditional execution."""
495
def maybe_async_operation():
496
if should_run_async():
497
return defer.succeed("async_result")
498
else:
499
return "sync_result"
500
501
result = yield maybe_async_operation()
502
self.assertIn("result", result)
503
```