350
392
'transport_readonly_server': 'b',
351
393
'transport_server': 'a',
352
394
'workingtree_format': formats[1]})],
356
class TestTreeScenarios(tests.TestCase):
358
def test_scenarios(self):
359
# the tree implementation scenario generator is meant to setup one
360
# instance for each working tree format, and one additional instance
361
# 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.
366
from bzrlib.tests.per_tree import (
367
_dirstate_tree_from_workingtree,
398
class TestTreeProviderAdapter(TestCase):
399
"""Test the setup of tree_implementation tests."""
401
def test_adapted_tests(self):
402
# the tree implementation adapter is meant to setup one instance for
403
# each working tree format, and one additional instance that will
404
# use the default wt format, but create a revision tree for the tests.
405
# this means that the wt ones should have the workingtree_to_test_tree
406
# attribute set to 'return_parameter' and the revision one set to
407
# revision_tree_from_workingtree.
409
from bzrlib.tests.tree_implementations import (
410
TreeTestProviderAdapter,
371
411
return_parameter,
372
412
revision_tree_from_workingtree
414
input_test = TestTreeProviderAdapter(
415
"test_adapted_tests")
376
418
formats = [workingtree.WorkingTreeFormat2(),
377
419
workingtree.WorkingTreeFormat3(),]
378
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()
383
expected_scenarios = [
384
('WorkingTreeFormat2',
385
{'bzrdir_format': formats[0]._matchingbzrdir,
386
'transport_readonly_server': 'b',
387
'transport_server': 'a',
388
'workingtree_format': formats[0],
389
'_workingtree_to_test_tree': return_parameter,
391
('WorkingTreeFormat3',
392
{'bzrdir_format': formats[1]._matchingbzrdir,
393
'transport_readonly_server': 'b',
394
'transport_server': 'a',
395
'workingtree_format': formats[1],
396
'_workingtree_to_test_tree': return_parameter,
399
{'_workingtree_to_test_tree': revision_tree_from_workingtree,
400
'bzrdir_format': default_wt_format._matchingbzrdir,
401
'transport_readonly_server': 'b',
402
'transport_server': 'a',
403
'workingtree_format': default_wt_format,
405
('DirStateRevisionTree,WT4',
406
{'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
407
'bzrdir_format': wt4_format._matchingbzrdir,
408
'transport_readonly_server': 'b',
409
'transport_server': 'a',
410
'workingtree_format': wt4_format,
412
('DirStateRevisionTree,WT5',
413
{'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
414
'bzrdir_format': wt5_format._matchingbzrdir,
415
'transport_readonly_server': 'b',
416
'transport_server': 'a',
417
'workingtree_format': wt5_format,
420
{'_workingtree_to_test_tree': preview_tree_pre,
421
'bzrdir_format': default_wt_format._matchingbzrdir,
422
'transport_readonly_server': 'b',
423
'transport_server': 'a',
424
'workingtree_format': default_wt_format}),
426
{'_workingtree_to_test_tree': preview_tree_post,
427
'bzrdir_format': default_wt_format._matchingbzrdir,
428
'transport_readonly_server': 'b',
429
'transport_server': 'a',
430
'workingtree_format': default_wt_format}),
432
self.assertEqual(expected_scenarios, scenarios)
435
class TestInterTreeScenarios(tests.TestCase):
420
adapter = TreeTestProviderAdapter(server1, server2, formats)
421
suite = adapter.adapt(input_test)
422
tests = list(iter(suite))
423
# XXX We should not have tests fail as we add more scenarios
425
self.assertEqual(7, len(tests))
426
# this must match the default format setp up in
427
# TreeTestProviderAdapter.adapt
428
default_format = workingtree.WorkingTreeFormat3
429
self.assertEqual(tests[0].workingtree_format, formats[0])
430
self.assertEqual(tests[0].bzrdir_format, formats[0]._matchingbzrdir)
431
self.assertEqual(tests[0].transport_server, server1)
432
self.assertEqual(tests[0].transport_readonly_server, server2)
433
self.assertEqual(tests[0]._workingtree_to_test_tree, return_parameter)
434
self.assertEqual(tests[1].workingtree_format, formats[1])
435
self.assertEqual(tests[1].bzrdir_format, formats[1]._matchingbzrdir)
436
self.assertEqual(tests[1].transport_server, server1)
437
self.assertEqual(tests[1].transport_readonly_server, server2)
438
self.assertEqual(tests[1]._workingtree_to_test_tree, return_parameter)
439
self.assertIsInstance(tests[2].workingtree_format, default_format)
440
#self.assertEqual(tests[2].bzrdir_format,
441
# default_format._matchingbzrdir)
442
self.assertEqual(tests[2].transport_server, server1)
443
self.assertEqual(tests[2].transport_readonly_server, server2)
444
self.assertEqual(tests[2]._workingtree_to_test_tree,
445
revision_tree_from_workingtree)
448
class TestInterTreeProviderAdapter(TestCase):
436
449
"""A group of tests that test the InterTreeTestAdapter."""
438
def test_scenarios(self):
451
def test_adapted_tests(self):
439
452
# check that constructor parameters are passed through to the adapted
441
454
# for InterTree tests we want the machinery to bring up two trees in
1193
def _patch_get_bzr_source_tree(self):
1194
# Reading from the actual source tree breaks isolation, but we don't
1195
# want to assume that thats *all* that would happen.
1196
self._get_source_tree_calls = []
1198
self._get_source_tree_calls.append("called")
1200
self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', new_get)
1202
1207
def test_bench_history(self):
1203
# tests that the running the benchmark passes bench_history into
1204
# the test result object. We can tell that happens if
1205
# _get_bzr_source_tree is called.
1206
self._patch_get_bzr_source_tree()
1208
# tests that the running the benchmark produces a history file
1209
# containing a timestamp and the revision id of the bzrlib source which
1211
workingtree = _get_bzr_source_tree()
1207
1212
test = TestRunner('dummy_test')
1208
1213
output = StringIO()
1209
runner = tests.TextTestRunner(stream=self._log_file,
1210
bench_history=output)
1214
runner = TextTestRunner(stream=self._log_file, bench_history=output)
1211
1215
result = self.run_test_runner(runner, test)
1212
1216
output_string = output.getvalue()
1213
1217
self.assertContainsRe(output_string, "--date [0-9.]+")
1214
self.assertLength(1, self._get_source_tree_calls)
1216
def test_startTestRun(self):
1217
"""run should call result.startTestRun()"""
1219
class LoggingDecorator(tests.ForwardingResult):
1220
def startTestRun(self):
1221
tests.ForwardingResult.startTestRun(self)
1222
calls.append('startTestRun')
1223
test = unittest.FunctionTestCase(lambda:None)
1225
runner = tests.TextTestRunner(stream=stream,
1226
result_decorators=[LoggingDecorator])
1227
result = self.run_test_runner(runner, test)
1228
self.assertLength(1, calls)
1230
def test_stopTestRun(self):
1231
"""run should call result.stopTestRun()"""
1233
class LoggingDecorator(tests.ForwardingResult):
1234
def stopTestRun(self):
1235
tests.ForwardingResult.stopTestRun(self)
1236
calls.append('stopTestRun')
1237
test = unittest.FunctionTestCase(lambda:None)
1239
runner = tests.TextTestRunner(stream=stream,
1240
result_decorators=[LoggingDecorator])
1241
result = self.run_test_runner(runner, test)
1242
self.assertLength(1, calls)
1245
class SampleTestCase(tests.TestCase):
1218
if workingtree is not None:
1219
revision_id = workingtree.get_parent_ids()[0]
1220
self.assertEndsWith(output_string.rstrip(), revision_id)
1222
def assertLogDeleted(self, test):
1223
log = test._get_log()
1224
self.assertEqual("DELETED log file to reduce memory footprint", log)
1225
self.assertEqual('', test._log_contents)
1226
self.assertIs(None, test._log_file_name)
1228
def test_success_log_deleted(self):
1229
"""Successful tests have their log deleted"""
1231
class LogTester(TestCase):
1233
def test_success(self):
1234
self.log('this will be removed\n')
1236
sio = cStringIO.StringIO()
1237
runner = TextTestRunner(stream=sio)
1238
test = LogTester('test_success')
1239
result = self.run_test_runner(runner, test)
1241
self.assertLogDeleted(test)
1243
def test_skipped_log_deleted(self):
1244
"""Skipped tests have their log deleted"""
1246
class LogTester(TestCase):
1248
def test_skipped(self):
1249
self.log('this will be removed\n')
1250
raise tests.TestSkipped()
1252
sio = cStringIO.StringIO()
1253
runner = TextTestRunner(stream=sio)
1254
test = LogTester('test_skipped')
1255
result = self.run_test_runner(runner, test)
1257
self.assertLogDeleted(test)
1259
def test_not_aplicable_log_deleted(self):
1260
"""Not applicable tests have their log deleted"""
1262
class LogTester(TestCase):
1264
def test_not_applicable(self):
1265
self.log('this will be removed\n')
1266
raise tests.TestNotApplicable()
1268
sio = cStringIO.StringIO()
1269
runner = TextTestRunner(stream=sio)
1270
test = LogTester('test_not_applicable')
1271
result = self.run_test_runner(runner, test)
1273
self.assertLogDeleted(test)
1275
def test_known_failure_log_deleted(self):
1276
"""Know failure tests have their log deleted"""
1278
class LogTester(TestCase):
1280
def test_known_failure(self):
1281
self.log('this will be removed\n')
1282
raise tests.KnownFailure()
1284
sio = cStringIO.StringIO()
1285
runner = TextTestRunner(stream=sio)
1286
test = LogTester('test_known_failure')
1287
result = self.run_test_runner(runner, test)
1289
self.assertLogDeleted(test)
1291
def test_fail_log_kept(self):
1292
"""Failed tests have their log kept"""
1294
class LogTester(TestCase):
1296
def test_fail(self):
1297
self.log('this will be kept\n')
1298
self.fail('this test fails')
1300
sio = cStringIO.StringIO()
1301
runner = TextTestRunner(stream=sio)
1302
test = LogTester('test_fail')
1303
result = self.run_test_runner(runner, test)
1305
text = sio.getvalue()
1306
self.assertContainsRe(text, 'this will be kept')
1307
self.assertContainsRe(text, 'this test fails')
1309
log = test._get_log()
1310
self.assertContainsRe(log, 'this will be kept')
1311
self.assertEqual(log, test._log_contents)
1313
def test_error_log_kept(self):
1314
"""Tests with errors have their log kept"""
1316
class LogTester(TestCase):
1318
def test_error(self):
1319
self.log('this will be kept\n')
1320
raise ValueError('random exception raised')
1322
sio = cStringIO.StringIO()
1323
runner = TextTestRunner(stream=sio)
1324
test = LogTester('test_error')
1325
result = self.run_test_runner(runner, test)
1327
text = sio.getvalue()
1328
self.assertContainsRe(text, 'this will be kept')
1329
self.assertContainsRe(text, 'random exception raised')
1331
log = test._get_log()
1332
self.assertContainsRe(log, 'this will be kept')
1333
self.assertEqual(log, test._log_contents)
1336
class SampleTestCase(TestCase):
1247
1338
def _test_pass(self):
1677
1618
def sample_normal_method(self):
1678
1619
"""A undeprecated method."""
1680
@deprecated_method(deprecated_in((0, 10, 0)))
1621
@symbol_versioning.deprecated_method(zero_ten)
1681
1622
def sample_nested_deprecation(self):
1682
1623
return sample_deprecated_function()
1685
class TestExtraAssertions(tests.TestCase):
1626
class TestExtraAssertions(TestCase):
1686
1627
"""Tests for new test assertions in bzrlib test suite"""
1688
1629
def test_assert_isinstance(self):
1689
1630
self.assertIsInstance(2, int)
1690
1631
self.assertIsInstance(u'', basestring)
1691
e = self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1692
self.assertEquals(str(e),
1693
"None is an instance of <type 'NoneType'> rather than <type 'int'>")
1632
self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1694
1633
self.assertRaises(AssertionError, self.assertIsInstance, 23.3, int)
1695
e = self.assertRaises(AssertionError,
1696
self.assertIsInstance, None, int, "it's just not")
1697
self.assertEquals(str(e),
1698
"None is an instance of <type 'NoneType'> rather than <type 'int'>"
1701
1635
def test_assertEndsWith(self):
1702
1636
self.assertEndsWith('foo', 'oo')
1703
1637
self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1705
def test_assertEqualDiff(self):
1706
e = self.assertRaises(AssertionError,
1707
self.assertEqualDiff, '', '\n')
1708
self.assertEquals(str(e),
1709
# Don't blink ! The '+' applies to the second string
1710
'first string is missing a final newline.\n+ \n')
1711
e = self.assertRaises(AssertionError,
1712
self.assertEqualDiff, '\n', '')
1713
self.assertEquals(str(e),
1714
# Don't blink ! The '-' applies to the second string
1715
'second string is missing a final newline.\n- \n')
1718
class TestDeprecations(tests.TestCase):
1720
1639
def test_applyDeprecated_not_deprecated(self):
1721
1640
sample_object = ApplyDeprecatedHelper()
1722
1641
# calling an undeprecated callable raises an assertion
1723
self.assertRaises(AssertionError, self.applyDeprecated,
1724
deprecated_in((0, 11, 0)),
1642
self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
1725
1643
sample_object.sample_normal_method)
1726
self.assertRaises(AssertionError, self.applyDeprecated,
1727
deprecated_in((0, 11, 0)),
1644
self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
1728
1645
sample_undeprecated_function, "a param value")
1729
1646
# calling a deprecated callable (function or method) with the wrong
1730
1647
# expected deprecation fails.
1731
self.assertRaises(AssertionError, self.applyDeprecated,
1732
deprecated_in((0, 10, 0)),
1648
self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
1733
1649
sample_object.sample_deprecated_method, "a param value")
1734
self.assertRaises(AssertionError, self.applyDeprecated,
1735
deprecated_in((0, 10, 0)),
1650
self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
1736
1651
sample_deprecated_function)
1737
1652
# calling a deprecated callable (function or method) with the right
1738
1653
# expected deprecation returns the functions result.
1739
self.assertEqual("a param value",
1740
self.applyDeprecated(deprecated_in((0, 11, 0)),
1654
self.assertEqual("a param value", self.applyDeprecated(zero_eleven,
1741
1655
sample_object.sample_deprecated_method, "a param value"))
1742
self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 11, 0)),
1656
self.assertEqual(2, self.applyDeprecated(zero_eleven,
1743
1657
sample_deprecated_function))
1744
1658
# calling a nested deprecation with the wrong deprecation version
1745
# fails even if a deeper nested function was deprecated with the
1659
# fails even if a deeper nested function was deprecated with the
1746
1660
# supplied version.
1747
1661
self.assertRaises(AssertionError, self.applyDeprecated,
1748
deprecated_in((0, 11, 0)), sample_object.sample_nested_deprecation)
1662
zero_eleven, sample_object.sample_nested_deprecation)
1749
1663
# calling a nested deprecation with the right deprecation value
1750
1664
# returns the calls result.
1751
self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 10, 0)),
1665
self.assertEqual(2, self.applyDeprecated(zero_ten,
1752
1666
sample_object.sample_nested_deprecation))
1754
1668
def test_callDeprecated(self):
1755
1669
def testfunc(be_deprecated, result=None):
1756
1670
if be_deprecated is True:
1757
symbol_versioning.warn('i am deprecated', DeprecationWarning,
1671
symbol_versioning.warn('i am deprecated', DeprecationWarning,
1760
1674
result = self.callDeprecated(['i am deprecated'], testfunc, True)
1799
1713
tree = self.make_branch_and_memory_tree('a')
1800
1714
self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1802
def test_make_tree_for_local_vfs_backed_transport(self):
1803
# make_branch_and_tree has to use local branch and repositories
1804
# when the vfs transport and local disk are colocated, even if
1805
# a different transport is in use for url generation.
1806
self.transport_server = test_server.FakeVFATServer
1807
self.assertFalse(self.get_url('t1').startswith('file://'))
1717
class TestSFTPMakeBranchAndTree(TestCaseWithSFTPServer):
1719
def test_make_tree_for_sftp_branch(self):
1720
"""Transports backed by local directories create local trees."""
1808
1722
tree = self.make_branch_and_tree('t1')
1809
1723
base = tree.bzrdir.root_transport.base
1810
self.assertStartsWith(base, 'file://')
1724
self.failIf(base.startswith('sftp'),
1725
'base %r is on sftp but should be local' % base)
1811
1726
self.assertEquals(tree.bzrdir.root_transport,
1812
1727
tree.branch.bzrdir.root_transport)
1813
1728
self.assertEquals(tree.bzrdir.root_transport,
1814
1729
tree.branch.repository.bzrdir.root_transport)
1817
class SelfTestHelper:
1819
def run_selftest(self, **kwargs):
1820
"""Run selftest returning its output."""
1822
old_transport = bzrlib.tests.default_transport
1823
old_root = tests.TestCaseWithMemoryTransport.TEST_ROOT
1824
tests.TestCaseWithMemoryTransport.TEST_ROOT = None
1826
self.assertEqual(True, tests.selftest(stream=output, **kwargs))
1828
bzrlib.tests.default_transport = old_transport
1829
tests.TestCaseWithMemoryTransport.TEST_ROOT = old_root
1834
class TestSelftest(tests.TestCase, SelfTestHelper):
1732
class TestSelftest(TestCase):
1835
1733
"""Tests of bzrlib.tests.selftest."""
1837
1735
def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
1838
1736
factory_called = []
1840
1738
factory_called.append(True)
1841
return TestUtil.TestSuite()
1842
1740
out = StringIO()
1843
1741
err = StringIO()
1844
self.apply_redirected(out, err, None, bzrlib.tests.selftest,
1742
self.apply_redirected(out, err, None, bzrlib.tests.selftest,
1845
1743
test_suite_factory=factory)
1846
1744
self.assertEqual([True], factory_called)
1849
"""A test suite factory."""
1850
class Test(tests.TestCase):
1857
return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
1859
def test_list_only(self):
1860
output = self.run_selftest(test_suite_factory=self.factory,
1862
self.assertEqual(3, len(output.readlines()))
1864
def test_list_only_filtered(self):
1865
output = self.run_selftest(test_suite_factory=self.factory,
1866
list_only=True, pattern="Test.b")
1867
self.assertEndsWith(output.getvalue(), "Test.b\n")
1868
self.assertLength(1, output.readlines())
1870
def test_list_only_excludes(self):
1871
output = self.run_selftest(test_suite_factory=self.factory,
1872
list_only=True, exclude_pattern="Test.b")
1873
self.assertNotContainsRe("Test.b", output.getvalue())
1874
self.assertLength(2, output.readlines())
1876
def test_lsprof_tests(self):
1877
self.requireFeature(test_lsprof.LSProfFeature)
1880
def __call__(test, result):
1882
def run(test, result):
1883
self.assertIsInstance(result, tests.ForwardingResult)
1884
calls.append("called")
1885
def countTestCases(self):
1887
self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
1888
self.assertLength(1, calls)
1890
def test_random(self):
1891
# test randomising by listing a number of tests.
1892
output_123 = self.run_selftest(test_suite_factory=self.factory,
1893
list_only=True, random_seed="123")
1894
output_234 = self.run_selftest(test_suite_factory=self.factory,
1895
list_only=True, random_seed="234")
1896
self.assertNotEqual(output_123, output_234)
1897
# "Randominzing test order..\n\n
1898
self.assertLength(5, output_123.readlines())
1899
self.assertLength(5, output_234.readlines())
1901
def test_random_reuse_is_same_order(self):
1902
# test randomising by listing a number of tests.
1903
expected = self.run_selftest(test_suite_factory=self.factory,
1904
list_only=True, random_seed="123")
1905
repeated = self.run_selftest(test_suite_factory=self.factory,
1906
list_only=True, random_seed="123")
1907
self.assertEqual(expected.getvalue(), repeated.getvalue())
1909
def test_runner_class(self):
1910
self.requireFeature(features.subunit)
1911
from subunit import ProtocolTestCase
1912
stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
1913
test_suite_factory=self.factory)
1914
test = ProtocolTestCase(stream)
1915
result = unittest.TestResult()
1917
self.assertEqual(3, result.testsRun)
1919
def test_starting_with_single_argument(self):
1920
output = self.run_selftest(test_suite_factory=self.factory,
1921
starting_with=['bzrlib.tests.test_selftest.Test.a'],
1923
self.assertEqual('bzrlib.tests.test_selftest.Test.a\n',
1926
def test_starting_with_multiple_argument(self):
1927
output = self.run_selftest(test_suite_factory=self.factory,
1928
starting_with=['bzrlib.tests.test_selftest.Test.a',
1929
'bzrlib.tests.test_selftest.Test.b'],
1931
self.assertEqual('bzrlib.tests.test_selftest.Test.a\n'
1932
'bzrlib.tests.test_selftest.Test.b\n',
1935
def check_transport_set(self, transport_server):
1936
captured_transport = []
1937
def seen_transport(a_transport):
1938
captured_transport.append(a_transport)
1939
class Capture(tests.TestCase):
1941
seen_transport(bzrlib.tests.default_transport)
1943
return TestUtil.TestSuite([Capture("a")])
1944
self.run_selftest(transport=transport_server, test_suite_factory=factory)
1945
self.assertEqual(transport_server, captured_transport[0])
1947
def test_transport_sftp(self):
1948
self.requireFeature(features.paramiko)
1949
from bzrlib.tests import stub_sftp
1950
self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
1952
def test_transport_memory(self):
1953
self.check_transport_set(memory.MemoryServer)
1956
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
1957
# Does IO: reads test.list
1959
def test_load_list(self):
1960
# Provide a list with one test - this test.
1961
test_id_line = '%s\n' % self.id()
1962
self.build_tree_contents([('test.list', test_id_line)])
1963
# And generate a list of the tests in the suite.
1964
stream = self.run_selftest(load_list='test.list', list_only=True)
1965
self.assertEqual(test_id_line, stream.getvalue())
1967
def test_load_unknown(self):
1968
# Provide a list with one test - this test.
1969
# And generate a list of the tests in the suite.
1970
err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
1971
load_list='missing file name', list_only=True)
1974
class TestRunBzr(tests.TestCase):
1979
def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
1981
"""Override _run_bzr_core to test how it is invoked by run_bzr.
1983
Attempts to run bzr from inside this class don't actually run it.
1985
We test how run_bzr actually invokes bzr in another location. Here we
1986
only need to test that it passes the right parameters to run_bzr.
1988
self.argv = list(argv)
1989
self.retcode = retcode
1990
self.encoding = encoding
1992
self.working_dir = working_dir
1993
return self.retcode, self.out, self.err
1995
def test_run_bzr_error(self):
1996
self.out = "It sure does!\n"
1997
out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
1998
self.assertEqual(['rocks'], self.argv)
1999
self.assertEqual(34, self.retcode)
2000
self.assertEqual('It sure does!\n', out)
2001
self.assertEquals(out, self.out)
2002
self.assertEqual('', err)
2003
self.assertEquals(err, self.err)
2005
def test_run_bzr_error_regexes(self):
2007
self.err = "bzr: ERROR: foobarbaz is not versioned"
2008
out, err = self.run_bzr_error(
2009
["bzr: ERROR: foobarbaz is not versioned"],
2010
['file-id', 'foobarbaz'])
2012
def test_encoding(self):
2013
"""Test that run_bzr passes encoding to _run_bzr_core"""
2014
self.run_bzr('foo bar')
2015
self.assertEqual(None, self.encoding)
2016
self.assertEqual(['foo', 'bar'], self.argv)
2018
self.run_bzr('foo bar', encoding='baz')
2019
self.assertEqual('baz', self.encoding)
2020
self.assertEqual(['foo', 'bar'], self.argv)
2022
def test_retcode(self):
2023
"""Test that run_bzr passes retcode to _run_bzr_core"""
2024
# Default is retcode == 0
2025
self.run_bzr('foo bar')
2026
self.assertEqual(0, self.retcode)
2027
self.assertEqual(['foo', 'bar'], self.argv)
2029
self.run_bzr('foo bar', retcode=1)
2030
self.assertEqual(1, self.retcode)
2031
self.assertEqual(['foo', 'bar'], self.argv)
2033
self.run_bzr('foo bar', retcode=None)
2034
self.assertEqual(None, self.retcode)
2035
self.assertEqual(['foo', 'bar'], self.argv)
2037
self.run_bzr(['foo', 'bar'], retcode=3)
2038
self.assertEqual(3, self.retcode)
2039
self.assertEqual(['foo', 'bar'], self.argv)
2041
def test_stdin(self):
2042
# test that the stdin keyword to run_bzr is passed through to
2043
# _run_bzr_core as-is. We do this by overriding
2044
# _run_bzr_core in this class, and then calling run_bzr,
2045
# which is a convenience function for _run_bzr_core, so
2047
self.run_bzr('foo bar', stdin='gam')
2048
self.assertEqual('gam', self.stdin)
2049
self.assertEqual(['foo', 'bar'], self.argv)
2051
self.run_bzr('foo bar', stdin='zippy')
2052
self.assertEqual('zippy', self.stdin)
2053
self.assertEqual(['foo', 'bar'], self.argv)
2055
def test_working_dir(self):
2056
"""Test that run_bzr passes working_dir to _run_bzr_core"""
2057
self.run_bzr('foo bar')
2058
self.assertEqual(None, self.working_dir)
2059
self.assertEqual(['foo', 'bar'], self.argv)
2061
self.run_bzr('foo bar', working_dir='baz')
2062
self.assertEqual('baz', self.working_dir)
2063
self.assertEqual(['foo', 'bar'], self.argv)
2065
def test_reject_extra_keyword_arguments(self):
2066
self.assertRaises(TypeError, self.run_bzr, "foo bar",
2067
error_regex=['error message'])
2070
class TestRunBzrCaptured(tests.TestCaseWithTransport):
2071
# Does IO when testing the working_dir parameter.
2073
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
2074
a_callable=None, *args, **kwargs):
2076
self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
2077
self.factory = bzrlib.ui.ui_factory
2078
self.working_dir = osutils.getcwd()
2079
stdout.write('foo\n')
2080
stderr.write('bar\n')
2083
def test_stdin(self):
2084
# test that the stdin keyword to _run_bzr_core is passed through to
2085
# apply_redirected as a StringIO. We do this by overriding
2086
# apply_redirected in this class, and then calling _run_bzr_core,
2087
# which calls apply_redirected.
2088
self.run_bzr(['foo', 'bar'], stdin='gam')
2089
self.assertEqual('gam', self.stdin.read())
2090
self.assertTrue(self.stdin is self.factory_stdin)
2091
self.run_bzr(['foo', 'bar'], stdin='zippy')
2092
self.assertEqual('zippy', self.stdin.read())
2093
self.assertTrue(self.stdin is self.factory_stdin)
2095
def test_ui_factory(self):
2096
# each invocation of self.run_bzr should get its
2097
# own UI factory, which is an instance of TestUIFactory,
2098
# with stdin, stdout and stderr attached to the stdin,
2099
# stdout and stderr of the invoked run_bzr
2100
current_factory = bzrlib.ui.ui_factory
2101
self.run_bzr(['foo'])
2102
self.failIf(current_factory is self.factory)
2103
self.assertNotEqual(sys.stdout, self.factory.stdout)
2104
self.assertNotEqual(sys.stderr, self.factory.stderr)
2105
self.assertEqual('foo\n', self.factory.stdout.getvalue())
2106
self.assertEqual('bar\n', self.factory.stderr.getvalue())
2107
self.assertIsInstance(self.factory, tests.TestUIFactory)
2109
def test_working_dir(self):
2110
self.build_tree(['one/', 'two/'])
2111
cwd = osutils.getcwd()
2113
# Default is to work in the current directory
2114
self.run_bzr(['foo', 'bar'])
2115
self.assertEqual(cwd, self.working_dir)
2117
self.run_bzr(['foo', 'bar'], working_dir=None)
2118
self.assertEqual(cwd, self.working_dir)
2120
# The function should be run in the alternative directory
2121
# but afterwards the current working dir shouldn't be changed
2122
self.run_bzr(['foo', 'bar'], working_dir='one')
2123
self.assertNotEqual(cwd, self.working_dir)
2124
self.assertEndsWith(self.working_dir, 'one')
2125
self.assertEqual(cwd, osutils.getcwd())
2127
self.run_bzr(['foo', 'bar'], working_dir='two')
2128
self.assertNotEqual(cwd, self.working_dir)
2129
self.assertEndsWith(self.working_dir, 'two')
2130
self.assertEqual(cwd, osutils.getcwd())
2133
class StubProcess(object):
2134
"""A stub process for testing run_bzr_subprocess."""
2136
def __init__(self, out="", err="", retcode=0):
2139
self.returncode = retcode
2141
def communicate(self):
2142
return self.out, self.err
2145
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
2146
"""Base class for tests testing how we might run bzr."""
2149
tests.TestCaseWithTransport.setUp(self)
2150
self.subprocess_calls = []
2152
def start_bzr_subprocess(self, process_args, env_changes=None,
2153
skip_if_plan_to_signal=False,
2155
allow_plugins=False):
2156
"""capture what run_bzr_subprocess tries to do."""
2157
self.subprocess_calls.append({'process_args':process_args,
2158
'env_changes':env_changes,
2159
'skip_if_plan_to_signal':skip_if_plan_to_signal,
2160
'working_dir':working_dir, 'allow_plugins':allow_plugins})
2161
return self.next_subprocess
2164
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
2166
def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
2167
"""Run run_bzr_subprocess with args and kwargs using a stubbed process.
2169
Inside TestRunBzrSubprocessCommands we use a stub start_bzr_subprocess
2170
that will return static results. This assertion method populates those
2171
results and also checks the arguments run_bzr_subprocess generates.
2173
self.next_subprocess = process
2175
result = self.run_bzr_subprocess(*args, **kwargs)
2177
self.next_subprocess = None
2178
for key, expected in expected_args.iteritems():
2179
self.assertEqual(expected, self.subprocess_calls[-1][key])
1747
class TestKnownFailure(TestCase):
1749
def test_known_failure(self):
1750
"""Check that KnownFailure is defined appropriately."""
1751
# a KnownFailure is an assertion error for compatability with unaware
1753
self.assertIsInstance(KnownFailure(""), AssertionError)
1755
def test_expect_failure(self):
1757
self.expectFailure("Doomed to failure", self.assertTrue, False)
1758
except KnownFailure, e:
1759
self.assertEqual('Doomed to failure', e.args[0])
1761
self.expectFailure("Doomed to failure", self.assertTrue, True)
1762
except AssertionError, e:
1763
self.assertEqual('Unexpected success. Should have failed:'
1764
' Doomed to failure', e.args[0])
2182
self.next_subprocess = None
2183
for key, expected in expected_args.iteritems():
2184
self.assertEqual(expected, self.subprocess_calls[-1][key])
2187
def test_run_bzr_subprocess(self):
2188
"""The run_bzr_helper_external command behaves nicely."""
2189
self.assertRunBzrSubprocess({'process_args':['--version']},
2190
StubProcess(), '--version')
2191
self.assertRunBzrSubprocess({'process_args':['--version']},
2192
StubProcess(), ['--version'])
2193
# retcode=None disables retcode checking
2194
result = self.assertRunBzrSubprocess({},
2195
StubProcess(retcode=3), '--version', retcode=None)
2196
result = self.assertRunBzrSubprocess({},
2197
StubProcess(out="is free software"), '--version')
2198
self.assertContainsRe(result[0], 'is free software')
2199
# Running a subcommand that is missing errors
2200
self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
2201
{'process_args':['--versionn']}, StubProcess(retcode=3),
2203
# Unless it is told to expect the error from the subprocess
2204
result = self.assertRunBzrSubprocess({},
2205
StubProcess(retcode=3), '--versionn', retcode=3)
2206
# Or to ignore retcode checking
2207
result = self.assertRunBzrSubprocess({},
2208
StubProcess(err="unknown command", retcode=3), '--versionn',
2210
self.assertContainsRe(result[1], 'unknown command')
2212
def test_env_change_passes_through(self):
2213
self.assertRunBzrSubprocess(
2214
{'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
2216
env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
2218
def test_no_working_dir_passed_as_None(self):
2219
self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
2221
def test_no_working_dir_passed_through(self):
2222
self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
2225
def test_run_bzr_subprocess_no_plugins(self):
2226
self.assertRunBzrSubprocess({'allow_plugins': False},
2229
def test_allow_plugins(self):
2230
self.assertRunBzrSubprocess({'allow_plugins': True},
2231
StubProcess(), '', allow_plugins=True)
2234
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2236
def test_finish_bzr_subprocess_with_error(self):
2237
"""finish_bzr_subprocess allows specification of the desired exit code.
2239
process = StubProcess(err="unknown command", retcode=3)
2240
result = self.finish_bzr_subprocess(process, retcode=3)
2241
self.assertEqual('', result[0])
2242
self.assertContainsRe(result[1], 'unknown command')
2244
def test_finish_bzr_subprocess_ignoring_retcode(self):
2245
"""finish_bzr_subprocess allows the exit code to be ignored."""
2246
process = StubProcess(err="unknown command", retcode=3)
2247
result = self.finish_bzr_subprocess(process, retcode=None)
2248
self.assertEqual('', result[0])
2249
self.assertContainsRe(result[1], 'unknown command')
2251
def test_finish_subprocess_with_unexpected_retcode(self):
2252
"""finish_bzr_subprocess raises self.failureException if the retcode is
2253
not the expected one.
2255
process = StubProcess(err="unknown command", retcode=3)
2256
self.assertRaises(self.failureException, self.finish_bzr_subprocess,
2260
class _DontSpawnProcess(Exception):
2261
"""A simple exception which just allows us to skip unnecessary steps"""
2264
class TestStartBzrSubProcess(tests.TestCase):
2266
def check_popen_state(self):
2267
"""Replace to make assertions when popen is called."""
2269
def _popen(self, *args, **kwargs):
2270
"""Record the command that is run, so that we can ensure it is correct"""
2271
self.check_popen_state()
2272
self._popen_args = args
2273
self._popen_kwargs = kwargs
2274
raise _DontSpawnProcess()
2276
def test_run_bzr_subprocess_no_plugins(self):
2277
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2278
command = self._popen_args[0]
2279
self.assertEqual(sys.executable, command[0])
2280
self.assertEqual(self.get_bzr_path(), command[1])
2281
self.assertEqual(['--no-plugins'], command[2:])
2283
def test_allow_plugins(self):
2284
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2286
command = self._popen_args[0]
2287
self.assertEqual([], command[2:])
2289
def test_set_env(self):
2290
self.failIf('EXISTANT_ENV_VAR' in os.environ)
2292
def check_environment():
2293
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2294
self.check_popen_state = check_environment
2295
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2296
env_changes={'EXISTANT_ENV_VAR':'set variable'})
2297
# not set in theparent
2298
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2300
def test_run_bzr_subprocess_env_del(self):
2301
"""run_bzr_subprocess can remove environment variables too."""
2302
self.failIf('EXISTANT_ENV_VAR' in os.environ)
2303
def check_environment():
2304
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2305
os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2306
self.check_popen_state = check_environment
2307
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2308
env_changes={'EXISTANT_ENV_VAR':None})
2309
# Still set in parent
2310
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2311
del os.environ['EXISTANT_ENV_VAR']
2313
def test_env_del_missing(self):
2314
self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2315
def check_environment():
2316
self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2317
self.check_popen_state = check_environment
2318
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2319
env_changes={'NON_EXISTANT_ENV_VAR':None})
2321
def test_working_dir(self):
2322
"""Test that we can specify the working dir for the child"""
2323
orig_getcwd = osutils.getcwd
2324
orig_chdir = os.chdir
2332
osutils.getcwd = getcwd
2334
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2337
osutils.getcwd = orig_getcwd
2339
os.chdir = orig_chdir
2340
self.assertEqual(['foo', 'current'], chdirs)
2343
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2344
"""Tests that really need to do things with an external bzr."""
2346
def test_start_and_stop_bzr_subprocess_send_signal(self):
2347
"""finish_bzr_subprocess raises self.failureException if the retcode is
2348
not the expected one.
2350
self.disable_missing_extensions_warning()
2351
process = self.start_bzr_subprocess(['wait-until-signalled'],
2352
skip_if_plan_to_signal=True)
2353
self.assertEqual('running\n', process.stdout.readline())
2354
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):
1766
self.fail('Assertion not raised')
1769
class TestFeature(TestCase):
2362
1771
def test_caching(self):
2363
1772
"""Feature._probe is called by the feature at most once."""
2364
class InstrumentedFeature(tests.Feature):
1773
class InstrumentedFeature(Feature):
2365
1774
def __init__(self):
2366
super(InstrumentedFeature, self).__init__()
1775
Feature.__init__(self)
2367
1776
self.calls = []
2368
1777
def _probe(self):
2369
1778
self.calls.append('_probe')