lib/testtools/testtools/testsuite.py

   1 # Copyright (c) 2009-2011 testtools developers. See LICENSE for details.
   2
   3 """Test suites and related things."""
   4
   5 __metaclass__ = type
   6 __all__ = [
   7   'ConcurrentTestSuite',
   8   'iterate_tests',
   9   ]
  10
  11 from testtools.helpers import try_imports
  12
  13 Queue = try_imports(['Queue.Queue', 'queue.Queue'])
  14
  15 import threading
  16 import unittest
  17
  18 import testtools
  19
  20
  21 def iterate_tests(test_suite_or_case):
  22     """Iterate through all of the test cases in 'test_suite_or_case'."""
  23     try:
  24         suite = iter(test_suite_or_case)
  25     except TypeError:
  26         yield test_suite_or_case
  27     else:
  28         for test in suite:
  29             for subtest in iterate_tests(test):
  30                 yield subtest
  31
  32
  33 class ConcurrentTestSuite(unittest.TestSuite):
  34     """A TestSuite whose run() calls out to a concurrency strategy."""
  35
  36     def __init__(self, suite, make_tests):
  37         """Create a ConcurrentTestSuite to execute suite.
  38
  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)
  44             method.
  45         """
  46         super(ConcurrentTestSuite, self).__init__([suite])
  47         self.make_tests = make_tests
  48
  49     def run(self, result):
  50         """Run the tests concurrently.
  51
  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
  54         a time.
  55
  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.
  61         """
  62         tests = self.make_tests(self)
  63         try:
  64             threads = {}
  65             queue = Queue()
  66             result_semaphore = threading.Semaphore(1)
  67             for test in tests:
  68                 process_result = testtools.ThreadsafeForwardingResult(result,
  69                     result_semaphore)
  70                 reader_thread = threading.Thread(
  71                     target=self._run_test, args=(test, process_result, queue))
  72                 threads[test] = reader_thread, process_result
  73                 reader_thread.start()
  74             while threads:
  75                 finished_test = queue.get()
  76                 threads[finished_test][0].join()
  77                 del threads[finished_test]
  78         except:
  79             for thread, process_result in threads.values():
  80                 process_result.stop()
  81             raise
  82
  83     def _run_test(self, test, process_result, queue):
  84         try:
  85             test.run(process_result)
  86         finally:
  87             queue.put(test)
  88
  89
  90 class FixtureSuite(unittest.TestSuite):
  91
  92     def __init__(self, fixture, tests):
  93         super(FixtureSuite, self).__init__(tests)
  94         self._fixture = fixture
  95
  96     def run(self, result):
  97         self._fixture.setUp()
  98         try:
  99             super(FixtureSuite, self).run(result)
 100         finally:
 101             self._fixture.cleanUp()