835
840
self.assertContainsRe(output,
836
841
r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
843
def test_uses_time_from_testtools(self):
844
"""Test case timings in verbose results should use testtools times"""
846
class TimeAddedVerboseTestResult(tests.VerboseTestResult):
847
def startTest(self, test):
848
self.time(datetime.datetime.utcfromtimestamp(1.145))
849
super(TimeAddedVerboseTestResult, self).startTest(test)
850
def addSuccess(self, test):
851
self.time(datetime.datetime.utcfromtimestamp(51.147))
852
super(TimeAddedVerboseTestResult, self).addSuccess(test)
853
def report_tests_starting(self): pass
855
self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
856
self.assertEndsWith(sio.getvalue(), "OK 50002ms\n")
838
858
def test_known_failure(self):
839
859
"""A KnownFailure being raised should trigger several result actions."""
840
860
class InstrumentedTestResult(tests.ExtendedTestResult):
841
861
def stopTestRun(self): pass
842
def startTests(self): pass
843
def report_test_start(self, test): pass
862
def report_tests_starting(self): pass
844
863
def report_known_failure(self, test, err=None, details=None):
845
864
self._call = test, 'known failure'
846
865
result = InstrumentedTestResult(None, None, None, None)
1213
1242
self.assertContainsRe(output_string, "--date [0-9.]+")
1214
1243
self.assertLength(1, self._get_source_tree_calls)
1245
def test_verbose_test_count(self):
1246
"""A verbose test run reports the right test count at the start"""
1247
suite = TestUtil.TestSuite([
1248
unittest.FunctionTestCase(lambda:None),
1249
unittest.FunctionTestCase(lambda:None)])
1250
self.assertEqual(suite.countTestCases(), 2)
1252
runner = tests.TextTestRunner(stream=stream, verbosity=2)
1253
# Need to use the CountingDecorator as that's what sets num_tests
1254
result = self.run_test_runner(runner, tests.CountingDecorator(suite))
1255
self.assertStartsWith(stream.getvalue(), "running 2 tests")
1216
1257
def test_startTestRun(self):
1217
1258
"""run should call result.startTestRun()"""
1219
class LoggingDecorator(tests.ForwardingResult):
1260
class LoggingDecorator(ExtendedToOriginalDecorator):
1220
1261
def startTestRun(self):
1221
tests.ForwardingResult.startTestRun(self)
1262
ExtendedToOriginalDecorator.startTestRun(self)
1222
1263
calls.append('startTestRun')
1223
1264
test = unittest.FunctionTestCase(lambda:None)
1224
1265
stream = StringIO()
1655
1696
self.assertEqual('original', obj.test_attr)
1699
class _MissingFeature(tests.Feature):
1702
missing_feature = _MissingFeature()
1705
def _get_test(name):
1706
"""Get an instance of a specific example test.
1708
We protect this in a function so that they don't auto-run in the test
1712
class ExampleTests(tests.TestCase):
1714
def test_fail(self):
1715
mutter('this was a failing test')
1716
self.fail('this test will fail')
1718
def test_error(self):
1719
mutter('this test errored')
1720
raise RuntimeError('gotcha')
1722
def test_missing_feature(self):
1723
mutter('missing the feature')
1724
self.requireFeature(missing_feature)
1726
def test_skip(self):
1727
mutter('this test will be skipped')
1728
raise tests.TestSkipped('reason')
1730
def test_success(self):
1731
mutter('this test succeeds')
1733
def test_xfail(self):
1734
mutter('test with expected failure')
1735
self.knownFailure('this_fails')
1737
def test_unexpected_success(self):
1738
mutter('test with unexpected success')
1739
self.expectFailure('should_fail', lambda: None)
1741
return ExampleTests(name)
1744
class TestTestCaseLogDetails(tests.TestCase):
1746
def _run_test(self, test_name):
1747
test = _get_test(test_name)
1748
result = testtools.TestResult()
1752
def test_fail_has_log(self):
1753
result = self._run_test('test_fail')
1754
self.assertEqual(1, len(result.failures))
1755
result_content = result.failures[0][1]
1756
self.assertContainsRe(result_content, 'Text attachment: log')
1757
self.assertContainsRe(result_content, 'this was a failing test')
1759
def test_error_has_log(self):
1760
result = self._run_test('test_error')
1761
self.assertEqual(1, len(result.errors))
1762
result_content = result.errors[0][1]
1763
self.assertContainsRe(result_content, 'Text attachment: log')
1764
self.assertContainsRe(result_content, 'this test errored')
1766
def test_skip_has_no_log(self):
1767
result = self._run_test('test_skip')
1768
self.assertEqual(['reason'], result.skip_reasons.keys())
1769
skips = result.skip_reasons['reason']
1770
self.assertEqual(1, len(skips))
1772
self.assertFalse('log' in test.getDetails())
1774
def test_missing_feature_has_no_log(self):
1775
# testtools doesn't know about addNotSupported, so it just gets
1776
# considered as a skip
1777
result = self._run_test('test_missing_feature')
1778
self.assertEqual([missing_feature], result.skip_reasons.keys())
1779
skips = result.skip_reasons[missing_feature]
1780
self.assertEqual(1, len(skips))
1782
self.assertFalse('log' in test.getDetails())
1784
def test_xfail_has_no_log(self):
1785
result = self._run_test('test_xfail')
1786
self.assertEqual(1, len(result.expectedFailures))
1787
result_content = result.expectedFailures[0][1]
1788
self.assertNotContainsRe(result_content, 'Text attachment: log')
1789
self.assertNotContainsRe(result_content, 'test with expected failure')
1791
def test_unexpected_success_has_log(self):
1792
result = self._run_test('test_unexpected_success')
1793
self.assertEqual(1, len(result.unexpectedSuccesses))
1794
# Inconsistency, unexpectedSuccesses is a list of tests,
1795
# expectedFailures is a list of reasons?
1796
test = result.unexpectedSuccesses[0]
1797
details = test.getDetails()
1798
self.assertTrue('log' in details)
1801
class TestTestCloning(tests.TestCase):
1802
"""Tests that test cloning of TestCases (as used by multiply_tests)."""
1804
def test_cloned_testcase_does_not_share_details(self):
1805
"""A TestCase cloned with clone_test does not share mutable attributes
1806
such as details or cleanups.
1808
class Test(tests.TestCase):
1810
self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1811
orig_test = Test('test_foo')
1812
cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1813
orig_test.run(unittest.TestResult())
1814
self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1815
self.assertEqual(None, cloned_test.getDetails().get('foo'))
1817
def test_double_apply_scenario_preserves_first_scenario(self):
1818
"""Applying two levels of scenarios to a test preserves the attributes
1819
added by both scenarios.
1821
class Test(tests.TestCase):
1824
test = Test('test_foo')
1825
scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1826
scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1827
suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1828
suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1829
all_tests = list(tests.iter_suite_tests(suite))
1830
self.assertLength(4, all_tests)
1831
all_xys = sorted((t.x, t.y) for t in all_tests)
1832
self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1658
1835
# NB: Don't delete this; it's not actually from 0.11!
1659
1836
@deprecated_function(deprecated_in((0, 11, 0)))
1660
1837
def sample_deprecated_function():
1971
2148
load_list='missing file name', list_only=True)
2151
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2153
_test_needs_features = [features.subunit]
2155
def run_subunit_stream(self, test_name):
2156
from subunit import ProtocolTestCase
2158
return TestUtil.TestSuite([_get_test(test_name)])
2159
stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2160
test_suite_factory=factory)
2161
test = ProtocolTestCase(stream)
2162
result = testtools.TestResult()
2164
content = stream.getvalue()
2165
return content, result
2167
def test_fail_has_log(self):
2168
content, result = self.run_subunit_stream('test_fail')
2169
self.assertEqual(1, len(result.failures))
2170
self.assertContainsRe(content, '(?m)^log$')
2171
self.assertContainsRe(content, 'this test will fail')
2173
def test_error_has_log(self):
2174
content, result = self.run_subunit_stream('test_error')
2175
self.assertContainsRe(content, '(?m)^log$')
2176
self.assertContainsRe(content, 'this test errored')
2178
def test_skip_has_no_log(self):
2179
content, result = self.run_subunit_stream('test_skip')
2180
self.assertNotContainsRe(content, '(?m)^log$')
2181
self.assertNotContainsRe(content, 'this test will be skipped')
2182
self.assertEqual(['reason'], result.skip_reasons.keys())
2183
skips = result.skip_reasons['reason']
2184
self.assertEqual(1, len(skips))
2186
# RemotedTestCase doesn't preserve the "details"
2187
## self.assertFalse('log' in test.getDetails())
2189
def test_missing_feature_has_no_log(self):
2190
content, result = self.run_subunit_stream('test_missing_feature')
2191
self.assertNotContainsRe(content, '(?m)^log$')
2192
self.assertNotContainsRe(content, 'missing the feature')
2193
self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
2194
skips = result.skip_reasons['_MissingFeature\n']
2195
self.assertEqual(1, len(skips))
2197
# RemotedTestCase doesn't preserve the "details"
2198
## self.assertFalse('log' in test.getDetails())
2200
def test_xfail_has_no_log(self):
2201
content, result = self.run_subunit_stream('test_xfail')
2202
self.assertNotContainsRe(content, '(?m)^log$')
2203
self.assertNotContainsRe(content, 'test with expected failure')
2204
self.assertEqual(1, len(result.expectedFailures))
2205
result_content = result.expectedFailures[0][1]
2206
self.assertNotContainsRe(result_content, 'Text attachment: log')
2207
self.assertNotContainsRe(result_content, 'test with expected failure')
2209
def test_unexpected_success_has_log(self):
2210
content, result = self.run_subunit_stream('test_unexpected_success')
2211
self.assertContainsRe(content, '(?m)^log$')
2212
self.assertContainsRe(content, 'test with unexpected success')
2213
self.expectFailure('subunit treats "unexpectedSuccess"'
2214
' as a plain success',
2215
self.assertEqual, 1, len(result.unexpectedSuccesses))
2216
self.assertEqual(1, len(result.unexpectedSuccesses))
2217
test = result.unexpectedSuccesses[0]
2218
# RemotedTestCase doesn't preserve the "details"
2219
## self.assertTrue('log' in test.getDetails())
2221
def test_success_has_no_log(self):
2222
content, result = self.run_subunit_stream('test_success')
2223
self.assertEqual(1, result.testsRun)
2224
self.assertNotContainsRe(content, '(?m)^log$')
2225
self.assertNotContainsRe(content, 'this test succeeds')
1974
2228
class TestRunBzr(tests.TestCase):
2960
3219
self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3222
class TestThreadLeakDetection(tests.TestCase):
3223
"""Ensure when tests leak threads we detect and report it"""
3225
class LeakRecordingResult(tests.ExtendedTestResult):
3227
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3229
def _report_thread_leak(self, test, leaks, alive):
3230
self.leaks.append((test, leaks))
3232
def test_testcase_without_addCleanups(self):
3233
"""Check old TestCase instances don't break with leak detection"""
3234
class Test(unittest.TestCase):
3237
addCleanup = None # for when on Python 2.7 with native addCleanup
3238
result = self.LeakRecordingResult()
3240
self.assertIs(getattr(test, "addCleanup", None), None)
3241
result.startTestRun()
3243
result.stopTestRun()
3244
self.assertEqual(result._tests_leaking_threads_count, 0)
3245
self.assertEqual(result.leaks, [])
3247
def test_thread_leak(self):
3248
"""Ensure a thread that outlives the running of a test is reported
3250
Uses a thread that blocks on an event, and is started by the inner
3251
test case. As the thread outlives the inner case's run, it should be
3252
detected as a leak, but the event is then set so that the thread can
3253
be safely joined in cleanup so it's not leaked for real.
3255
event = threading.Event()
3256
thread = threading.Thread(name="Leaker", target=event.wait)
3257
class Test(tests.TestCase):
3258
def test_leak(self):
3260
result = self.LeakRecordingResult()
3261
test = Test("test_leak")
3262
self.addCleanup(thread.join)
3263
self.addCleanup(event.set)
3264
result.startTestRun()
3266
result.stopTestRun()
3267
self.assertEqual(result._tests_leaking_threads_count, 1)
3268
self.assertEqual(result._first_thread_leaker_id, test.id())
3269
self.assertEqual(result.leaks, [(test, set([thread]))])
3270
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3272
def test_multiple_leaks(self):
3273
"""Check multiple leaks are blamed on the test cases at fault
3275
Same concept as the previous test, but has one inner test method that
3276
leaks two threads, and one that doesn't leak at all.
3278
event = threading.Event()
3279
thread_a = threading.Thread(name="LeakerA", target=event.wait)
3280
thread_b = threading.Thread(name="LeakerB", target=event.wait)
3281
thread_c = threading.Thread(name="LeakerC", target=event.wait)
3282
class Test(tests.TestCase):
3283
def test_first_leak(self):
3285
def test_second_no_leak(self):
3287
def test_third_leak(self):
3290
result = self.LeakRecordingResult()
3291
first_test = Test("test_first_leak")
3292
third_test = Test("test_third_leak")
3293
self.addCleanup(thread_a.join)
3294
self.addCleanup(thread_b.join)
3295
self.addCleanup(thread_c.join)
3296
self.addCleanup(event.set)
3297
result.startTestRun()
3299
[first_test, Test("test_second_no_leak"), third_test]
3301
result.stopTestRun()
3302
self.assertEqual(result._tests_leaking_threads_count, 2)
3303
self.assertEqual(result._first_thread_leaker_id, first_test.id())
3304
self.assertEqual(result.leaks, [
3305
(first_test, set([thread_b])),
3306
(third_test, set([thread_a, thread_c]))])
3307
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3310
class TestPostMortemDebugging(tests.TestCase):
3311
"""Check post mortem debugging works when tests fail or error"""
3313
class TracebackRecordingResult(tests.ExtendedTestResult):
3315
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3316
self.postcode = None
3317
def _post_mortem(self, tb=None):
3318
"""Record the code object at the end of the current traceback"""
3319
tb = tb or sys.exc_info()[2]
3322
while next is not None:
3325
self.postcode = tb.tb_frame.f_code
3326
def report_error(self, test, err):
3328
def report_failure(self, test, err):
3331
def test_location_unittest_error(self):
3332
"""Needs right post mortem traceback with erroring unittest case"""
3333
class Test(unittest.TestCase):
3336
result = self.TracebackRecordingResult()
3338
self.assertEqual(result.postcode, Test.runTest.func_code)
3340
def test_location_unittest_failure(self):
3341
"""Needs right post mortem traceback with failing unittest case"""
3342
class Test(unittest.TestCase):
3344
raise self.failureException
3345
result = self.TracebackRecordingResult()
3347
self.assertEqual(result.postcode, Test.runTest.func_code)
3349
def test_location_bt_error(self):
3350
"""Needs right post mortem traceback with erroring bzrlib.tests case"""
3351
class Test(tests.TestCase):
3352
def test_error(self):
3354
result = self.TracebackRecordingResult()
3355
Test("test_error").run(result)
3356
self.assertEqual(result.postcode, Test.test_error.func_code)
3358
def test_location_bt_failure(self):
3359
"""Needs right post mortem traceback with failing bzrlib.tests case"""
3360
class Test(tests.TestCase):
3361
def test_failure(self):
3362
raise self.failureException
3363
result = self.TracebackRecordingResult()
3364
Test("test_failure").run(result)
3365
self.assertEqual(result.postcode, Test.test_failure.func_code)
3367
def test_env_var_triggers_post_mortem(self):
3368
"""Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
3370
result = tests.ExtendedTestResult(StringIO(), 0, 1)
3371
post_mortem_calls = []
3372
self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3373
self.addCleanup(osutils.set_or_unset_env, "BZR_TEST_PDB",
3374
osutils.set_or_unset_env("BZR_TEST_PDB", None))
3375
result._post_mortem(1)
3376
os.environ["BZR_TEST_PDB"] = "on"
3377
result._post_mortem(2)
3378
self.assertEqual([2], post_mortem_calls)
2963
3381
class TestRunSuite(tests.TestCase):
2965
3383
def test_runner_class(self):