1 # Copyright (c) 2009-2011 testtools developers. See LICENSE for details.
3 """Test suites and related things."""
11 from testtools
.helpers
import try_imports
13 Queue
= try_imports(['Queue.Queue', 'queue.Queue'])
21 def iterate_tests(test_suite_or_case
):
22 """Iterate through all of the test cases in 'test_suite_or_case'."""
24 suite
= iter(test_suite_or_case
)
26 yield test_suite_or_case
29 for subtest
in iterate_tests(test
):
33 class ConcurrentTestSuite(unittest
.TestSuite
):
34 """A TestSuite whose run() calls out to a concurrency strategy."""
36 def __init__(self
, suite
, make_tests
):
37 """Create a ConcurrentTestSuite to execute suite.
39 :param suite: A suite to run concurrently.
40 :param make_tests: A helper function to split the tests in the
41 ConcurrentTestSuite into some number of concurrently executing
42 sub-suites. make_tests must take a suite, and return an iterable
43 of TestCase-like object, each of which must have a run(result)
46 super(ConcurrentTestSuite
, self
).__init
__([suite
])
47 self
.make_tests
= make_tests
49 def run(self
, result
):
50 """Run the tests concurrently.
52 This calls out to the provided make_tests helper, and then serialises
53 the results so that result only sees activity from one TestCase at
56 ConcurrentTestSuite provides no special mechanism to stop the tests
57 returned by make_tests, it is up to the make_tests to honour the
58 shouldStop attribute on the result object they are run with, which will
59 be set if an exception is raised in the thread which
60 ConcurrentTestSuite.run is called in.
62 tests
= self
.make_tests(self
)
66 result_semaphore
= threading
.Semaphore(1)
68 process_result
= testtools
.ThreadsafeForwardingResult(result
,
70 reader_thread
= threading
.Thread(
71 target
=self
._run
_test
, args
=(test
, process_result
, queue
))
72 threads
[test
] = reader_thread
, process_result
75 finished_test
= queue
.get()
76 threads
[finished_test
][0].join()
77 del threads
[finished_test
]
79 for thread
, process_result
in threads
.values():
83 def _run_test(self
, test
, process_result
, queue
):
85 test
.run(process_result
)
90 class FixtureSuite(unittest
.TestSuite
):
92 def __init__(self
, fixture
, tests
):
93 super(FixtureSuite
, self
).__init
__(tests
)
94 self
._fixture
= fixture
96 def run(self
, result
):
99 super(FixtureSuite
, self
).run(result
)
101 self
._fixture
.cleanUp()