333
337
def test_scenarios(self):
334
338
# check that constructor parameters are passed through to the adapted
336
from bzrlib.tests.per_workingtree import make_scenarios
340
from .per_workingtree import make_scenarios
339
formats = [workingtree.WorkingTreeFormat2(),
340
workingtree.WorkingTreeFormat3(),]
341
scenarios = make_scenarios(server1, server2, formats)
343
formats = [workingtree_4.WorkingTreeFormat4(),
344
workingtree_3.WorkingTreeFormat3(),
345
workingtree_4.WorkingTreeFormat6()]
346
scenarios = make_scenarios(server1, server2, formats,
347
remote_server='c', remote_readonly_server='d',
348
remote_backing_server='e')
342
349
self.assertEqual([
343
('WorkingTreeFormat2',
344
{'bzrdir_format': formats[0]._matchingbzrdir,
350
('WorkingTreeFormat4',
351
{'bzrdir_format': formats[0]._matchingcontroldir,
345
352
'transport_readonly_server': 'b',
346
353
'transport_server': 'a',
347
354
'workingtree_format': formats[0]}),
348
355
('WorkingTreeFormat3',
349
{'bzrdir_format': formats[1]._matchingbzrdir,
350
'transport_readonly_server': 'b',
351
'transport_server': 'a',
352
'workingtree_format': formats[1]})],
356
{'bzrdir_format': formats[1]._matchingcontroldir,
357
'transport_readonly_server': 'b',
358
'transport_server': 'a',
359
'workingtree_format': formats[1]}),
360
('WorkingTreeFormat6',
361
{'bzrdir_format': formats[2]._matchingcontroldir,
362
'transport_readonly_server': 'b',
363
'transport_server': 'a',
364
'workingtree_format': formats[2]}),
365
('WorkingTreeFormat6,remote',
366
{'bzrdir_format': formats[2]._matchingcontroldir,
367
'repo_is_remote': True,
368
'transport_readonly_server': 'd',
369
'transport_server': 'c',
370
'vfs_transport_factory': 'e',
371
'workingtree_format': formats[2]}),
356
375
class TestTreeScenarios(tests.TestCase):
358
377
def test_scenarios(self):
359
378
# the tree implementation scenario generator is meant to setup one
360
# instance for each working tree format, and one additional instance
379
# instance for each working tree format, one additional instance
361
380
# that will use the default wt format, but create a revision tree for
362
# the tests. this means that the wt ones should have the
363
# workingtree_to_test_tree attribute set to 'return_parameter' and the
364
# revision one set to revision_tree_from_workingtree.
381
# the tests, and one more that uses the default wt format as a
382
# lightweight checkout of a remote repository. This means that the wt
383
# ones should have the workingtree_to_test_tree attribute set to
384
# 'return_parameter' and the revision one set to
385
# revision_tree_from_workingtree.
366
from bzrlib.tests.per_tree import (
387
from .per_tree import (
367
388
_dirstate_tree_from_workingtree,
369
390
preview_tree_pre,
376
formats = [workingtree.WorkingTreeFormat2(),
377
workingtree.WorkingTreeFormat3(),]
397
smart_server = test_server.SmartTCPServer_for_testing
398
smart_readonly_server = test_server.ReadonlySmartTCPServer_for_testing
399
mem_server = memory.MemoryServer
400
formats = [workingtree_4.WorkingTreeFormat4(),
401
workingtree_3.WorkingTreeFormat3(),]
378
402
scenarios = make_scenarios(server1, server2, formats)
379
self.assertEqual(7, len(scenarios))
380
default_wt_format = workingtree.WorkingTreeFormat4._default_format
381
wt4_format = workingtree.WorkingTreeFormat4()
382
wt5_format = workingtree.WorkingTreeFormat5()
403
self.assertEqual(8, len(scenarios))
404
default_wt_format = workingtree.format_registry.get_default()
405
wt4_format = workingtree_4.WorkingTreeFormat4()
406
wt5_format = workingtree_4.WorkingTreeFormat5()
407
wt6_format = workingtree_4.WorkingTreeFormat6()
383
408
expected_scenarios = [
384
('WorkingTreeFormat2',
385
{'bzrdir_format': formats[0]._matchingbzrdir,
409
('WorkingTreeFormat4',
410
{'bzrdir_format': formats[0]._matchingcontroldir,
386
411
'transport_readonly_server': 'b',
387
412
'transport_server': 'a',
388
413
'workingtree_format': formats[0],
389
414
'_workingtree_to_test_tree': return_parameter,
391
416
('WorkingTreeFormat3',
392
{'bzrdir_format': formats[1]._matchingbzrdir,
417
{'bzrdir_format': formats[1]._matchingcontroldir,
393
418
'transport_readonly_server': 'b',
394
419
'transport_server': 'a',
395
420
'workingtree_format': formats[1],
396
421
'_workingtree_to_test_tree': return_parameter,
423
('WorkingTreeFormat6,remote',
424
{'bzrdir_format': wt6_format._matchingcontroldir,
425
'repo_is_remote': True,
426
'transport_readonly_server': smart_readonly_server,
427
'transport_server': smart_server,
428
'vfs_transport_factory': mem_server,
429
'workingtree_format': wt6_format,
430
'_workingtree_to_test_tree': return_parameter,
399
433
{'_workingtree_to_test_tree': revision_tree_from_workingtree,
400
'bzrdir_format': default_wt_format._matchingbzrdir,
434
'bzrdir_format': default_wt_format._matchingcontroldir,
401
435
'transport_readonly_server': 'b',
402
436
'transport_server': 'a',
403
437
'workingtree_format': default_wt_format,
405
439
('DirStateRevisionTree,WT4',
406
440
{'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
407
'bzrdir_format': wt4_format._matchingbzrdir,
441
'bzrdir_format': wt4_format._matchingcontroldir,
408
442
'transport_readonly_server': 'b',
409
443
'transport_server': 'a',
410
444
'workingtree_format': wt4_format,
412
446
('DirStateRevisionTree,WT5',
413
447
{'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
414
'bzrdir_format': wt5_format._matchingbzrdir,
448
'bzrdir_format': wt5_format._matchingcontroldir,
415
449
'transport_readonly_server': 'b',
416
450
'transport_server': 'a',
417
451
'workingtree_format': wt5_format,
420
454
{'_workingtree_to_test_tree': preview_tree_pre,
421
'bzrdir_format': default_wt_format._matchingbzrdir,
455
'bzrdir_format': default_wt_format._matchingcontroldir,
422
456
'transport_readonly_server': 'b',
423
457
'transport_server': 'a',
424
458
'workingtree_format': default_wt_format}),
425
459
('PreviewTreePost',
426
460
{'_workingtree_to_test_tree': preview_tree_post,
427
'bzrdir_format': default_wt_format._matchingbzrdir,
461
'bzrdir_format': default_wt_format._matchingcontroldir,
428
462
'transport_readonly_server': 'b',
429
463
'transport_server': 'a',
430
464
'workingtree_format': default_wt_format}),
592
632
the_branch = builder.get_branch()
593
633
# Guard against regression into MemoryTransport leaking
594
634
# files to disk instead of keeping them in memory.
595
self.failIf(osutils.lexists('dir'))
596
dir_format = bzrdir.format_registry.make_bzrdir('knit')
635
self.assertFalse(osutils.lexists('dir'))
636
dir_format = controldir.format_registry.make_controldir('knit')
597
637
self.assertEqual(dir_format.repository_format.__class__,
598
638
the_branch.repository._format.__class__)
599
self.assertEqual('Bazaar-NG Knit Repository Format 1',
639
self.assertEqual(b'Bazaar-NG Knit Repository Format 1',
600
640
self.get_transport().get_bytes(
601
641
'dir/.bzr/repository/format'))
603
643
def test_dangling_locks_cause_failures(self):
604
644
class TestDanglingLock(tests.TestCaseWithMemoryTransport):
605
645
def test_function(self):
606
t = self.get_transport('.')
646
t = self.get_transport_from_path('.')
607
647
l = lockdir.LockDir(t, 'lock')
610
650
test = TestDanglingLock('test_function')
611
651
result = test.run()
652
total_failures = result.errors + result.failures
612
653
if self._lock_check_thorough:
613
self.assertEqual(1, len(result.errors))
654
self.assertEqual(1, len(total_failures))
615
656
# When _lock_check_thorough is disabled, then we don't trigger a
617
self.assertEqual(0, len(result.errors))
658
self.assertEqual(0, len(total_failures))
620
661
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
621
662
"""Tests for the convenience functions TestCaseWithTransport introduces."""
623
664
def test_get_readonly_url_none(self):
624
from bzrlib.transport import get_transport
625
from bzrlib.transport.readonly import ReadonlyTransportDecorator
665
from ..transport.readonly import ReadonlyTransportDecorator
626
666
self.vfs_transport_factory = memory.MemoryServer
627
667
self.transport_readonly_server = None
628
668
# calling get_readonly_transport() constructs a decorator on the url
630
670
url = self.get_readonly_url()
631
671
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))
672
t = transport.get_transport_from_url(url)
673
t2 = transport.get_transport_from_url(url2)
674
self.assertIsInstance(t, ReadonlyTransportDecorator)
675
self.assertIsInstance(t2, ReadonlyTransportDecorator)
636
676
self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
638
678
def test_get_readonly_url_http(self):
639
from bzrlib.tests.http_server import HttpServer
640
from bzrlib.transport import get_transport
641
from bzrlib.transport.http import HttpTransportBase
679
from .http_server import HttpServer
680
from ..transport.http import HttpTransport
642
681
self.transport_server = test_server.LocalURLServer
643
682
self.transport_readonly_server = HttpServer
644
683
# calling get_readonly_transport() gives us a HTTP server instance.
645
684
url = self.get_readonly_url()
646
685
url2 = self.get_readonly_url('foo/bar')
647
686
# the transport returned may be any HttpTransportBase subclass
648
t = get_transport(url)
649
t2 = get_transport(url2)
650
self.failUnless(isinstance(t, HttpTransportBase))
651
self.failUnless(isinstance(t2, HttpTransportBase))
687
t = transport.get_transport_from_url(url)
688
t2 = transport.get_transport_from_url(url2)
689
self.assertIsInstance(t, HttpTransport)
690
self.assertIsInstance(t2, HttpTransport)
652
691
self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
654
693
def test_is_directory(self):
743
778
r"^ +[0-9]+ms\*$")
745
780
def test_unittest_reporting_unittest_class(self):
746
# getting the time from a non-bzrlib test works ok
781
# getting the time from a non-breezy test works ok
747
782
class ShortDelayTestCase(unittest.TestCase):
748
783
def test_short_delay(self):
749
784
time.sleep(0.003)
750
785
self.check_timing(ShortDelayTestCase('test_short_delay'),
753
def _patch_get_bzr_source_tree(self):
754
# Reading from the actual source tree breaks isolation, but we don't
755
# want to assume that thats *all* that would happen.
756
self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', lambda: None)
758
def test_assigned_benchmark_file_stores_date(self):
759
self._patch_get_bzr_source_tree()
761
result = bzrlib.tests.TextTestResult(self._log_file,
766
output_string = output.getvalue()
767
# if you are wondering about the regexp please read the comment in
768
# test_bench_history (bzrlib.tests.test_selftest.TestRunner)
769
# XXX: what comment? -- Andrew Bennetts
770
self.assertContainsRe(output_string, "--date [0-9.]+")
772
def test_benchhistory_records_test_times(self):
773
self._patch_get_bzr_source_tree()
774
result_stream = StringIO()
775
result = bzrlib.tests.TextTestResult(
779
bench_history=result_stream
782
# we want profile a call and check that its test duration is recorded
783
# make a new test instance that when run will generate a benchmark
784
example_test_case = TestTestResult("_time_hello_world_encoding")
785
# execute the test, which should succeed and record times
786
example_test_case.run(result)
787
lines = result_stream.getvalue().splitlines()
788
self.assertEqual(2, len(lines))
789
self.assertContainsRe(lines[1],
790
" *[0-9]+ms bzrlib.tests.test_selftest.TestTestResult"
791
"._time_hello_world_encoding")
793
788
def _time_hello_world_encoding(self):
794
789
"""Profile two sleep calls
796
791
This is used to exercise the test framework.
798
self.time(unicode, 'hello', errors='replace')
799
self.time(unicode, 'world', errors='replace')
793
self.time(text_type, 'hello', errors='replace')
794
self.time(text_type, 'world', errors='replace')
801
796
def test_lsprofiling(self):
802
797
"""Verbose test result prints lsprof statistics from test cases."""
803
self.requireFeature(test_lsprof.LSProfFeature)
798
self.requireFeature(features.lsprof_feature)
804
799
result_stream = StringIO()
805
result = bzrlib.tests.VerboseTestResult(
806
unittest._WritelnDecorator(result_stream),
800
result = breezy.tests.VerboseTestResult(
1643
1671
class Test(tests.TestCase):
1645
1673
def setUp(self):
1646
tests.TestCase.setUp(self)
1674
super(Test, self).setUp()
1647
1675
self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1649
1677
def test_value(self):
1650
1678
self.assertEqual('original', self.orig)
1651
1679
self.assertEqual('modified', obj.test_attr)
1653
test = Test('test_value')
1654
test.run(unittest.TestResult())
1681
self._run_successful_test(Test('test_value'))
1655
1682
self.assertEqual('original', obj.test_attr)
1684
def test_overrideAttr_with_no_existing_value_and_value(self):
1685
# Do not define the test_attribute
1686
obj = self # Make 'obj' visible to the embedded test
1687
class Test(tests.TestCase):
1690
tests.TestCase.setUp(self)
1691
self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1693
def test_value(self):
1694
self.assertEqual(tests._unitialized_attr, self.orig)
1695
self.assertEqual('modified', obj.test_attr)
1697
self._run_successful_test(Test('test_value'))
1698
self.assertRaises(AttributeError, getattr, obj, 'test_attr')
1700
def test_overrideAttr_with_no_existing_value_and_no_value(self):
1701
# Do not define the test_attribute
1702
obj = self # Make 'obj' visible to the embedded test
1703
class Test(tests.TestCase):
1706
tests.TestCase.setUp(self)
1707
self.orig = self.overrideAttr(obj, 'test_attr')
1709
def test_value(self):
1710
self.assertEqual(tests._unitialized_attr, self.orig)
1711
self.assertRaises(AttributeError, getattr, obj, 'test_attr')
1713
self._run_successful_test(Test('test_value'))
1714
self.assertRaises(AttributeError, getattr, obj, 'test_attr')
1716
def test_recordCalls(self):
1717
from breezy.tests import test_selftest
1718
calls = self.recordCalls(
1719
test_selftest, '_add_numbers')
1720
self.assertEqual(test_selftest._add_numbers(2, 10),
1722
self.assertEqual(calls, [((2, 10), {})])
1725
def _add_numbers(a, b):
1729
class _MissingFeature(features.Feature):
1732
missing_feature = _MissingFeature()
1735
def _get_test(name):
1736
"""Get an instance of a specific example test.
1738
We protect this in a function so that they don't auto-run in the test
1742
class ExampleTests(tests.TestCase):
1744
def test_fail(self):
1745
mutter('this was a failing test')
1746
self.fail('this test will fail')
1748
def test_error(self):
1749
mutter('this test errored')
1750
raise RuntimeError('gotcha')
1752
def test_missing_feature(self):
1753
mutter('missing the feature')
1754
self.requireFeature(missing_feature)
1756
def test_skip(self):
1757
mutter('this test will be skipped')
1758
raise tests.TestSkipped('reason')
1760
def test_success(self):
1761
mutter('this test succeeds')
1763
def test_xfail(self):
1764
mutter('test with expected failure')
1765
self.knownFailure('this_fails')
1767
def test_unexpected_success(self):
1768
mutter('test with unexpected success')
1769
self.expectFailure('should_fail', lambda: None)
1771
return ExampleTests(name)
1774
def _get_skip_reasons(result):
1775
# GZ 2017-06-06: Newer testtools doesn't have this, uses detail instead
1776
return result.skip_reasons
1779
class TestTestCaseLogDetails(tests.TestCase):
1781
def _run_test(self, test_name):
1782
test = _get_test(test_name)
1783
result = testtools.TestResult()
1787
def test_fail_has_log(self):
1788
result = self._run_test('test_fail')
1789
self.assertEqual(1, len(result.failures))
1790
result_content = result.failures[0][1]
1791
self.assertContainsRe(result_content,
1792
'(?m)^(?:Text attachment: )?log(?:$|: )')
1793
self.assertContainsRe(result_content, 'this was a failing test')
1795
def test_error_has_log(self):
1796
result = self._run_test('test_error')
1797
self.assertEqual(1, len(result.errors))
1798
result_content = result.errors[0][1]
1799
self.assertContainsRe(result_content,
1800
'(?m)^(?:Text attachment: )?log(?:$|: )')
1801
self.assertContainsRe(result_content, 'this test errored')
1803
def test_skip_has_no_log(self):
1804
result = self._run_test('test_skip')
1805
reasons = _get_skip_reasons(result)
1806
self.assertEqual({'reason'}, set(reasons))
1807
skips = reasons['reason']
1808
self.assertEqual(1, len(skips))
1810
self.assertFalse('log' in test.getDetails())
1812
def test_missing_feature_has_no_log(self):
1813
# testtools doesn't know about addNotSupported, so it just gets
1814
# considered as a skip
1815
result = self._run_test('test_missing_feature')
1816
reasons = _get_skip_reasons(result)
1817
self.assertEqual({str(missing_feature)}, set(reasons))
1818
skips = reasons[str(missing_feature)]
1819
self.assertEqual(1, len(skips))
1821
self.assertFalse('log' in test.getDetails())
1823
def test_xfail_has_no_log(self):
1824
result = self._run_test('test_xfail')
1825
self.assertEqual(1, len(result.expectedFailures))
1826
result_content = result.expectedFailures[0][1]
1827
self.assertNotContainsRe(result_content,
1828
'(?m)^(?:Text attachment: )?log(?:$|: )')
1829
self.assertNotContainsRe(result_content, 'test with expected failure')
1831
def test_unexpected_success_has_log(self):
1832
result = self._run_test('test_unexpected_success')
1833
self.assertEqual(1, len(result.unexpectedSuccesses))
1834
# Inconsistency, unexpectedSuccesses is a list of tests,
1835
# expectedFailures is a list of reasons?
1836
test = result.unexpectedSuccesses[0]
1837
details = test.getDetails()
1838
self.assertTrue('log' in details)
1841
class TestTestCloning(tests.TestCase):
1842
"""Tests that test cloning of TestCases (as used by multiply_tests)."""
1844
def test_cloned_testcase_does_not_share_details(self):
1845
"""A TestCase cloned with clone_test does not share mutable attributes
1846
such as details or cleanups.
1848
class Test(tests.TestCase):
1850
self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1851
orig_test = Test('test_foo')
1852
cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1853
orig_test.run(unittest.TestResult())
1854
self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1855
self.assertEqual(None, cloned_test.getDetails().get('foo'))
1857
def test_double_apply_scenario_preserves_first_scenario(self):
1858
"""Applying two levels of scenarios to a test preserves the attributes
1859
added by both scenarios.
1861
class Test(tests.TestCase):
1864
test = Test('test_foo')
1865
scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1866
scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1867
suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1868
suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1869
all_tests = list(tests.iter_suite_tests(suite))
1870
self.assertLength(4, all_tests)
1871
all_xys = sorted((t.x, t.y) for t in all_tests)
1872
self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1658
1875
# NB: Don't delete this; it's not actually from 0.11!
1659
1876
@deprecated_function(deprecated_in((0, 11, 0)))
1971
2187
load_list='missing file name', list_only=True)
2190
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2192
_test_needs_features = [features.subunit]
2194
def run_subunit_stream(self, test_name):
2195
from subunit import ProtocolTestCase
2197
return TestUtil.TestSuite([_get_test(test_name)])
2198
stream = self.run_selftest(
2199
runner_class=tests.SubUnitBzrRunnerv1,
2200
test_suite_factory=factory)
2201
test = ProtocolTestCase(stream)
2202
result = testtools.TestResult()
2204
content = stream.getvalue()
2205
return content, result
2207
def test_fail_has_log(self):
2208
content, result = self.run_subunit_stream('test_fail')
2209
self.assertEqual(1, len(result.failures))
2210
self.assertContainsRe(content, '(?m)^log$')
2211
self.assertContainsRe(content, 'this test will fail')
2213
def test_error_has_log(self):
2214
content, result = self.run_subunit_stream('test_error')
2215
self.assertContainsRe(content, '(?m)^log$')
2216
self.assertContainsRe(content, 'this test errored')
2218
def test_skip_has_no_log(self):
2219
content, result = self.run_subunit_stream('test_skip')
2220
self.assertNotContainsRe(content, '(?m)^log$')
2221
self.assertNotContainsRe(content, 'this test will be skipped')
2222
reasons = _get_skip_reasons(result)
2223
self.assertEqual({'reason'}, set(reasons))
2224
skips = reasons['reason']
2225
self.assertEqual(1, len(skips))
2227
# RemotedTestCase doesn't preserve the "details"
2228
## self.assertFalse('log' in test.getDetails())
2230
def test_missing_feature_has_no_log(self):
2231
content, result = self.run_subunit_stream('test_missing_feature')
2232
self.assertNotContainsRe(content, '(?m)^log$')
2233
self.assertNotContainsRe(content, 'missing the feature')
2234
reasons = _get_skip_reasons(result)
2235
self.assertEqual({'_MissingFeature\n'}, set(reasons))
2236
skips = reasons['_MissingFeature\n']
2237
self.assertEqual(1, len(skips))
2239
# RemotedTestCase doesn't preserve the "details"
2240
## self.assertFalse('log' in test.getDetails())
2242
def test_xfail_has_no_log(self):
2243
content, result = self.run_subunit_stream('test_xfail')
2244
self.assertNotContainsRe(content, '(?m)^log$')
2245
self.assertNotContainsRe(content, 'test with expected failure')
2246
self.assertEqual(1, len(result.expectedFailures))
2247
result_content = result.expectedFailures[0][1]
2248
self.assertNotContainsRe(result_content,
2249
'(?m)^(?:Text attachment: )?log(?:$|: )')
2250
self.assertNotContainsRe(result_content, 'test with expected failure')
2252
def test_unexpected_success_has_log(self):
2253
content, result = self.run_subunit_stream('test_unexpected_success')
2254
self.assertContainsRe(content, '(?m)^log$')
2255
self.assertContainsRe(content, 'test with unexpected success')
2256
# GZ 2011-05-18: Old versions of subunit treat unexpected success as a
2257
# success, if a min version check is added remove this
2258
from subunit import TestProtocolClient as _Client
2259
if _Client.addUnexpectedSuccess.__func__ is _Client.addSuccess.__func__:
2260
self.expectFailure('subunit treats "unexpectedSuccess"'
2261
' as a plain success',
2262
self.assertEqual, 1, len(result.unexpectedSuccesses))
2263
self.assertEqual(1, len(result.unexpectedSuccesses))
2264
test = result.unexpectedSuccesses[0]
2265
# RemotedTestCase doesn't preserve the "details"
2266
## self.assertTrue('log' in test.getDetails())
2268
def test_success_has_no_log(self):
2269
content, result = self.run_subunit_stream('test_success')
2270
self.assertEqual(1, result.testsRun)
2271
self.assertNotContainsRe(content, '(?m)^log$')
2272
self.assertNotContainsRe(content, 'this test succeeds')
1974
2275
class TestRunBzr(tests.TestCase):
2264
2565
class TestStartBzrSubProcess(tests.TestCase):
2566
"""Stub test start_bzr_subprocess."""
2266
def check_popen_state(self):
2267
"""Replace to make assertions when popen is called."""
2568
def _subprocess_log_cleanup(self):
2569
"""Inhibits the base version as we don't produce a log file."""
2269
2571
def _popen(self, *args, **kwargs):
2270
"""Record the command that is run, so that we can ensure it is correct"""
2572
"""Override the base version to record the command that is run.
2574
From there we can ensure it is correct without spawning a real process.
2271
2576
self.check_popen_state()
2272
2577
self._popen_args = args
2273
2578
self._popen_kwargs = kwargs
2274
2579
raise _DontSpawnProcess()
2581
def check_popen_state(self):
2582
"""Replace to make assertions when popen is called."""
2276
2584
def test_run_bzr_subprocess_no_plugins(self):
2277
2585
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2278
2586
command = self._popen_args[0]
2279
2587
self.assertEqual(sys.executable, command[0])
2280
self.assertEqual(self.get_bzr_path(), command[1])
2588
self.assertEqual(self.get_brz_path(), command[1])
2281
2589
self.assertEqual(['--no-plugins'], command[2:])
2283
2591
def test_allow_plugins(self):
2284
2592
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2286
2594
command = self._popen_args[0]
2287
2595
self.assertEqual([], command[2:])
2289
2597
def test_set_env(self):
2290
self.failIf('EXISTANT_ENV_VAR' in os.environ)
2598
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2291
2599
# set in the child
2292
2600
def check_environment():
2293
2601
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2294
2602
self.check_popen_state = check_environment
2295
2603
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2296
env_changes={'EXISTANT_ENV_VAR':'set variable'})
2604
env_changes={'EXISTANT_ENV_VAR':'set variable'})
2297
2605
# not set in theparent
2298
2606
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2300
2608
def test_run_bzr_subprocess_env_del(self):
2301
2609
"""run_bzr_subprocess can remove environment variables too."""
2302
self.failIf('EXISTANT_ENV_VAR' in os.environ)
2610
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2303
2611
def check_environment():
2304
2612
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2305
2613
os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2306
2614
self.check_popen_state = check_environment
2307
2615
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2308
env_changes={'EXISTANT_ENV_VAR':None})
2616
env_changes={'EXISTANT_ENV_VAR':None})
2309
2617
# Still set in parent
2310
2618
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2311
2619
del os.environ['EXISTANT_ENV_VAR']
2313
2621
def test_env_del_missing(self):
2314
self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2622
self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2315
2623
def check_environment():
2316
2624
self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2317
2625
self.check_popen_state = check_environment
2318
2626
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2319
env_changes={'NON_EXISTANT_ENV_VAR':None})
2627
env_changes={'NON_EXISTANT_ENV_VAR':None})
2321
2629
def test_working_dir(self):
2322
2630
"""Test that we can specify the working dir for the child"""
2350
2657
self.disable_missing_extensions_warning()
2351
2658
process = self.start_bzr_subprocess(['wait-until-signalled'],
2352
2659
skip_if_plan_to_signal=True)
2353
self.assertEqual('running\n', process.stdout.readline())
2660
self.assertEqual(b'running\n', process.stdout.readline())
2354
2661
result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
2356
self.assertEqual('', result[0])
2357
self.assertEqual('bzr: interrupted\n', result[1])
2360
class TestFeature(tests.TestCase):
2362
def test_caching(self):
2363
"""Feature._probe is called by the feature at most once."""
2364
class InstrumentedFeature(tests.Feature):
2366
super(InstrumentedFeature, self).__init__()
2369
self.calls.append('_probe')
2371
feature = InstrumentedFeature()
2373
self.assertEqual(['_probe'], feature.calls)
2375
self.assertEqual(['_probe'], feature.calls)
2377
def test_named_str(self):
2378
"""Feature.__str__ should thunk to feature_name()."""
2379
class NamedFeature(tests.Feature):
2380
def feature_name(self):
2382
feature = NamedFeature()
2383
self.assertEqual('symlinks', str(feature))
2385
def test_default_str(self):
2386
"""Feature.__str__ should default to __class__.__name__."""
2387
class NamedFeature(tests.Feature):
2389
feature = NamedFeature()
2390
self.assertEqual('NamedFeature', str(feature))
2393
class TestUnavailableFeature(tests.TestCase):
2395
def test_access_feature(self):
2396
feature = tests.Feature()
2397
exception = tests.UnavailableFeature(feature)
2398
self.assertIs(feature, exception.args[0])
2401
simple_thunk_feature = tests._CompatabilityThunkFeature(
2402
deprecated_in((2, 1, 0)),
2403
'bzrlib.tests.test_selftest',
2404
'simple_thunk_feature','UnicodeFilename',
2405
replacement_module='bzrlib.tests'
2408
class Test_CompatibilityFeature(tests.TestCase):
2410
def test_does_thunk(self):
2411
res = self.callDeprecated(
2412
['bzrlib.tests.test_selftest.simple_thunk_feature was deprecated'
2413
' in version 2.1.0. Use bzrlib.tests.UnicodeFilename instead.'],
2414
simple_thunk_feature.available)
2415
self.assertEqual(tests.UnicodeFilename.available(), res)
2418
class TestModuleAvailableFeature(tests.TestCase):
2420
def test_available_module(self):
2421
feature = tests.ModuleAvailableFeature('bzrlib.tests')
2422
self.assertEqual('bzrlib.tests', feature.module_name)
2423
self.assertEqual('bzrlib.tests', str(feature))
2424
self.assertTrue(feature.available())
2425
self.assertIs(tests, feature.module)
2427
def test_unavailable_module(self):
2428
feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
2429
self.assertEqual('bzrlib.no_such_module_exists', str(feature))
2430
self.assertFalse(feature.available())
2431
self.assertIs(None, feature.module)
2663
self.assertEqual(b'', result[0])
2664
self.assertEqual(b'brz: interrupted\n', result[1])
2434
2667
class TestSelftestFiltering(tests.TestCase):
2436
2669
def setUp(self):
2437
tests.TestCase.setUp(self)
2670
super(TestSelftestFiltering, self).setUp()
2438
2671
self.suite = TestUtil.TestSuite()
2439
2672
self.loader = TestUtil.TestLoader()
2440
2673
self.suite.addTest(self.loader.loadTestsFromModule(
2441
sys.modules['bzrlib.tests.test_selftest']))
2674
sys.modules['breezy.tests.test_selftest']))
2442
2675
self.all_names = _test_ids(self.suite)
2444
2677
def test_condition_id_re(self):
2445
test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2678
test_name = ('breezy.tests.test_selftest.TestSelftestFiltering.'
2446
2679
'test_condition_id_re')
2447
2680
filtered_suite = tests.filter_suite_by_condition(
2448
2681
self.suite, tests.condition_id_re('test_condition_id_re'))
2449
2682
self.assertEqual([test_name], _test_ids(filtered_suite))
2451
2684
def test_condition_id_in_list(self):
2452
test_names = ['bzrlib.tests.test_selftest.TestSelftestFiltering.'
2685
test_names = ['breezy.tests.test_selftest.TestSelftestFiltering.'
2453
2686
'test_condition_id_in_list']
2454
2687
id_list = tests.TestIdList(test_names)
2455
2688
filtered_suite = tests.filter_suite_by_condition(
2953
3191
def test_predefined_prefixes(self):
2954
3192
tpr = tests.test_prefix_alias_registry
2955
self.assertEquals('bzrlib', tpr.resolve_alias('bzrlib'))
2956
self.assertEquals('bzrlib.doc', tpr.resolve_alias('bd'))
2957
self.assertEquals('bzrlib.utils', tpr.resolve_alias('bu'))
2958
self.assertEquals('bzrlib.tests', tpr.resolve_alias('bt'))
2959
self.assertEquals('bzrlib.tests.blackbox', tpr.resolve_alias('bb'))
2960
self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3193
self.assertEqual('breezy', tpr.resolve_alias('breezy'))
3194
self.assertEqual('breezy.doc', tpr.resolve_alias('bd'))
3195
self.assertEqual('breezy.utils', tpr.resolve_alias('bu'))
3196
self.assertEqual('breezy.tests', tpr.resolve_alias('bt'))
3197
self.assertEqual('breezy.tests.blackbox', tpr.resolve_alias('bb'))
3198
self.assertEqual('breezy.plugins', tpr.resolve_alias('bp'))
3201
class TestThreadLeakDetection(tests.TestCase):
3202
"""Ensure when tests leak threads we detect and report it"""
3204
class LeakRecordingResult(tests.ExtendedTestResult):
3206
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3208
def _report_thread_leak(self, test, leaks, alive):
3209
self.leaks.append((test, leaks))
3211
def test_testcase_without_addCleanups(self):
3212
"""Check old TestCase instances don't break with leak detection"""
3213
class Test(unittest.TestCase):
3216
result = self.LeakRecordingResult()
3218
result.startTestRun()
3220
result.stopTestRun()
3221
self.assertEqual(result._tests_leaking_threads_count, 0)
3222
self.assertEqual(result.leaks, [])
3224
def test_thread_leak(self):
3225
"""Ensure a thread that outlives the running of a test is reported
3227
Uses a thread that blocks on an event, and is started by the inner
3228
test case. As the thread outlives the inner case's run, it should be
3229
detected as a leak, but the event is then set so that the thread can
3230
be safely joined in cleanup so it's not leaked for real.
3232
event = threading.Event()
3233
thread = threading.Thread(name="Leaker", target=event.wait)
3234
class Test(tests.TestCase):
3235
def test_leak(self):
3237
result = self.LeakRecordingResult()
3238
test = Test("test_leak")
3239
self.addCleanup(thread.join)
3240
self.addCleanup(event.set)
3241
result.startTestRun()
3243
result.stopTestRun()
3244
self.assertEqual(result._tests_leaking_threads_count, 1)
3245
self.assertEqual(result._first_thread_leaker_id, test.id())
3246
self.assertEqual(result.leaks, [(test, {thread})])
3247
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3249
def test_multiple_leaks(self):
3250
"""Check multiple leaks are blamed on the test cases at fault
3252
Same concept as the previous test, but has one inner test method that
3253
leaks two threads, and one that doesn't leak at all.
3255
event = threading.Event()
3256
thread_a = threading.Thread(name="LeakerA", target=event.wait)
3257
thread_b = threading.Thread(name="LeakerB", target=event.wait)
3258
thread_c = threading.Thread(name="LeakerC", target=event.wait)
3259
class Test(tests.TestCase):
3260
def test_first_leak(self):
3262
def test_second_no_leak(self):
3264
def test_third_leak(self):
3267
result = self.LeakRecordingResult()
3268
first_test = Test("test_first_leak")
3269
third_test = Test("test_third_leak")
3270
self.addCleanup(thread_a.join)
3271
self.addCleanup(thread_b.join)
3272
self.addCleanup(thread_c.join)
3273
self.addCleanup(event.set)
3274
result.startTestRun()
3276
[first_test, Test("test_second_no_leak"), third_test]
3278
result.stopTestRun()
3279
self.assertEqual(result._tests_leaking_threads_count, 2)
3280
self.assertEqual(result._first_thread_leaker_id, first_test.id())
3281
self.assertEqual(result.leaks, [
3282
(first_test, {thread_b}),
3283
(third_test, {thread_a, thread_c})])
3284
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3287
class TestPostMortemDebugging(tests.TestCase):
3288
"""Check post mortem debugging works when tests fail or error"""
3290
class TracebackRecordingResult(tests.ExtendedTestResult):
3292
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3293
self.postcode = None
3294
def _post_mortem(self, tb=None):
3295
"""Record the code object at the end of the current traceback"""
3296
tb = tb or sys.exc_info()[2]
3299
while next is not None:
3302
self.postcode = tb.tb_frame.f_code
3303
def report_error(self, test, err):
3305
def report_failure(self, test, err):
3308
def test_location_unittest_error(self):
3309
"""Needs right post mortem traceback with erroring unittest case"""
3310
class Test(unittest.TestCase):
3313
result = self.TracebackRecordingResult()
3315
self.assertEqual(result.postcode, Test.runTest.__code__)
3317
def test_location_unittest_failure(self):
3318
"""Needs right post mortem traceback with failing unittest case"""
3319
class Test(unittest.TestCase):
3321
raise self.failureException
3322
result = self.TracebackRecordingResult()
3324
self.assertEqual(result.postcode, Test.runTest.__code__)
3326
def test_location_bt_error(self):
3327
"""Needs right post mortem traceback with erroring breezy.tests case"""
3328
class Test(tests.TestCase):
3329
def test_error(self):
3331
result = self.TracebackRecordingResult()
3332
Test("test_error").run(result)
3333
self.assertEqual(result.postcode, Test.test_error.__code__)
3335
def test_location_bt_failure(self):
3336
"""Needs right post mortem traceback with failing breezy.tests case"""
3337
class Test(tests.TestCase):
3338
def test_failure(self):
3339
raise self.failureException
3340
result = self.TracebackRecordingResult()
3341
Test("test_failure").run(result)
3342
self.assertEqual(result.postcode, Test.test_failure.__code__)
3344
def test_env_var_triggers_post_mortem(self):
3345
"""Check pdb.post_mortem is called iff BRZ_TEST_PDB is set"""
3347
result = tests.ExtendedTestResult(StringIO(), 0, 1)
3348
post_mortem_calls = []
3349
self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3350
self.overrideEnv('BRZ_TEST_PDB', None)
3351
result._post_mortem(1)
3352
self.overrideEnv('BRZ_TEST_PDB', 'on')
3353
result._post_mortem(2)
3354
self.assertEqual([2], post_mortem_calls)
2963
3357
class TestRunSuite(tests.TestCase):
2976
3370
self.verbosity)
2977
3371
tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2978
3372
self.assertLength(1, calls)
3375
class _Selftest(object):
3376
"""Mixin for tests needing full selftest output"""
3378
def _inject_stream_into_subunit(self, stream):
3379
"""To be overridden by subclasses that run tests out of process"""
3381
def _run_selftest(self, **kwargs):
3383
self._inject_stream_into_subunit(sio)
3384
tests.selftest(stream=sio, stop_on_failure=False, **kwargs)
3385
return sio.getvalue()
3388
class _ForkedSelftest(_Selftest):
3389
"""Mixin for tests needing full selftest output with forked children"""
3391
_test_needs_features = [features.subunit]
3393
def _inject_stream_into_subunit(self, stream):
3394
"""Monkey-patch subunit so the extra output goes to stream not stdout
3396
Some APIs need rewriting so this kind of bogus hackery can be replaced
3397
by passing the stream param from run_tests down into ProtocolTestCase.
3399
from subunit import ProtocolTestCase
3400
_original_init = ProtocolTestCase.__init__
3401
def _init_with_passthrough(self, *args, **kwargs):
3402
_original_init(self, *args, **kwargs)
3403
self._passthrough = stream
3404
self.overrideAttr(ProtocolTestCase, "__init__", _init_with_passthrough)
3406
def _run_selftest(self, **kwargs):
3407
# GZ 2011-05-26: Add a PosixSystem feature so this check can go away
3408
if getattr(os, "fork", None) is None:
3409
raise tests.TestNotApplicable("Platform doesn't support forking")
3410
# Make sure the fork code is actually invoked by claiming two cores
3411
self.overrideAttr(osutils, "local_concurrency", lambda: 2)
3412
kwargs.setdefault("suite_decorators", []).append(tests.fork_decorator)
3413
return super(_ForkedSelftest, self)._run_selftest(**kwargs)
3416
class TestParallelFork(_ForkedSelftest, tests.TestCase):
3417
"""Check operation of --parallel=fork selftest option"""
3419
def test_error_in_child_during_fork(self):
3420
"""Error in a forked child during test setup should get reported"""
3421
class Test(tests.TestCase):
3422
def testMethod(self):
3424
# We don't care what, just break something that a child will run
3425
self.overrideAttr(tests, "workaround_zealous_crypto_random", None)
3426
out = self._run_selftest(test_suite_factory=Test)
3427
# Lines from the tracebacks of the two child processes may be mixed
3428
# together due to the way subunit parses and forwards the streams,
3429
# so permit extra lines between each part of the error output.
3430
self.assertContainsRe(out,
3433
".+ in fork_for_tests\n"
3435
"\\s*workaround_zealous_crypto_random\\(\\)\n"
3440
class TestUncollectedWarnings(_Selftest, tests.TestCase):
3441
"""Check a test case still alive after being run emits a warning"""
3443
class Test(tests.TestCase):
3444
def test_pass(self):
3446
def test_self_ref(self):
3447
self.also_self = self.test_self_ref
3448
def test_skip(self):
3449
self.skipTest("Don't need")
3451
def _get_suite(self):
3452
return TestUtil.TestSuite([
3453
self.Test("test_pass"),
3454
self.Test("test_self_ref"),
3455
self.Test("test_skip"),
3458
def _run_selftest_with_suite(self, **kwargs):
3459
old_flags = tests.selftest_debug_flags
3460
tests.selftest_debug_flags = old_flags.union(["uncollected_cases"])
3461
gc_on = gc.isenabled()
3465
output = self._run_selftest(test_suite_factory=self._get_suite,
3470
tests.selftest_debug_flags = old_flags
3471
self.assertNotContainsRe(output, "Uncollected test case.*test_pass")
3472
self.assertContainsRe(output, "Uncollected test case.*test_self_ref")
3475
def test_testsuite(self):
3476
self._run_selftest_with_suite()
3478
def test_pattern(self):
3479
out = self._run_selftest_with_suite(pattern="test_(?:pass|self_ref)$")
3480
self.assertNotContainsRe(out, "test_skip")
3482
def test_exclude_pattern(self):
3483
out = self._run_selftest_with_suite(exclude_pattern="test_skip$")
3484
self.assertNotContainsRe(out, "test_skip")
3486
def test_random_seed(self):
3487
self._run_selftest_with_suite(random_seed="now")
3489
def test_matching_tests_first(self):
3490
self._run_selftest_with_suite(matching_tests_first=True,
3491
pattern="test_self_ref$")
3493
def test_starting_with_and_exclude(self):
3494
out = self._run_selftest_with_suite(starting_with=["bt."],
3495
exclude_pattern="test_skip$")
3496
self.assertNotContainsRe(out, "test_skip")
3498
def test_additonal_decorator(self):
3499
out = self._run_selftest_with_suite(
3500
suite_decorators=[tests.TestDecorator])
3503
class TestUncollectedWarningsSubunit(TestUncollectedWarnings):
3504
"""Check warnings from tests staying alive are emitted with subunit"""
3506
_test_needs_features = [features.subunit]
3508
def _run_selftest_with_suite(self, **kwargs):
3509
return TestUncollectedWarnings._run_selftest_with_suite(
3510
self, runner_class=tests.SubUnitBzrRunnerv1, **kwargs)
3513
class TestUncollectedWarningsForked(_ForkedSelftest, TestUncollectedWarnings):
3514
"""Check warnings from tests staying alive are emitted when forking"""
3517
class TestEnvironHandling(tests.TestCase):
3519
def test_overrideEnv_None_called_twice_doesnt_leak(self):
3520
self.assertFalse('MYVAR' in os.environ)
3521
self.overrideEnv('MYVAR', '42')
3522
# We use an embedded test to make sure we fix the _captureVar bug
3523
class Test(tests.TestCase):
3525
# The first call save the 42 value
3526
self.overrideEnv('MYVAR', None)
3527
self.assertEqual(None, os.environ.get('MYVAR'))
3528
# Make sure we can call it twice
3529
self.overrideEnv('MYVAR', None)
3530
self.assertEqual(None, os.environ.get('MYVAR'))
3532
result = tests.TextTestResult(output, 0, 1)
3533
Test('test_me').run(result)
3534
if not result.wasStrictlySuccessful():
3535
self.fail(output.getvalue())
3536
# We get our value back
3537
self.assertEqual('42', os.environ.get('MYVAR'))
3540
class TestIsolatedEnv(tests.TestCase):
3541
"""Test isolating tests from os.environ.
3543
Since we use tests that are already isolated from os.environ a bit of care
3544
should be taken when designing the tests to avoid bootstrap side-effects.
3545
The tests start an already clean os.environ which allow doing valid
3546
assertions about which variables are present or not and design tests around
3550
class ScratchMonkey(tests.TestCase):
3555
def test_basics(self):
3556
# Make sure we know the definition of BRZ_HOME: not part of os.environ
3557
# for tests.TestCase.
3558
self.assertTrue('BRZ_HOME' in tests.isolated_environ)
3559
self.assertEqual(None, tests.isolated_environ['BRZ_HOME'])
3560
# Being part of isolated_environ, BRZ_HOME should not appear here
3561
self.assertFalse('BRZ_HOME' in os.environ)
3562
# Make sure we know the definition of LINES: part of os.environ for
3564
self.assertTrue('LINES' in tests.isolated_environ)
3565
self.assertEqual('25', tests.isolated_environ['LINES'])
3566
self.assertEqual('25', os.environ['LINES'])
3568
def test_injecting_unknown_variable(self):
3569
# BRZ_HOME is known to be absent from os.environ
3570
test = self.ScratchMonkey('test_me')
3571
tests.override_os_environ(test, {'BRZ_HOME': 'foo'})
3572
self.assertEqual('foo', os.environ['BRZ_HOME'])
3573
tests.restore_os_environ(test)
3574
self.assertFalse('BRZ_HOME' in os.environ)
3576
def test_injecting_known_variable(self):
3577
test = self.ScratchMonkey('test_me')
3578
# LINES is known to be present in os.environ
3579
tests.override_os_environ(test, {'LINES': '42'})
3580
self.assertEqual('42', os.environ['LINES'])
3581
tests.restore_os_environ(test)
3582
self.assertEqual('25', os.environ['LINES'])
3584
def test_deleting_variable(self):
3585
test = self.ScratchMonkey('test_me')
3586
# LINES is known to be present in os.environ
3587
tests.override_os_environ(test, {'LINES': None})
3588
self.assertTrue('LINES' not in os.environ)
3589
tests.restore_os_environ(test)
3590
self.assertEqual('25', os.environ['LINES'])
3593
class TestDocTestSuiteIsolation(tests.TestCase):
3594
"""Test that `tests.DocTestSuite` isolates doc tests from os.environ.
3596
Since tests.TestCase alreay provides an isolation from os.environ, we use
3597
the clean environment as a base for testing. To precisely capture the
3598
isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
3601
We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
3602
not `os.environ` so each test overrides it to suit its needs.
3606
def get_doctest_suite_for_string(self, klass, string):
3607
class Finder(doctest.DocTestFinder):
3609
def find(*args, **kwargs):
3610
test = doctest.DocTestParser().get_doctest(
3611
string, {}, 'foo', 'foo.py', 0)
3614
suite = klass(test_finder=Finder())
3617
def run_doctest_suite_for_string(self, klass, string):
3618
suite = self.get_doctest_suite_for_string(klass, string)
3620
result = tests.TextTestResult(output, 0, 1)
3622
return result, output
3624
def assertDocTestStringSucceds(self, klass, string):
3625
result, output = self.run_doctest_suite_for_string(klass, string)
3626
if not result.wasStrictlySuccessful():
3627
self.fail(output.getvalue())
3629
def assertDocTestStringFails(self, klass, string):
3630
result, output = self.run_doctest_suite_for_string(klass, string)
3631
if result.wasStrictlySuccessful():
3632
self.fail(output.getvalue())
3634
def test_injected_variable(self):
3635
self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
3638
>>> os.environ['LINES']
3641
# doctest.DocTestSuite fails as it sees '25'
3642
self.assertDocTestStringFails(doctest.DocTestSuite, test)
3643
# tests.DocTestSuite sees '42'
3644
self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3646
def test_deleted_variable(self):
3647
self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
3650
>>> os.environ.get('LINES')
3652
# doctest.DocTestSuite fails as it sees '25'
3653
self.assertDocTestStringFails(doctest.DocTestSuite, test)
3654
# tests.DocTestSuite sees None
3655
self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3658
class TestSelftestExcludePatterns(tests.TestCase):
3661
super(TestSelftestExcludePatterns, self).setUp()
3662
self.overrideAttr(tests, 'test_suite', self.suite_factory)
3664
def suite_factory(self, keep_only=None, starting_with=None):
3665
"""A test suite factory with only a few tests."""
3666
class Test(tests.TestCase):
3668
# We don't need the full class path
3669
return self._testMethodName
3676
return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
3678
def assertTestList(self, expected, *selftest_args):
3679
# We rely on setUp installing the right test suite factory so we can
3680
# test at the command level without loading the whole test suite
3681
out, err = self.run_bzr(('selftest', '--list') + selftest_args)
3682
actual = out.splitlines()
3683
self.assertEqual(expected, actual)
3685
def test_full_list(self):
3686
self.assertTestList(['a', 'b', 'c'])
3688
def test_single_exclude(self):
3689
self.assertTestList(['b', 'c'], '-x', 'a')
3691
def test_mutiple_excludes(self):
3692
self.assertTestList(['c'], '-x', 'a', '-x', 'b')
3695
class TestCounterHooks(tests.TestCase, SelfTestHelper):
3697
_test_needs_features = [features.subunit]
3700
super(TestCounterHooks, self).setUp()
3701
class Test(tests.TestCase):
3704
super(Test, self).setUp()
3705
self.hooks = hooks.Hooks()
3706
self.hooks.add_hook('myhook', 'Foo bar blah', (2, 4))
3707
self.install_counter_hook(self.hooks, 'myhook')
3712
def run_hook_once(self):
3713
for hook in self.hooks['myhook']:
3716
self.test_class = Test
3718
def assertHookCalls(self, expected_calls, test_name):
3719
test = self.test_class(test_name)
3720
result = unittest.TestResult()
3722
self.assertTrue(hasattr(test, '_counters'))
3723
self.assertTrue('myhook' in test._counters)
3724
self.assertEqual(expected_calls, test._counters['myhook'])
3726
def test_no_hook(self):
3727
self.assertHookCalls(0, 'no_hook')
3729
def test_run_hook_once(self):
3730
tt = features.testtools
3731
if tt.module.__version__ < (0, 9, 8):
3732
raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
3733
self.assertHookCalls(1, 'run_hook_once')