# Copyright (c) 2009-2015 testtools developers. See LICENSE for details. """Test ConcurrentTestSuite and related things.""" import doctest from pprint import pformat import unittest import unittest2 from extras import try_import from testtools import ( ConcurrentTestSuite, ConcurrentStreamTestSuite, iterate_tests, PlaceHolder, TestByTestResult, TestCase, ) from testtools.compat import _u from testtools.matchers import DocTestMatches, Equals from testtools.testsuite import FixtureSuite, sorted_tests from testtools.tests.helpers import LoggingResult from testtools.testresult.doubles import StreamResult as LoggingStream FunctionFixture = try_import('fixtures.FunctionFixture') class Sample(TestCase): def __hash__(self): return id(self) def test_method1(self): pass def test_method2(self): pass class TestConcurrentTestSuiteRun(TestCase): def test_broken_test(self): log = [] def on_test(test, status, start_time, stop_time, tags, details): log.append((test.id(), status, set(details.keys()))) class BrokenTest(object): # Simple break - no result parameter to run() def __call__(self): pass run = __call__ original_suite = unittest.TestSuite([BrokenTest()]) suite = ConcurrentTestSuite(original_suite, self.split_suite) suite.run(TestByTestResult(on_test)) self.assertEqual([('broken-runner', 'error', set(['traceback']))], log) def test_trivial(self): log = [] result = LoggingResult(log) test1 = Sample('test_method1') test2 = Sample('test_method2') original_suite = unittest.TestSuite([test1, test2]) suite = ConcurrentTestSuite(original_suite, self.split_suite) suite.run(result) # log[0] is the timestamp for the first test starting. test1 = log[1][1] test2 = log[-1][1] self.assertIsInstance(test1, Sample) self.assertIsInstance(test2, Sample) self.assertNotEqual(test1.id(), test2.id()) def test_wrap_result(self): # ConcurrentTestSuite has a hook for wrapping the per-thread result. wrap_log = [] def wrap_result(thread_safe_result, thread_number): wrap_log.append( (thread_safe_result.result.decorated, thread_number)) return thread_safe_result result_log = [] result = LoggingResult(result_log) test1 = Sample('test_method1') test2 = Sample('test_method2') original_suite = unittest.TestSuite([test1, test2]) suite = ConcurrentTestSuite( original_suite, self.split_suite, wrap_result=wrap_result) suite.run(result) self.assertEqual( [(result, 0), (result, 1), ], wrap_log) # Smoke test to make sure everything ran OK. self.assertNotEqual([], result_log) def split_suite(self, suite): return list(iterate_tests(suite)) class TestConcurrentStreamTestSuiteRun(TestCase): def test_trivial(self): result = LoggingStream() test1 = Sample('test_method1') test2 = Sample('test_method2') cases = lambda:[(test1, '0'), (test2, '1')] suite = ConcurrentStreamTestSuite(cases) suite.run(result) def freeze(set_or_none): if set_or_none is None: return set_or_none return frozenset(set_or_none) # Ignore event order: we're testing the code is all glued together, # which just means we can pump events through and they get route codes # added appropriately. self.assertEqual(set([ ('status', 'testtools.tests.test_testsuite.Sample.test_method1', 'inprogress', None, True, None, None, False, None, '0', None, ), ('status', 'testtools.tests.test_testsuite.Sample.test_method1', 'success', frozenset(), True, None, None, False, None, '0', None, ), ('status', 'testtools.tests.test_testsuite.Sample.test_method2', 'inprogress', None, True, None, None, False, None, '1', None, ), ('status', 'testtools.tests.test_testsuite.Sample.test_method2', 'success', frozenset(), True, None, None, False, None, '1', None, ), ]), set(event[0:3] + (freeze(event[3]),) + event[4:10] + (None,) for event in result._events)) def test_broken_runner(self): # If the object called breaks, the stream is informed about it # regardless. class BrokenTest(object): # broken - no result parameter! def __call__(self): pass def run(self): pass result = LoggingStream() cases = lambda:[(BrokenTest(), '0')] suite = ConcurrentStreamTestSuite(cases) suite.run(result) events = result._events # Check the traceback loosely. self.assertEqual(events[1][6].decode('utf8'), "Traceback (most recent call last):\n") self.assertThat(events[2][6].decode('utf8'), DocTestMatches("""\ File "...testtools/testsuite.py", line ..., in _run_test test.run(process_result) """, doctest.ELLIPSIS)) self.assertThat(events[3][6].decode('utf8'), DocTestMatches("""\ TypeError: run() takes ...1 ...argument...2...given... """, doctest.ELLIPSIS)) events = [event[0:10] + (None,) for event in events] events[1] = events[1][:6] + (None,) + events[1][7:] events[2] = events[2][:6] + (None,) + events[2][7:] events[3] = events[3][:6] + (None,) + events[3][7:] self.assertEqual([ ('status', "broken-runner-'0'", 'inprogress', None, True, None, None, False, None, _u('0'), None), ('status', "broken-runner-'0'", None, None, True, 'traceback', None, False, 'text/x-traceback; charset="utf8"; language="python"', '0', None), ('status', "broken-runner-'0'", None, None, True, 'traceback', None, False, 'text/x-traceback; charset="utf8"; language="python"', '0', None), ('status', "broken-runner-'0'", None, None, True, 'traceback', None, True, 'text/x-traceback; charset="utf8"; language="python"', '0', None), ('status', "broken-runner-'0'", 'fail', set(), True, None, None, False, None, _u('0'), None) ], events) def split_suite(self, suite): tests = list(enumerate(iterate_tests(suite))) return [(test, _u(str(pos))) for pos, test in tests] def test_setupclass_skip(self): # We should support setupclass skipping using cls.skipException. # Because folk have used that. class Skips(TestCase): @classmethod def setUpClass(cls): raise cls.skipException('foo') def test_notrun(self): pass # Test discovery uses the default suite from unittest2 (unless users # deliberately change things, in which case they keep both pieces). suite = unittest2.TestSuite([Skips("test_notrun")]) log = [] result = LoggingResult(log) suite.run(result) self.assertEqual(['addSkip'], [item[0] for item in log]) def test_setupclass_upcall(self): # Note that this is kindof-a-case-test, kindof-suite, because # setUpClass is linked between them. class Simples(TestCase): @classmethod def setUpClass(cls): super(Simples, cls).setUpClass() def test_simple(self): pass # Test discovery uses the default suite from unittest2 (unless users # deliberately change things, in which case they keep both pieces). suite = unittest2.TestSuite([Simples("test_simple")]) log = [] result = LoggingResult(log) suite.run(result) self.assertEqual( ['startTest', 'addSuccess', 'stopTest'], [item[0] for item in log]) class TestFixtureSuite(TestCase): def setUp(self): super(TestFixtureSuite, self).setUp() if FunctionFixture is None: self.skip("Need fixtures") def test_fixture_suite(self): log = [] class Sample(TestCase): def test_one(self): log.append(1) def test_two(self): log.append(2) fixture = FunctionFixture( lambda: log.append('setUp'), lambda fixture: log.append('tearDown')) suite = FixtureSuite(fixture, [Sample('test_one'), Sample('test_two')]) suite.run(LoggingResult([])) self.assertEqual(['setUp', 1, 2, 'tearDown'], log) def test_fixture_suite_sort(self): log = [] class Sample(TestCase): def test_one(self): log.append(1) def test_two(self): log.append(2) fixture = FunctionFixture( lambda: log.append('setUp'), lambda fixture: log.append('tearDown')) suite = FixtureSuite(fixture, [Sample('test_one'), Sample('test_one')]) self.assertRaises(ValueError, suite.sort_tests) class TestSortedTests(TestCase): def test_sorts_custom_suites(self): a = PlaceHolder('a') b = PlaceHolder('b') class Subclass(unittest.TestSuite): def sort_tests(self): self._tests = sorted_tests(self, True) input_suite = Subclass([b, a]) suite = sorted_tests(input_suite) self.assertEqual([a, b], list(iterate_tests(suite))) self.assertEqual([input_suite], list(iter(suite))) def test_custom_suite_without_sort_tests_works(self): a = PlaceHolder('a') b = PlaceHolder('b') class Subclass(unittest.TestSuite):pass input_suite = Subclass([b, a]) suite = sorted_tests(input_suite) self.assertEqual([b, a], list(iterate_tests(suite))) self.assertEqual([input_suite], list(iter(suite))) def test_sorts_simple_suites(self): a = PlaceHolder('a') b = PlaceHolder('b') suite = sorted_tests(unittest.TestSuite([b, a])) self.assertEqual([a, b], list(iterate_tests(suite))) def test_duplicate_simple_suites(self): a = PlaceHolder('a') b = PlaceHolder('b') c = PlaceHolder('a') self.assertRaises( ValueError, sorted_tests, unittest.TestSuite([a, b, c])) def test_multiple_duplicates(self): # If there are multiple duplicates on a test suite, we report on them # all. a = PlaceHolder('a') b = PlaceHolder('b') c = PlaceHolder('a') d = PlaceHolder('b') error = self.assertRaises( ValueError, sorted_tests, unittest.TestSuite([a, b, c, d])) self.assertThat( str(error), Equals("Duplicate test ids detected: %s" % ( pformat({'a': 2, 'b': 2}),))) def test_suite(): from unittest import TestLoader return TestLoader().loadTestsFromName(__name__)