s4:dsdb/drepl: update the source_dsa_obj/invocation_id in repsFrom
[Samba/gebeck_regimport.git] / lib / testtools / testtools / tests / test_testsuite.py
blob3fc837c701d3c28adbc48beb94d1f3a59cc213f1
1 # Copyright (c) 2009-2011 testtools developers. See LICENSE for details.
3 """Test ConcurrentTestSuite and related things."""
5 __metaclass__ = type
7 import unittest
9 from testtools import (
10 ConcurrentTestSuite,
11 iterate_tests,
12 PlaceHolder,
13 TestCase,
15 from testtools.helpers import try_import
16 from testtools.testsuite import FixtureSuite, iterate_tests, sorted_tests
17 from testtools.tests.helpers import LoggingResult
19 FunctionFixture = try_import('fixtures.FunctionFixture')
21 class Sample(TestCase):
22 def __hash__(self):
23 return id(self)
24 def test_method1(self):
25 pass
26 def test_method2(self):
27 pass
29 class TestConcurrentTestSuiteRun(TestCase):
31 def test_trivial(self):
32 log = []
33 result = LoggingResult(log)
34 test1 = Sample('test_method1')
35 test2 = Sample('test_method2')
36 original_suite = unittest.TestSuite([test1, test2])
37 suite = ConcurrentTestSuite(original_suite, self.split_suite)
38 suite.run(result)
39 # log[0] is the timestamp for the first test starting.
40 test1 = log[1][1]
41 test2 = log[-1][1]
42 self.assertIsInstance(test1, Sample)
43 self.assertIsInstance(test2, Sample)
44 self.assertNotEqual(test1.id(), test2.id())
46 def test_wrap_result(self):
47 # ConcurrentTestSuite has a hook for wrapping the per-thread result.
48 wrap_log = []
50 def wrap_result(thread_safe_result, thread_number):
51 wrap_log.append(
52 (thread_safe_result.result.decorated, thread_number))
53 return thread_safe_result
55 result_log = []
56 result = LoggingResult(result_log)
57 test1 = Sample('test_method1')
58 test2 = Sample('test_method2')
59 original_suite = unittest.TestSuite([test1, test2])
60 suite = ConcurrentTestSuite(
61 original_suite, self.split_suite, wrap_result=wrap_result)
62 suite.run(result)
63 self.assertEqual(
64 [(result, 0),
65 (result, 1),
66 ], wrap_log)
67 # Smoke test to make sure everything ran OK.
68 self.assertNotEqual([], result_log)
70 def split_suite(self, suite):
71 tests = list(iterate_tests(suite))
72 return tests[0], tests[1]
75 class TestFixtureSuite(TestCase):
77 def setUp(self):
78 super(TestFixtureSuite, self).setUp()
79 if FunctionFixture is None:
80 self.skip("Need fixtures")
82 def test_fixture_suite(self):
83 log = []
84 class Sample(TestCase):
85 def test_one(self):
86 log.append(1)
87 def test_two(self):
88 log.append(2)
89 fixture = FunctionFixture(
90 lambda: log.append('setUp'),
91 lambda fixture: log.append('tearDown'))
92 suite = FixtureSuite(fixture, [Sample('test_one'), Sample('test_two')])
93 suite.run(LoggingResult([]))
94 self.assertEqual(['setUp', 1, 2, 'tearDown'], log)
97 class TestSortedTests(TestCase):
99 def test_sorts_custom_suites(self):
100 a = PlaceHolder('a')
101 b = PlaceHolder('b')
102 class Subclass(unittest.TestSuite):
103 def sort_tests(self):
104 self._tests = sorted_tests(self, True)
105 input_suite = Subclass([b, a])
106 suite = sorted_tests(input_suite)
107 self.assertEqual([a, b], list(iterate_tests(suite)))
108 self.assertEqual([input_suite], list(iter(suite)))
110 def test_custom_suite_without_sort_tests_works(self):
111 a = PlaceHolder('a')
112 b = PlaceHolder('b')
113 class Subclass(unittest.TestSuite):pass
114 input_suite = Subclass([b, a])
115 suite = sorted_tests(input_suite)
116 self.assertEqual([b, a], list(iterate_tests(suite)))
117 self.assertEqual([input_suite], list(iter(suite)))
119 def test_sorts_simple_suites(self):
120 a = PlaceHolder('a')
121 b = PlaceHolder('b')
122 suite = sorted_tests(unittest.TestSuite([b, a]))
123 self.assertEqual([a, b], list(iterate_tests(suite)))
126 def test_suite():
127 from unittest import TestLoader
128 return TestLoader().loadTestsFromName(__name__)