602
620
def test_dangling_locks_cause_failures(self):
603
621
class TestDanglingLock(tests.TestCaseWithMemoryTransport):
604
622
def test_function(self):
605
t = self.get_transport('.')
623
t = self.get_transport_from_path('.')
606
624
l = lockdir.LockDir(t, 'lock')
609
627
test = TestDanglingLock('test_function')
610
628
result = test.run()
629
total_failures = result.errors + result.failures
611
630
if self._lock_check_thorough:
612
self.assertEqual(1, len(result.errors))
631
self.assertEqual(1, len(total_failures))
614
633
# When _lock_check_thorough is disabled, then we don't trigger a
616
self.assertEqual(0, len(result.errors))
635
self.assertEqual(0, len(total_failures))
619
638
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
620
639
"""Tests for the convenience functions TestCaseWithTransport introduces."""
622
641
def test_get_readonly_url_none(self):
623
from bzrlib.transport import get_transport
624
from bzrlib.transport.memory import MemoryServer
625
642
from bzrlib.transport.readonly import ReadonlyTransportDecorator
626
self.vfs_transport_factory = MemoryServer
643
self.vfs_transport_factory = memory.MemoryServer
627
644
self.transport_readonly_server = None
628
645
# calling get_readonly_transport() constructs a decorator on the url
630
647
url = self.get_readonly_url()
631
648
url2 = self.get_readonly_url('foo/bar')
632
t = get_transport(url)
633
t2 = get_transport(url2)
634
self.failUnless(isinstance(t, ReadonlyTransportDecorator))
635
self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
649
t = transport.get_transport_from_url(url)
650
t2 = transport.get_transport_from_url(url2)
651
self.assertIsInstance(t, ReadonlyTransportDecorator)
652
self.assertIsInstance(t2, ReadonlyTransportDecorator)
636
653
self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
638
655
def test_get_readonly_url_http(self):
639
656
from bzrlib.tests.http_server import HttpServer
640
from bzrlib.transport import get_transport
641
from bzrlib.transport.local import LocalURLServer
642
657
from bzrlib.transport.http import HttpTransportBase
643
self.transport_server = LocalURLServer
658
self.transport_server = test_server.LocalURLServer
644
659
self.transport_readonly_server = HttpServer
645
660
# calling get_readonly_transport() gives us a HTTP server instance.
646
661
url = self.get_readonly_url()
647
662
url2 = self.get_readonly_url('foo/bar')
648
663
# the transport returned may be any HttpTransportBase subclass
649
t = get_transport(url)
650
t2 = get_transport(url2)
651
self.failUnless(isinstance(t, HttpTransportBase))
652
self.failUnless(isinstance(t2, HttpTransportBase))
664
t = transport.get_transport_from_url(url)
665
t2 = transport.get_transport_from_url(url2)
666
self.assertIsInstance(t, HttpTransportBase)
667
self.assertIsInstance(t2, HttpTransportBase)
653
668
self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
655
670
def test_is_directory(self):
1634
1624
self.assertRaises(AssertionError,
1635
1625
self.assertListRaises, _TestException, success_generator)
1627
def test_overrideAttr_without_value(self):
1628
self.test_attr = 'original' # Define a test attribute
1629
obj = self # Make 'obj' visible to the embedded test
1630
class Test(tests.TestCase):
1633
tests.TestCase.setUp(self)
1634
self.orig = self.overrideAttr(obj, 'test_attr')
1636
def test_value(self):
1637
self.assertEqual('original', self.orig)
1638
self.assertEqual('original', obj.test_attr)
1639
obj.test_attr = 'modified'
1640
self.assertEqual('modified', obj.test_attr)
1642
test = Test('test_value')
1643
test.run(unittest.TestResult())
1644
self.assertEqual('original', obj.test_attr)
1646
def test_overrideAttr_with_value(self):
1647
self.test_attr = 'original' # Define a test attribute
1648
obj = self # Make 'obj' visible to the embedded test
1649
class Test(tests.TestCase):
1652
tests.TestCase.setUp(self)
1653
self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1655
def test_value(self):
1656
self.assertEqual('original', self.orig)
1657
self.assertEqual('modified', obj.test_attr)
1659
test = Test('test_value')
1660
test.run(unittest.TestResult())
1661
self.assertEqual('original', obj.test_attr)
1663
def test_recordCalls(self):
1664
from bzrlib.tests import test_selftest
1665
calls = self.recordCalls(
1666
test_selftest, '_add_numbers')
1667
self.assertEqual(test_selftest._add_numbers(2, 10),
1669
self.assertEquals(calls, [((2, 10), {})])
1672
def _add_numbers(a, b):
1676
class _MissingFeature(features.Feature):
1679
missing_feature = _MissingFeature()
1682
def _get_test(name):
1683
"""Get an instance of a specific example test.
1685
We protect this in a function so that they don't auto-run in the test
1689
class ExampleTests(tests.TestCase):
1691
def test_fail(self):
1692
mutter('this was a failing test')
1693
self.fail('this test will fail')
1695
def test_error(self):
1696
mutter('this test errored')
1697
raise RuntimeError('gotcha')
1699
def test_missing_feature(self):
1700
mutter('missing the feature')
1701
self.requireFeature(missing_feature)
1703
def test_skip(self):
1704
mutter('this test will be skipped')
1705
raise tests.TestSkipped('reason')
1707
def test_success(self):
1708
mutter('this test succeeds')
1710
def test_xfail(self):
1711
mutter('test with expected failure')
1712
self.knownFailure('this_fails')
1714
def test_unexpected_success(self):
1715
mutter('test with unexpected success')
1716
self.expectFailure('should_fail', lambda: None)
1718
return ExampleTests(name)
1721
class TestTestCaseLogDetails(tests.TestCase):
1723
def _run_test(self, test_name):
1724
test = _get_test(test_name)
1725
result = testtools.TestResult()
1729
def test_fail_has_log(self):
1730
result = self._run_test('test_fail')
1731
self.assertEqual(1, len(result.failures))
1732
result_content = result.failures[0][1]
1733
self.assertContainsRe(result_content,
1734
'(?m)^(?:Text attachment: )?log(?:$|: )')
1735
self.assertContainsRe(result_content, 'this was a failing test')
1737
def test_error_has_log(self):
1738
result = self._run_test('test_error')
1739
self.assertEqual(1, len(result.errors))
1740
result_content = result.errors[0][1]
1741
self.assertContainsRe(result_content,
1742
'(?m)^(?:Text attachment: )?log(?:$|: )')
1743
self.assertContainsRe(result_content, 'this test errored')
1745
def test_skip_has_no_log(self):
1746
result = self._run_test('test_skip')
1747
self.assertEqual(['reason'], result.skip_reasons.keys())
1748
skips = result.skip_reasons['reason']
1749
self.assertEqual(1, len(skips))
1751
self.assertFalse('log' in test.getDetails())
1753
def test_missing_feature_has_no_log(self):
1754
# testtools doesn't know about addNotSupported, so it just gets
1755
# considered as a skip
1756
result = self._run_test('test_missing_feature')
1757
self.assertEqual([missing_feature], result.skip_reasons.keys())
1758
skips = result.skip_reasons[missing_feature]
1759
self.assertEqual(1, len(skips))
1761
self.assertFalse('log' in test.getDetails())
1763
def test_xfail_has_no_log(self):
1764
result = self._run_test('test_xfail')
1765
self.assertEqual(1, len(result.expectedFailures))
1766
result_content = result.expectedFailures[0][1]
1767
self.assertNotContainsRe(result_content,
1768
'(?m)^(?:Text attachment: )?log(?:$|: )')
1769
self.assertNotContainsRe(result_content, 'test with expected failure')
1771
def test_unexpected_success_has_log(self):
1772
result = self._run_test('test_unexpected_success')
1773
self.assertEqual(1, len(result.unexpectedSuccesses))
1774
# Inconsistency, unexpectedSuccesses is a list of tests,
1775
# expectedFailures is a list of reasons?
1776
test = result.unexpectedSuccesses[0]
1777
details = test.getDetails()
1778
self.assertTrue('log' in details)
1781
class TestTestCloning(tests.TestCase):
1782
"""Tests that test cloning of TestCases (as used by multiply_tests)."""
1784
def test_cloned_testcase_does_not_share_details(self):
1785
"""A TestCase cloned with clone_test does not share mutable attributes
1786
such as details or cleanups.
1788
class Test(tests.TestCase):
1790
self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1791
orig_test = Test('test_foo')
1792
cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1793
orig_test.run(unittest.TestResult())
1794
self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1795
self.assertEqual(None, cloned_test.getDetails().get('foo'))
1797
def test_double_apply_scenario_preserves_first_scenario(self):
1798
"""Applying two levels of scenarios to a test preserves the attributes
1799
added by both scenarios.
1801
class Test(tests.TestCase):
1804
test = Test('test_foo')
1805
scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1806
scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1807
suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1808
suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1809
all_tests = list(tests.iter_suite_tests(suite))
1810
self.assertLength(4, all_tests)
1811
all_xys = sorted((t.x, t.y) for t in all_tests)
1812
self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1638
1815
# NB: Don't delete this; it's not actually from 0.11!
1639
1816
@deprecated_function(deprecated_in((0, 11, 0)))
1951
2125
load_list='missing file name', list_only=True)
2128
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2130
_test_needs_features = [features.subunit]
2132
def run_subunit_stream(self, test_name):
2133
from subunit import ProtocolTestCase
2135
return TestUtil.TestSuite([_get_test(test_name)])
2136
stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2137
test_suite_factory=factory)
2138
test = ProtocolTestCase(stream)
2139
result = testtools.TestResult()
2141
content = stream.getvalue()
2142
return content, result
2144
def test_fail_has_log(self):
2145
content, result = self.run_subunit_stream('test_fail')
2146
self.assertEqual(1, len(result.failures))
2147
self.assertContainsRe(content, '(?m)^log$')
2148
self.assertContainsRe(content, 'this test will fail')
2150
def test_error_has_log(self):
2151
content, result = self.run_subunit_stream('test_error')
2152
self.assertContainsRe(content, '(?m)^log$')
2153
self.assertContainsRe(content, 'this test errored')
2155
def test_skip_has_no_log(self):
2156
content, result = self.run_subunit_stream('test_skip')
2157
self.assertNotContainsRe(content, '(?m)^log$')
2158
self.assertNotContainsRe(content, 'this test will be skipped')
2159
self.assertEqual(['reason'], result.skip_reasons.keys())
2160
skips = result.skip_reasons['reason']
2161
self.assertEqual(1, len(skips))
2163
# RemotedTestCase doesn't preserve the "details"
2164
## self.assertFalse('log' in test.getDetails())
2166
def test_missing_feature_has_no_log(self):
2167
content, result = self.run_subunit_stream('test_missing_feature')
2168
self.assertNotContainsRe(content, '(?m)^log$')
2169
self.assertNotContainsRe(content, 'missing the feature')
2170
self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
2171
skips = result.skip_reasons['_MissingFeature\n']
2172
self.assertEqual(1, len(skips))
2174
# RemotedTestCase doesn't preserve the "details"
2175
## self.assertFalse('log' in test.getDetails())
2177
def test_xfail_has_no_log(self):
2178
content, result = self.run_subunit_stream('test_xfail')
2179
self.assertNotContainsRe(content, '(?m)^log$')
2180
self.assertNotContainsRe(content, 'test with expected failure')
2181
self.assertEqual(1, len(result.expectedFailures))
2182
result_content = result.expectedFailures[0][1]
2183
self.assertNotContainsRe(result_content,
2184
'(?m)^(?:Text attachment: )?log(?:$|: )')
2185
self.assertNotContainsRe(result_content, 'test with expected failure')
2187
def test_unexpected_success_has_log(self):
2188
content, result = self.run_subunit_stream('test_unexpected_success')
2189
self.assertContainsRe(content, '(?m)^log$')
2190
self.assertContainsRe(content, 'test with unexpected success')
2191
# GZ 2011-05-18: Old versions of subunit treat unexpected success as a
2192
# success, if a min version check is added remove this
2193
from subunit import TestProtocolClient as _Client
2194
if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
2195
self.expectFailure('subunit treats "unexpectedSuccess"'
2196
' as a plain success',
2197
self.assertEqual, 1, len(result.unexpectedSuccesses))
2198
self.assertEqual(1, len(result.unexpectedSuccesses))
2199
test = result.unexpectedSuccesses[0]
2200
# RemotedTestCase doesn't preserve the "details"
2201
## self.assertTrue('log' in test.getDetails())
2203
def test_success_has_no_log(self):
2204
content, result = self.run_subunit_stream('test_success')
2205
self.assertEqual(1, result.testsRun)
2206
self.assertNotContainsRe(content, '(?m)^log$')
2207
self.assertNotContainsRe(content, 'this test succeeds')
1954
2210
class TestRunBzr(tests.TestCase):
2337
2599
self.assertEqual('bzr: interrupted\n', result[1])
2340
class TestFeature(tests.TestCase):
2342
def test_caching(self):
2343
"""Feature._probe is called by the feature at most once."""
2344
class InstrumentedFeature(tests.Feature):
2346
super(InstrumentedFeature, self).__init__()
2349
self.calls.append('_probe')
2351
feature = InstrumentedFeature()
2353
self.assertEqual(['_probe'], feature.calls)
2355
self.assertEqual(['_probe'], feature.calls)
2357
def test_named_str(self):
2358
"""Feature.__str__ should thunk to feature_name()."""
2359
class NamedFeature(tests.Feature):
2360
def feature_name(self):
2362
feature = NamedFeature()
2363
self.assertEqual('symlinks', str(feature))
2365
def test_default_str(self):
2366
"""Feature.__str__ should default to __class__.__name__."""
2367
class NamedFeature(tests.Feature):
2369
feature = NamedFeature()
2370
self.assertEqual('NamedFeature', str(feature))
2373
class TestUnavailableFeature(tests.TestCase):
2375
def test_access_feature(self):
2376
feature = tests.Feature()
2377
exception = tests.UnavailableFeature(feature)
2378
self.assertIs(feature, exception.args[0])
2381
simple_thunk_feature = tests._CompatabilityThunkFeature(
2382
'bzrlib.tests', 'UnicodeFilename',
2383
'bzrlib.tests.test_selftest.simple_thunk_feature',
2384
deprecated_in((2,1,0)))
2386
class Test_CompatibilityFeature(tests.TestCase):
2388
def test_does_thunk(self):
2389
res = self.callDeprecated(
2390
['bzrlib.tests.test_selftest.simple_thunk_feature was deprecated'
2391
' in version 2.1.0. Use bzrlib.tests.UnicodeFilename instead.'],
2392
simple_thunk_feature.available)
2393
self.assertEqual(tests.UnicodeFilename.available(), res)
2396
class TestModuleAvailableFeature(tests.TestCase):
2398
def test_available_module(self):
2399
feature = tests.ModuleAvailableFeature('bzrlib.tests')
2400
self.assertEqual('bzrlib.tests', feature.module_name)
2401
self.assertEqual('bzrlib.tests', str(feature))
2402
self.assertTrue(feature.available())
2403
self.assertIs(tests, feature.module)
2405
def test_unavailable_module(self):
2406
feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
2407
self.assertEqual('bzrlib.no_such_module_exists', str(feature))
2408
self.assertFalse(feature.available())
2409
self.assertIs(None, feature.module)
2412
2602
class TestSelftestFiltering(tests.TestCase):
2414
2604
def setUp(self):
2935
3129
self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3132
class TestThreadLeakDetection(tests.TestCase):
3133
"""Ensure when tests leak threads we detect and report it"""
3135
class LeakRecordingResult(tests.ExtendedTestResult):
3137
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3139
def _report_thread_leak(self, test, leaks, alive):
3140
self.leaks.append((test, leaks))
3142
def test_testcase_without_addCleanups(self):
3143
"""Check old TestCase instances don't break with leak detection"""
3144
class Test(unittest.TestCase):
3147
result = self.LeakRecordingResult()
3149
result.startTestRun()
3151
result.stopTestRun()
3152
self.assertEqual(result._tests_leaking_threads_count, 0)
3153
self.assertEqual(result.leaks, [])
3155
def test_thread_leak(self):
3156
"""Ensure a thread that outlives the running of a test is reported
3158
Uses a thread that blocks on an event, and is started by the inner
3159
test case. As the thread outlives the inner case's run, it should be
3160
detected as a leak, but the event is then set so that the thread can
3161
be safely joined in cleanup so it's not leaked for real.
3163
event = threading.Event()
3164
thread = threading.Thread(name="Leaker", target=event.wait)
3165
class Test(tests.TestCase):
3166
def test_leak(self):
3168
result = self.LeakRecordingResult()
3169
test = Test("test_leak")
3170
self.addCleanup(thread.join)
3171
self.addCleanup(event.set)
3172
result.startTestRun()
3174
result.stopTestRun()
3175
self.assertEqual(result._tests_leaking_threads_count, 1)
3176
self.assertEqual(result._first_thread_leaker_id, test.id())
3177
self.assertEqual(result.leaks, [(test, set([thread]))])
3178
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3180
def test_multiple_leaks(self):
3181
"""Check multiple leaks are blamed on the test cases at fault
3183
Same concept as the previous test, but has one inner test method that
3184
leaks two threads, and one that doesn't leak at all.
3186
event = threading.Event()
3187
thread_a = threading.Thread(name="LeakerA", target=event.wait)
3188
thread_b = threading.Thread(name="LeakerB", target=event.wait)
3189
thread_c = threading.Thread(name="LeakerC", target=event.wait)
3190
class Test(tests.TestCase):
3191
def test_first_leak(self):
3193
def test_second_no_leak(self):
3195
def test_third_leak(self):
3198
result = self.LeakRecordingResult()
3199
first_test = Test("test_first_leak")
3200
third_test = Test("test_third_leak")
3201
self.addCleanup(thread_a.join)
3202
self.addCleanup(thread_b.join)
3203
self.addCleanup(thread_c.join)
3204
self.addCleanup(event.set)
3205
result.startTestRun()
3207
[first_test, Test("test_second_no_leak"), third_test]
3209
result.stopTestRun()
3210
self.assertEqual(result._tests_leaking_threads_count, 2)
3211
self.assertEqual(result._first_thread_leaker_id, first_test.id())
3212
self.assertEqual(result.leaks, [
3213
(first_test, set([thread_b])),
3214
(third_test, set([thread_a, thread_c]))])
3215
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3218
class TestPostMortemDebugging(tests.TestCase):
3219
"""Check post mortem debugging works when tests fail or error"""
3221
class TracebackRecordingResult(tests.ExtendedTestResult):
3223
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3224
self.postcode = None
3225
def _post_mortem(self, tb=None):
3226
"""Record the code object at the end of the current traceback"""
3227
tb = tb or sys.exc_info()[2]
3230
while next is not None:
3233
self.postcode = tb.tb_frame.f_code
3234
def report_error(self, test, err):
3236
def report_failure(self, test, err):
3239
def test_location_unittest_error(self):
3240
"""Needs right post mortem traceback with erroring unittest case"""
3241
class Test(unittest.TestCase):
3244
result = self.TracebackRecordingResult()
3246
self.assertEqual(result.postcode, Test.runTest.func_code)
3248
def test_location_unittest_failure(self):
3249
"""Needs right post mortem traceback with failing unittest case"""
3250
class Test(unittest.TestCase):
3252
raise self.failureException
3253
result = self.TracebackRecordingResult()
3255
self.assertEqual(result.postcode, Test.runTest.func_code)
3257
def test_location_bt_error(self):
3258
"""Needs right post mortem traceback with erroring bzrlib.tests case"""
3259
class Test(tests.TestCase):
3260
def test_error(self):
3262
result = self.TracebackRecordingResult()
3263
Test("test_error").run(result)
3264
self.assertEqual(result.postcode, Test.test_error.func_code)
3266
def test_location_bt_failure(self):
3267
"""Needs right post mortem traceback with failing bzrlib.tests case"""
3268
class Test(tests.TestCase):
3269
def test_failure(self):
3270
raise self.failureException
3271
result = self.TracebackRecordingResult()
3272
Test("test_failure").run(result)
3273
self.assertEqual(result.postcode, Test.test_failure.func_code)
3275
def test_env_var_triggers_post_mortem(self):
3276
"""Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
3278
result = tests.ExtendedTestResult(StringIO(), 0, 1)
3279
post_mortem_calls = []
3280
self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3281
self.overrideEnv('BZR_TEST_PDB', None)
3282
result._post_mortem(1)
3283
self.overrideEnv('BZR_TEST_PDB', 'on')
3284
result._post_mortem(2)
3285
self.assertEqual([2], post_mortem_calls)
2938
3288
class TestRunSuite(tests.TestCase):
2940
3290
def test_runner_class(self):
2951
3301
self.verbosity)
2952
3302
tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2953
3303
self.assertLength(1, calls)
3306
class _Selftest(object):
3307
"""Mixin for tests needing full selftest output"""
3309
def _inject_stream_into_subunit(self, stream):
3310
"""To be overridden by subclasses that run tests out of process"""
3312
def _run_selftest(self, **kwargs):
3314
self._inject_stream_into_subunit(sio)
3315
tests.selftest(stream=sio, stop_on_failure=False, **kwargs)
3316
return sio.getvalue()
3319
class _ForkedSelftest(_Selftest):
3320
"""Mixin for tests needing full selftest output with forked children"""
3322
_test_needs_features = [features.subunit]
3324
def _inject_stream_into_subunit(self, stream):
3325
"""Monkey-patch subunit so the extra output goes to stream not stdout
3327
Some APIs need rewriting so this kind of bogus hackery can be replaced
3328
by passing the stream param from run_tests down into ProtocolTestCase.
3330
from subunit import ProtocolTestCase
3331
_original_init = ProtocolTestCase.__init__
3332
def _init_with_passthrough(self, *args, **kwargs):
3333
_original_init(self, *args, **kwargs)
3334
self._passthrough = stream
3335
self.overrideAttr(ProtocolTestCase, "__init__", _init_with_passthrough)
3337
def _run_selftest(self, **kwargs):
3338
# GZ 2011-05-26: Add a PosixSystem feature so this check can go away
3339
if getattr(os, "fork", None) is None:
3340
raise tests.TestNotApplicable("Platform doesn't support forking")
3341
# Make sure the fork code is actually invoked by claiming two cores
3342
self.overrideAttr(osutils, "local_concurrency", lambda: 2)
3343
kwargs.setdefault("suite_decorators", []).append(tests.fork_decorator)
3344
return super(_ForkedSelftest, self)._run_selftest(**kwargs)
3347
class TestParallelFork(_ForkedSelftest, tests.TestCase):
3348
"""Check operation of --parallel=fork selftest option"""
3350
def test_error_in_child_during_fork(self):
3351
"""Error in a forked child during test setup should get reported"""
3352
class Test(tests.TestCase):
3353
def testMethod(self):
3355
# We don't care what, just break something that a child will run
3356
self.overrideAttr(tests, "workaround_zealous_crypto_random", None)
3357
out = self._run_selftest(test_suite_factory=Test)
3358
# Lines from the tracebacks of the two child processes may be mixed
3359
# together due to the way subunit parses and forwards the streams,
3360
# so permit extra lines between each part of the error output.
3361
self.assertContainsRe(out,
3364
".+ in fork_for_tests\n"
3366
"\s*workaround_zealous_crypto_random\(\)\n"
3371
class TestUncollectedWarnings(_Selftest, tests.TestCase):
3372
"""Check a test case still alive after being run emits a warning"""
3374
class Test(tests.TestCase):
3375
def test_pass(self):
3377
def test_self_ref(self):
3378
self.also_self = self.test_self_ref
3379
def test_skip(self):
3380
self.skip("Don't need")
3382
def _get_suite(self):
3383
return TestUtil.TestSuite([
3384
self.Test("test_pass"),
3385
self.Test("test_self_ref"),
3386
self.Test("test_skip"),
3389
def _run_selftest_with_suite(self, **kwargs):
3390
old_flags = tests.selftest_debug_flags
3391
tests.selftest_debug_flags = old_flags.union(["uncollected_cases"])
3392
gc_on = gc.isenabled()
3396
output = self._run_selftest(test_suite_factory=self._get_suite,
3401
tests.selftest_debug_flags = old_flags
3402
self.assertNotContainsRe(output, "Uncollected test case.*test_pass")
3403
self.assertContainsRe(output, "Uncollected test case.*test_self_ref")
3406
def test_testsuite(self):
3407
self._run_selftest_with_suite()
3409
def test_pattern(self):
3410
out = self._run_selftest_with_suite(pattern="test_(?:pass|self_ref)$")
3411
self.assertNotContainsRe(out, "test_skip")
3413
def test_exclude_pattern(self):
3414
out = self._run_selftest_with_suite(exclude_pattern="test_skip$")
3415
self.assertNotContainsRe(out, "test_skip")
3417
def test_random_seed(self):
3418
self._run_selftest_with_suite(random_seed="now")
3420
def test_matching_tests_first(self):
3421
self._run_selftest_with_suite(matching_tests_first=True,
3422
pattern="test_self_ref$")
3424
def test_starting_with_and_exclude(self):
3425
out = self._run_selftest_with_suite(starting_with=["bt."],
3426
exclude_pattern="test_skip$")
3427
self.assertNotContainsRe(out, "test_skip")
3429
def test_additonal_decorator(self):
3430
out = self._run_selftest_with_suite(
3431
suite_decorators=[tests.TestDecorator])
3434
class TestUncollectedWarningsSubunit(TestUncollectedWarnings):
3435
"""Check warnings from tests staying alive are emitted with subunit"""
3437
_test_needs_features = [features.subunit]
3439
def _run_selftest_with_suite(self, **kwargs):
3440
return TestUncollectedWarnings._run_selftest_with_suite(self,
3441
runner_class=tests.SubUnitBzrRunner, **kwargs)
3444
class TestUncollectedWarningsForked(_ForkedSelftest, TestUncollectedWarnings):
3445
"""Check warnings from tests staying alive are emitted when forking"""
3448
class TestEnvironHandling(tests.TestCase):
3450
def test_overrideEnv_None_called_twice_doesnt_leak(self):
3451
self.assertFalse('MYVAR' in os.environ)
3452
self.overrideEnv('MYVAR', '42')
3453
# We use an embedded test to make sure we fix the _captureVar bug
3454
class Test(tests.TestCase):
3456
# The first call save the 42 value
3457
self.overrideEnv('MYVAR', None)
3458
self.assertEquals(None, os.environ.get('MYVAR'))
3459
# Make sure we can call it twice
3460
self.overrideEnv('MYVAR', None)
3461
self.assertEquals(None, os.environ.get('MYVAR'))
3463
result = tests.TextTestResult(output, 0, 1)
3464
Test('test_me').run(result)
3465
if not result.wasStrictlySuccessful():
3466
self.fail(output.getvalue())
3467
# We get our value back
3468
self.assertEquals('42', os.environ.get('MYVAR'))
3471
class TestIsolatedEnv(tests.TestCase):
3472
"""Test isolating tests from os.environ.
3474
Since we use tests that are already isolated from os.environ a bit of care
3475
should be taken when designing the tests to avoid bootstrap side-effects.
3476
The tests start an already clean os.environ which allow doing valid
3477
assertions about which variables are present or not and design tests around
3481
class ScratchMonkey(tests.TestCase):
3486
def test_basics(self):
3487
# Make sure we know the definition of BZR_HOME: not part of os.environ
3488
# for tests.TestCase.
3489
self.assertTrue('BZR_HOME' in tests.isolated_environ)
3490
self.assertEquals(None, tests.isolated_environ['BZR_HOME'])
3491
# Being part of isolated_environ, BZR_HOME should not appear here
3492
self.assertFalse('BZR_HOME' in os.environ)
3493
# Make sure we know the definition of LINES: part of os.environ for
3495
self.assertTrue('LINES' in tests.isolated_environ)
3496
self.assertEquals('25', tests.isolated_environ['LINES'])
3497
self.assertEquals('25', os.environ['LINES'])
3499
def test_injecting_unknown_variable(self):
3500
# BZR_HOME is known to be absent from os.environ
3501
test = self.ScratchMonkey('test_me')
3502
tests.override_os_environ(test, {'BZR_HOME': 'foo'})
3503
self.assertEquals('foo', os.environ['BZR_HOME'])
3504
tests.restore_os_environ(test)
3505
self.assertFalse('BZR_HOME' in os.environ)
3507
def test_injecting_known_variable(self):
3508
test = self.ScratchMonkey('test_me')
3509
# LINES is known to be present in os.environ
3510
tests.override_os_environ(test, {'LINES': '42'})
3511
self.assertEquals('42', os.environ['LINES'])
3512
tests.restore_os_environ(test)
3513
self.assertEquals('25', os.environ['LINES'])
3515
def test_deleting_variable(self):
3516
test = self.ScratchMonkey('test_me')
3517
# LINES is known to be present in os.environ
3518
tests.override_os_environ(test, {'LINES': None})
3519
self.assertTrue('LINES' not in os.environ)
3520
tests.restore_os_environ(test)
3521
self.assertEquals('25', os.environ['LINES'])
3524
class TestDocTestSuiteIsolation(tests.TestCase):
3525
"""Test that `tests.DocTestSuite` isolates doc tests from os.environ.
3527
Since tests.TestCase alreay provides an isolation from os.environ, we use
3528
the clean environment as a base for testing. To precisely capture the
3529
isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
3532
We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
3533
not `os.environ` so each test overrides it to suit its needs.
3537
def get_doctest_suite_for_string(self, klass, string):
3538
class Finder(doctest.DocTestFinder):
3540
def find(*args, **kwargs):
3541
test = doctest.DocTestParser().get_doctest(
3542
string, {}, 'foo', 'foo.py', 0)
3545
suite = klass(test_finder=Finder())
3548
def run_doctest_suite_for_string(self, klass, string):
3549
suite = self.get_doctest_suite_for_string(klass, string)
3551
result = tests.TextTestResult(output, 0, 1)
3553
return result, output
3555
def assertDocTestStringSucceds(self, klass, string):
3556
result, output = self.run_doctest_suite_for_string(klass, string)
3557
if not result.wasStrictlySuccessful():
3558
self.fail(output.getvalue())
3560
def assertDocTestStringFails(self, klass, string):
3561
result, output = self.run_doctest_suite_for_string(klass, string)
3562
if result.wasStrictlySuccessful():
3563
self.fail(output.getvalue())
3565
def test_injected_variable(self):
3566
self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
3569
>>> os.environ['LINES']
3572
# doctest.DocTestSuite fails as it sees '25'
3573
self.assertDocTestStringFails(doctest.DocTestSuite, test)
3574
# tests.DocTestSuite sees '42'
3575
self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3577
def test_deleted_variable(self):
3578
self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
3581
>>> os.environ.get('LINES')
3583
# doctest.DocTestSuite fails as it sees '25'
3584
self.assertDocTestStringFails(doctest.DocTestSuite, test)
3585
# tests.DocTestSuite sees None
3586
self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3589
class TestSelftestExcludePatterns(tests.TestCase):
3592
super(TestSelftestExcludePatterns, self).setUp()
3593
self.overrideAttr(tests, 'test_suite', self.suite_factory)
3595
def suite_factory(self, keep_only=None, starting_with=None):
3596
"""A test suite factory with only a few tests."""
3597
class Test(tests.TestCase):
3599
# We don't need the full class path
3600
return self._testMethodName
3607
return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
3609
def assertTestList(self, expected, *selftest_args):
3610
# We rely on setUp installing the right test suite factory so we can
3611
# test at the command level without loading the whole test suite
3612
out, err = self.run_bzr(('selftest', '--list') + selftest_args)
3613
actual = out.splitlines()
3614
self.assertEquals(expected, actual)
3616
def test_full_list(self):
3617
self.assertTestList(['a', 'b', 'c'])
3619
def test_single_exclude(self):
3620
self.assertTestList(['b', 'c'], '-x', 'a')
3622
def test_mutiple_excludes(self):
3623
self.assertTestList(['c'], '-x', 'a', '-x', 'b')
3626
class TestCounterHooks(tests.TestCase, SelfTestHelper):
3628
_test_needs_features = [features.subunit]
3631
super(TestCounterHooks, self).setUp()
3632
class Test(tests.TestCase):
3635
super(Test, self).setUp()
3636
self.hooks = hooks.Hooks()
3637
self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
3638
self.install_counter_hook(self.hooks, 'myhook')
3643
def run_hook_once(self):
3644
for hook in self.hooks['myhook']:
3647
self.test_class = Test
3649
def assertHookCalls(self, expected_calls, test_name):
3650
test = self.test_class(test_name)
3651
result = unittest.TestResult()
3653
self.assertTrue(hasattr(test, '_counters'))
3654
self.assertTrue(test._counters.has_key('myhook'))
3655
self.assertEquals(expected_calls, test._counters['myhook'])
3657
def test_no_hook(self):
3658
self.assertHookCalls(0, 'no_hook')
3660
def test_run_hook_once(self):
3661
tt = features.testtools
3662
if tt.module.__version__ < (0, 9, 8):
3663
raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
3664
self.assertHookCalls(1, 'run_hook_once')
added
removed
bzrlib/tests/test_selftest.py
