← Back to branch summary

~testing-cabal/ubuntu/hardy/python-testtools/hardy-tweaks

« back to all changes in this revision

Viewing changes to testtools/testsuite.py

Add a test suite and a test result to support concurrent testing.

Thanks to Robert Collins.

Show diffs side-by-side

added added

removed removed

Lines of Context:
testtools/testsuite.py
 
1
# Copyright (c) 2009 Jonathan M. Lange. See LICENSE for details.
 
2
 
 
3
"""Test suites and related things."""
 
4
 
 
5
__metaclass__ = type
 
6
__all__ = [
 
7
  'ConcurrentTestSuite',
 
8
  ]
 
9
 
 
10
import Queue
 
11
import threading
 
12
import unittest
 
13
 
 
14
import testtools
 
15
 
 
16
 
 
17
class ConcurrentTestSuite(unittest.TestSuite):
 
18
    """A TestSuite whose run() calls out to a concurrency strategy.""" 
 
19
 
 
20
    def __init__(self, suite, make_tests):
 
21
        """Create a ConcurrentTestSuite to execute suite.
 
22
        
 
23
        :param suite: A suite to run concurrently.
 
24
        :param make_tests: A helper function to split the tests in the
 
25
            ConcurrentTestSuite into some number of concurrently executing
 
26
            sub-suites. make_tests must take a suite, and return an iterable
 
27
            of TestCase-like object, each of which must have a run(result)
 
28
            method.
 
29
        """
 
30
        super(ConcurrentTestSuite, self).__init__([suite])
 
31
        self.make_tests = make_tests
 
32
 
 
33
    def run(self, result):
 
34
        """Run the tests concurrently.
 
35
 
 
36
        This calls out to the provided make_tests helper, and then serialises
 
37
        the results so that result only sees activity from one TestCase at
 
38
        a time.
 
39
 
 
40
        ConcurrentTestSuite provides no special mechanism to stop the tests
 
41
        returned by make_tests, it is up to the make_tests to honour the
 
42
        shouldStop attribute on the result object they are run with, which will
 
43
        be set if an exception is raised in the thread which
 
44
        ConcurrentTestSuite.run is called in.
 
45
        """
 
46
        tests = self.make_tests(self)
 
47
        try:
 
48
            threads = {}
 
49
            queue = Queue.Queue()
 
50
            result_semaphore = threading.Semaphore(1)
 
51
            for test in tests:
 
52
                process_result = testtools.ThreadsafeForwardingResult(result,
 
53
                    result_semaphore)
 
54
                reader_thread = threading.Thread(target=self._run_test, args=(test,
 
55
                    process_result, queue))
 
56
                threads[test] = reader_thread, process_result
 
57
                reader_thread.start()
 
58
            while threads:
 
59
                finished_test = queue.get()
 
60
                threads[finished_test][0].join()
 
61
                del threads[finished_test]
 
62
        except:
 
63
            for thread, process_result in threads.itervalues():
 
64
                process_result.stop()
 
65
            raise
 
66
        
 
67
    def _run_test(self, test, process_result, queue):
 
68
        try:
 
69
            test.run(process_result)
 
70
        finally:
 
71
            queue.put(test)
 
72
 
 
73