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
|
# Copyright (c) 2009-2011 testtools developers. See LICENSE for details.
"""Test suites and related things."""
__metaclass__ = type
__all__ = [
'ConcurrentTestSuite',
'iterate_tests',
'sorted_tests',
]
from testtools.helpers import safe_hasattr, try_imports
Queue = try_imports(['Queue.Queue', 'queue.Queue'])
import threading
import unittest
import testtools
def iterate_tests(test_suite_or_case):
"""Iterate through all of the test cases in 'test_suite_or_case'."""
try:
suite = iter(test_suite_or_case)
except TypeError:
yield test_suite_or_case
else:
for test in suite:
for subtest in iterate_tests(test):
yield subtest
class ConcurrentTestSuite(unittest.TestSuite):
"""A TestSuite whose run() calls out to a concurrency strategy."""
def __init__(self, suite, make_tests, wrap_result=None):
"""Create a ConcurrentTestSuite to execute suite.
:param suite: A suite to run concurrently.
:param make_tests: A helper function to split the tests in the
ConcurrentTestSuite into some number of concurrently executing
sub-suites. make_tests must take a suite, and return an iterable
of TestCase-like object, each of which must have a run(result)
method.
:param wrap_result: An optional function that takes a thread-safe
result and a thread number and must return a ``TestResult``
object. If not provided, then ``ConcurrentTestSuite`` will just
use a ``ThreadsafeForwardingResult`` wrapped around the result
passed to ``run()``.
"""
super(ConcurrentTestSuite, self).__init__([suite])
self.make_tests = make_tests
if wrap_result:
self._wrap_result = wrap_result
def _wrap_result(self, thread_safe_result, thread_number):
"""Wrap a thread-safe result before sending it test results.
You can either override this in a subclass or pass your own
``wrap_result`` in to the constructor. The latter is preferred.
"""
return thread_safe_result
def run(self, result):
"""Run the tests concurrently.
This calls out to the provided make_tests helper, and then serialises
the results so that result only sees activity from one TestCase at
a time.
ConcurrentTestSuite provides no special mechanism to stop the tests
returned by make_tests, it is up to the make_tests to honour the
shouldStop attribute on the result object they are run with, which will
be set if an exception is raised in the thread which
ConcurrentTestSuite.run is called in.
"""
tests = self.make_tests(self)
try:
threads = {}
queue = Queue()
semaphore = threading.Semaphore(1)
for i, test in enumerate(tests):
process_result = self._wrap_result(
testtools.ThreadsafeForwardingResult(result, semaphore), i)
reader_thread = threading.Thread(
target=self._run_test, args=(test, process_result, queue))
threads[test] = reader_thread, process_result
reader_thread.start()
while threads:
finished_test = queue.get()
threads[finished_test][0].join()
del threads[finished_test]
except:
for thread, process_result in threads.values():
process_result.stop()
raise
def _run_test(self, test, process_result, queue):
try:
test.run(process_result)
finally:
queue.put(test)
class FixtureSuite(unittest.TestSuite):
def __init__(self, fixture, tests):
super(FixtureSuite, self).__init__(tests)
self._fixture = fixture
def run(self, result):
self._fixture.setUp()
try:
super(FixtureSuite, self).run(result)
finally:
self._fixture.cleanUp()
def sort_tests(self):
self._tests = sorted_tests(self, True)
def _flatten_tests(suite_or_case, unpack_outer=False):
try:
tests = iter(suite_or_case)
except TypeError:
# Not iterable, assume it's a test case.
return [(suite_or_case.id(), suite_or_case)]
if (type(suite_or_case) in (unittest.TestSuite,) or
unpack_outer):
# Plain old test suite (or any others we may add).
result = []
for test in tests:
# Recurse to flatten.
result.extend(_flatten_tests(test))
return result
else:
# Find any old actual test and grab its id.
suite_id = None
tests = iterate_tests(suite_or_case)
for test in tests:
suite_id = test.id()
break
# If it has a sort_tests method, call that.
if safe_hasattr(suite_or_case, 'sort_tests'):
suite_or_case.sort_tests()
return [(suite_id, suite_or_case)]
def sorted_tests(suite_or_case, unpack_outer=False):
"""Sort suite_or_case while preserving non-vanilla TestSuites."""
tests = _flatten_tests(suite_or_case, unpack_outer=unpack_outer)
tests.sort()
return unittest.TestSuite([test for (sort_key, test) in tests])
|
