346
350
{'bzrdir_format': formats[1]._matchingbzrdir,
347
351
'transport_readonly_server': 'b',
348
352
'transport_server': 'a',
349
'workingtree_format': formats[1]})],
353
'workingtree_format': formats[1]}),
354
('WorkingTreeFormat6',
355
{'bzrdir_format': formats[2]._matchingbzrdir,
356
'transport_readonly_server': 'b',
357
'transport_server': 'a',
358
'workingtree_format': formats[2]}),
359
('WorkingTreeFormat6,remote',
360
{'bzrdir_format': formats[2]._matchingbzrdir,
361
'repo_is_remote': True,
362
'transport_readonly_server': 'd',
363
'transport_server': 'c',
364
'vfs_transport_factory': 'e',
365
'workingtree_format': formats[2]}),
353
369
class TestTreeScenarios(tests.TestCase):
355
371
def test_scenarios(self):
356
372
# the tree implementation scenario generator is meant to setup one
357
# instance for each working tree format, and one additional instance
373
# instance for each working tree format, one additional instance
358
374
# that will use the default wt format, but create a revision tree for
359
# the tests. this means that the wt ones should have the
360
# workingtree_to_test_tree attribute set to 'return_parameter' and the
361
# revision one set to revision_tree_from_workingtree.
375
# the tests, and one more that uses the default wt format as a
376
# lightweight checkout of a remote repository. This means that the wt
377
# ones should have the workingtree_to_test_tree attribute set to
378
# 'return_parameter' and the revision one set to
379
# revision_tree_from_workingtree.
363
381
from bzrlib.tests.per_tree import (
364
382
_dirstate_tree_from_workingtree,
1062
1093
self.expectFailure("No absolute truth", self.assertTrue, True)
1063
1094
runner = tests.TextTestRunner(stream=StringIO())
1064
1095
result = self.run_test_runner(runner, Test("test_truth"))
1065
if testtools_version[:3] <= (0, 9, 11):
1066
self.assertContainsRe(runner.stream.getvalue(),
1068
"FAIL: \\S+\.test_truth\n"
1071
"No absolute truth\n"
1074
"Ran 1 test in .*\n"
1076
"FAILED \\(failures=1\\)\n\\Z")
1078
self.assertContainsRe(runner.stream.getvalue(),
1080
"FAIL: \\S+\.test_truth\n"
1082
"Empty attachments:\n"
1085
"reason: {{{No absolute truth}}}\n"
1087
"Ran 1 test in .*\n"
1089
"FAILED \\(failures=1\\)\n\\Z")
1096
self.assertContainsRe(runner.stream.getvalue(),
1098
"FAIL: \\S+\.test_truth\n"
1101
"\\s*(?:Text attachment: )?reason"
1107
"Ran 1 test in .*\n"
1109
"FAILED \\(failures=1\\)\n\\Z")
1091
1111
def test_result_decorator(self):
1092
1112
# decorate results
1743
1760
result = self._run_test('test_fail')
1744
1761
self.assertEqual(1, len(result.failures))
1745
1762
result_content = result.failures[0][1]
1746
if testtools_version < (0, 9, 12):
1747
self.assertContainsRe(result_content, 'Text attachment: log')
1763
self.assertContainsRe(result_content,
1764
'(?m)^(?:Text attachment: )?log(?:$|: )')
1748
1765
self.assertContainsRe(result_content, 'this was a failing test')
1750
1767
def test_error_has_log(self):
1751
1768
result = self._run_test('test_error')
1752
1769
self.assertEqual(1, len(result.errors))
1753
1770
result_content = result.errors[0][1]
1754
if testtools_version < (0, 9, 12):
1755
self.assertContainsRe(result_content, 'Text attachment: log')
1771
self.assertContainsRe(result_content,
1772
'(?m)^(?:Text attachment: )?log(?:$|: )')
1756
1773
self.assertContainsRe(result_content, 'this test errored')
1758
1775
def test_skip_has_no_log(self):
3314
3333
self.assertLength(1, calls)
3336
class _Selftest(object):
3337
"""Mixin for tests needing full selftest output"""
3339
def _inject_stream_into_subunit(self, stream):
3340
"""To be overridden by subclasses that run tests out of process"""
3342
def _run_selftest(self, **kwargs):
3344
self._inject_stream_into_subunit(sio)
3345
tests.selftest(stream=sio, stop_on_failure=False, **kwargs)
3346
return sio.getvalue()
3349
class _ForkedSelftest(_Selftest):
3350
"""Mixin for tests needing full selftest output with forked children"""
3352
_test_needs_features = [features.subunit]
3354
def _inject_stream_into_subunit(self, stream):
3355
"""Monkey-patch subunit so the extra output goes to stream not stdout
3357
Some APIs need rewriting so this kind of bogus hackery can be replaced
3358
by passing the stream param from run_tests down into ProtocolTestCase.
3360
from subunit import ProtocolTestCase
3361
_original_init = ProtocolTestCase.__init__
3362
def _init_with_passthrough(self, *args, **kwargs):
3363
_original_init(self, *args, **kwargs)
3364
self._passthrough = stream
3365
self.overrideAttr(ProtocolTestCase, "__init__", _init_with_passthrough)
3367
def _run_selftest(self, **kwargs):
3368
# GZ 2011-05-26: Add a PosixSystem feature so this check can go away
3369
if getattr(os, "fork", None) is None:
3370
raise tests.TestNotApplicable("Platform doesn't support forking")
3371
# Make sure the fork code is actually invoked by claiming two cores
3372
self.overrideAttr(osutils, "local_concurrency", lambda: 2)
3373
kwargs.setdefault("suite_decorators", []).append(tests.fork_decorator)
3374
return super(_ForkedSelftest, self)._run_selftest(**kwargs)
3377
class TestParallelFork(_ForkedSelftest, tests.TestCase):
3378
"""Check operation of --parallel=fork selftest option"""
3380
def test_error_in_child_during_fork(self):
3381
"""Error in a forked child during test setup should get reported"""
3382
class Test(tests.TestCase):
3383
def testMethod(self):
3385
# We don't care what, just break something that a child will run
3386
self.overrideAttr(tests, "workaround_zealous_crypto_random", None)
3387
out = self._run_selftest(test_suite_factory=Test)
3388
# Lines from the tracebacks of the two child processes may be mixed
3389
# together due to the way subunit parses and forwards the streams,
3390
# so permit extra lines between each part of the error output.
3391
self.assertContainsRe(out,
3394
".+ in fork_for_tests\n"
3396
"\s*workaround_zealous_crypto_random\(\)\n"
3401
class TestUncollectedWarnings(_Selftest, tests.TestCase):
3402
"""Check a test case still alive after being run emits a warning"""
3404
class Test(tests.TestCase):
3405
def test_pass(self):
3407
def test_self_ref(self):
3408
self.also_self = self.test_self_ref
3409
def test_skip(self):
3410
self.skip("Don't need")
3412
def _get_suite(self):
3413
return TestUtil.TestSuite([
3414
self.Test("test_pass"),
3415
self.Test("test_self_ref"),
3416
self.Test("test_skip"),
3419
def _run_selftest_with_suite(self, **kwargs):
3420
old_flags = tests.selftest_debug_flags
3421
tests.selftest_debug_flags = old_flags.union(["uncollected_cases"])
3422
gc_on = gc.isenabled()
3426
output = self._run_selftest(test_suite_factory=self._get_suite,
3431
tests.selftest_debug_flags = old_flags
3432
self.assertNotContainsRe(output, "Uncollected test case.*test_pass")
3433
self.assertContainsRe(output, "Uncollected test case.*test_self_ref")
3436
def test_testsuite(self):
3437
self._run_selftest_with_suite()
3439
def test_pattern(self):
3440
out = self._run_selftest_with_suite(pattern="test_(?:pass|self_ref)$")
3441
self.assertNotContainsRe(out, "test_skip")
3443
def test_exclude_pattern(self):
3444
out = self._run_selftest_with_suite(exclude_pattern="test_skip$")
3445
self.assertNotContainsRe(out, "test_skip")
3447
def test_random_seed(self):
3448
self._run_selftest_with_suite(random_seed="now")
3450
def test_matching_tests_first(self):
3451
self._run_selftest_with_suite(matching_tests_first=True,
3452
pattern="test_self_ref$")
3454
def test_starting_with_and_exclude(self):
3455
out = self._run_selftest_with_suite(starting_with=["bt."],
3456
exclude_pattern="test_skip$")
3457
self.assertNotContainsRe(out, "test_skip")
3459
def test_additonal_decorator(self):
3460
out = self._run_selftest_with_suite(
3461
suite_decorators=[tests.TestDecorator])
3464
class TestUncollectedWarningsSubunit(TestUncollectedWarnings):
3465
"""Check warnings from tests staying alive are emitted with subunit"""
3467
_test_needs_features = [features.subunit]
3469
def _run_selftest_with_suite(self, **kwargs):
3470
return TestUncollectedWarnings._run_selftest_with_suite(self,
3471
runner_class=tests.SubUnitBzrRunner, **kwargs)
3474
class TestUncollectedWarningsForked(_ForkedSelftest, TestUncollectedWarnings):
3475
"""Check warnings from tests staying alive are emitted when forking"""
3317
3478
class TestEnvironHandling(tests.TestCase):
3319
3480
def test_overrideEnv_None_called_twice_doesnt_leak(self):