0
# Test Execution Control
1
2
Control how individual tests run including custom test execution, skip decorators, concurrent test execution, and specialized test suite implementations.
3
4
## Capabilities
5
6
### Custom Test Execution
7
8
RunTest class for controlling individual test execution lifecycle.
9
10
```python { .api }
11
class RunTest:
12
"""
13
Controls how individual tests are run.
14
15
Provides hooks for custom test setup, execution,
16
and teardown with support for different execution strategies.
17
"""
18
19
def __init__(self, case, handlers=None):
20
"""
21
Create test runner for a test case.
22
23
Args:
24
case: TestCase instance to run
25
handlers: Optional exception handlers
26
"""
27
28
def run(self, result):
29
"""
30
Execute the test case.
31
32
Args:
33
result: TestResult to record results
34
35
Returns:
36
The test result
37
"""
38
39
def setUp(self):
40
"""
41
Perform test setup.
42
43
Called before test method execution.
44
Override for custom setup logic.
45
"""
46
47
def tearDown(self):
48
"""
49
Perform test teardown.
50
51
Called after test method execution.
52
Override for custom cleanup logic.
53
"""
54
55
def _run_user(self, fn):
56
"""
57
Execute user test method with exception handling.
58
59
Args:
60
fn: Test method to execute
61
"""
62
63
def _got_user_exception(self, exc_info):
64
"""
65
Handle exceptions from user test code.
66
67
Args:
68
exc_info: Exception information tuple
69
"""
70
```
71
72
### Exception Handling
73
74
Classes for managing test exceptions and error conditions.
75
76
```python { .api }
77
class MultipleExceptions(Exception):
78
"""
79
Exception class representing multiple test failures.
80
81
Collects multiple exceptions that occurred during
82
test execution for comprehensive error reporting.
83
"""
84
85
def __init__(self, *args):
86
"""
87
Create multiple exceptions container.
88
89
Args:
90
*args: Exception instances or information
91
"""
92
93
def __str__(self):
94
"""
95
String representation of all exceptions.
96
97
Returns:
98
str: Formatted list of all exceptions
99
"""
100
```
101
102
### Test Suite Implementations
103
104
Specialized test suite classes for different execution strategies.
105
106
```python { .api }
107
class ConcurrentTestSuite(unittest.TestSuite):
108
"""
109
Run tests concurrently across multiple processes/threads.
110
111
Provides parallel test execution for improved performance
112
with proper result aggregation and error handling.
113
"""
114
115
def __init__(self, suite, fork_for_tests):
116
"""
117
Create concurrent test suite.
118
119
Args:
120
suite: Test suite to run concurrently
121
fork_for_tests: Function to create test processes
122
"""
123
124
def run(self, result):
125
"""
126
Execute tests concurrently.
127
128
Args:
129
result: TestResult to collect results
130
131
Returns:
132
The aggregated test result
133
"""
134
135
class ConcurrentStreamTestSuite(unittest.TestSuite):
136
"""
137
Concurrent test suite using stream results.
138
139
Combines concurrent execution with streaming
140
result reporting for real-time feedback.
141
"""
142
143
def __init__(self, suite, fork_runner):
144
"""
145
Create concurrent streaming test suite.
146
147
Args:
148
suite: Test suite to run
149
fork_runner: Process creation function
150
"""
151
152
def run(self, result):
153
"""
154
Execute tests with concurrent streaming.
155
156
Args:
157
result: StreamResult for real-time reporting
158
159
Returns:
160
The stream result
161
"""
162
163
class FixtureSuite(unittest.TestSuite):
164
"""
165
Test suite with fixture support for shared setup/teardown.
166
167
Manages fixtures that need to be shared across
168
multiple tests for resource optimization.
169
"""
170
171
def __init__(self, fixture, tests):
172
"""
173
Create fixture-based test suite.
174
175
Args:
176
fixture: Fixture instance for setup/teardown
177
tests: Test cases or suites to run
178
"""
179
180
def run(self, result):
181
"""
182
Execute tests with fixture management.
183
184
Args:
185
result: TestResult to collect results
186
187
Returns:
188
The test result
189
"""
190
```
191
192
### Test Discovery and Organization
193
194
Functions for working with test suites and test organization.
195
196
```python { .api }
197
def iterate_tests(test_suite_or_case):
198
"""
199
Iterate through all individual tests in a test suite.
200
201
Recursively flattens test suites to yield individual
202
test cases for processing or analysis.
203
204
Args:
205
test_suite_or_case: TestSuite or TestCase to iterate
206
207
Yields:
208
TestCase: Individual test cases
209
"""
210
211
def filter_by_ids(test_suite, test_ids):
212
"""
213
Filter tests by test IDs.
214
215
Args:
216
test_suite: TestSuite to filter
217
test_ids: Set of test IDs to include
218
219
Returns:
220
TestSuite: Filtered test suite
221
"""
222
223
def sorted_tests(test_suite, cmp_func=None):
224
"""
225
Sort tests by some criteria.
226
227
Args:
228
test_suite: TestSuite to sort
229
cmp_func: Optional comparison function
230
231
Returns:
232
TestSuite: Sorted test suite
233
"""
234
```
235
236
### Test Decorators and Utilities
237
238
Functions for test method decoration and execution control.
239
240
```python { .api }
241
def run_test_with(test_runner):
242
"""
243
Decorator to specify custom test runner for a test method.
244
245
Args:
246
test_runner: RunTest class or callable
247
248
Returns:
249
Function: Test method decorator
250
251
Example:
252
@run_test_with(CustomRunTest)
253
def test_custom_execution(self):
254
pass
255
"""
256
257
def attr(**kwargs):
258
"""
259
Add attributes to test methods.
260
261
Args:
262
**kwargs: Attribute name-value pairs
263
264
Returns:
265
Function: Test method decorator
266
267
Example:
268
@attr(speed='slow', category='integration')
269
def test_database_integration(self):
270
pass
271
"""
272
273
def gather_details(source_dict, target_dict):
274
"""
275
Gather details from multiple sources into target dictionary.
276
277
Args:
278
source_dict (dict): Source details dictionary
279
target_dict (dict): Target details dictionary
280
"""
281
```
282
283
## Usage Examples
284
285
### Custom Test Runner
286
287
```python
288
import testtools
289
290
class TimingRunTest(testtools.RunTest):
291
"""Custom runner that times test execution."""
292
293
def __init__(self, case, handlers=None):
294
super().__init__(case, handlers)
295
self.start_time = None
296
self.end_time = None
297
298
def setUp(self):
299
import time
300
self.start_time = time.time()
301
super().setUp()
302
303
def tearDown(self):
304
import time
305
super().tearDown()
306
self.end_time = time.time()
307
duration = self.end_time - self.start_time
308
self.case.addDetail('execution_time',
309
testtools.content.text_content(f"{duration:.3f}s"))
310
311
class MyTest(testtools.TestCase):
312
313
@testtools.run_test_with(TimingRunTest)
314
def test_with_timing(self):
315
import time
316
time.sleep(0.1) # Simulate work
317
self.assertTrue(True)
318
```
319
320
### Concurrent Test Execution
321
322
```python
323
import testtools
324
from testtools.testsuite import ConcurrentTestSuite
325
326
def fork_for_tests(suite):
327
"""Simple process forker for tests."""
328
import multiprocessing
329
pool = multiprocessing.Pool(processes=4)
330
return pool
331
332
# Create test suite
333
suite = testtools.TestSuite()
334
suite.addTest(SlowTest('test_method_1'))
335
suite.addTest(SlowTest('test_method_2'))
336
suite.addTest(SlowTest('test_method_3'))
337
suite.addTest(SlowTest('test_method_4'))
338
339
# Run concurrently
340
concurrent_suite = ConcurrentTestSuite(suite, fork_for_tests)
341
result = testtools.TestResult()
342
concurrent_suite.run(result)
343
344
print(f"Executed {result.testsRun} tests concurrently")
345
```
346
347
### Fixture-Based Test Suite
348
349
```python
350
import testtools
351
from testtools.testsuite import FixtureSuite
352
from fixtures import TempDir
353
354
class DatabaseTest(testtools.TestCase):
355
def test_user_creation(self):
356
# Test uses shared database fixture
357
user = create_user("test_user")
358
self.assertIsNotNone(user.id)
359
360
def test_user_deletion(self):
361
# Test uses same shared database
362
user = create_user("delete_me")
363
delete_user(user.id)
364
self.assertIsNone(get_user(user.id))
365
366
# Create fixture suite with shared database
367
database_fixture = DatabaseFixture()
368
suite = FixtureSuite(database_fixture, [
369
DatabaseTest('test_user_creation'),
370
DatabaseTest('test_user_deletion'),
371
])
372
373
result = testtools.TestResult()
374
suite.run(result)
375
```
376
377
### Test Discovery and Filtering
378
379
```python
380
import testtools
381
382
def discover_slow_tests(suite):
383
"""Find all tests marked as slow."""
384
slow_tests = []
385
for test in testtools.iterate_tests(suite):
386
if hasattr(test, 'speed') and test.speed == 'slow':
387
slow_tests.append(test)
388
return slow_tests
389
390
# Create comprehensive test suite
391
full_suite = testtools.TestSuite()
392
# ... add many tests ...
393
394
# Filter for specific test IDs
395
test_ids = {'MyTest.test_specific', 'OtherTest.test_important'}
396
filtered_suite = filter_by_ids(full_suite, test_ids)
397
398
# Run only filtered tests
399
result = testtools.TestResult()
400
filtered_suite.run(result)
401
```
402
403
### Custom Exception Handling
404
405
```python
406
import testtools
407
408
class RetryRunTest(testtools.RunTest):
409
"""Runner that retries failed tests."""
410
411
def __init__(self, case, max_retries=3):
412
super().__init__(case)
413
self.max_retries = max_retries
414
415
def run(self, result):
416
for attempt in range(self.max_retries + 1):
417
temp_result = testtools.TestResult()
418
super().run(temp_result)
419
420
if temp_result.wasSuccessful():
421
# Success, forward to real result
422
result.addSuccess(self.case)
423
break
424
elif attempt == self.max_retries:
425
# Final attempt failed, forward failure
426
for error in temp_result.errors:
427
result.addError(*error)
428
for failure in temp_result.failures:
429
result.addFailure(*failure)
430
else:
431
# Retry
432
self.case.addDetail(f'retry_{attempt}',
433
testtools.content.text_content(f"Retrying after failure"))
434
435
class FlakyTest(testtools.TestCase):
436
437
@testtools.run_test_with(lambda case: RetryRunTest(case, max_retries=2))
438
def test_flaky_operation(self):
439
import random
440
if random.random() < 0.7: # 70% chance of failure
441
self.fail("Random failure")
442
self.assertTrue(True)
443
```
444
445
### Test Attribution and Metadata
446
447
```python
448
import testtools
449
450
class MetadataTest(testtools.TestCase):
451
452
@testtools.attr(category='unit', priority='high', owner='alice')
453
def test_critical_function(self):
454
self.assertEqual(critical_function(), 'expected')
455
456
@testtools.attr(category='integration', priority='low', owner='bob')
457
def test_external_service(self):
458
result = call_external_service()
459
self.assertIsNotNone(result)
460
461
# Discover tests by attributes
462
def find_tests_by_owner(suite, owner):
463
"""Find tests owned by specific person."""
464
return [test for test in testtools.iterate_tests(suite)
465
if hasattr(test, 'owner') and test.owner == owner]
466
467
def find_high_priority_tests(suite):
468
"""Find high priority tests."""
469
return [test for test in testtools.iterate_tests(suite)
470
if hasattr(test, 'priority') and test.priority == 'high']
471
472
# Use metadata for test organization
473
suite = testtools.TestSuite()
474
# ... add tests ...
475
476
alice_tests = find_tests_by_owner(suite, 'alice')
477
high_priority = find_high_priority_tests(suite)
478
```