1200
def _patch_get_bzr_source_tree(self):
1201
# Reading from the actual source tree breaks isolation, but we don't
1202
# want to assume that thats *all* that would happen.
1203
self._get_source_tree_calls = []
1204
def _get_bzr_source_tree():
1205
self._get_source_tree_calls.append("called")
1207
orig_get_bzr_source_tree = bzrlib.version._get_bzr_source_tree
1208
bzrlib.version._get_bzr_source_tree = _get_bzr_source_tree
1210
bzrlib.version._get_bzr_source_tree = orig_get_bzr_source_tree
1211
self.addCleanup(restore)
1233
1213
def test_bench_history(self):
1234
# tests that the running the benchmark produces a history file
1235
# containing a timestamp and the revision id of the bzrlib source which
1237
workingtree = _get_bzr_source_tree()
1214
# tests that the running the benchmark passes bench_history into
1215
# the test result object. We can tell that happens if
1216
# _get_bzr_source_tree is called.
1217
self._patch_get_bzr_source_tree()
1238
1218
test = TestRunner('dummy_test')
1239
1219
output = StringIO()
1240
runner = TextTestRunner(stream=self._log_file, bench_history=output)
1220
runner = tests.TextTestRunner(stream=self._log_file,
1221
bench_history=output)
1241
1222
result = self.run_test_runner(runner, test)
1242
1223
output_string = output.getvalue()
1243
1224
self.assertContainsRe(output_string, "--date [0-9.]+")
1244
if workingtree is not None:
1245
revision_id = workingtree.get_parent_ids()[0]
1246
self.assertEndsWith(output_string.rstrip(), revision_id)
1248
def assertLogDeleted(self, test):
1249
log = test._get_log()
1250
self.assertEqual("DELETED log file to reduce memory footprint", log)
1251
self.assertEqual('', test._log_contents)
1252
self.assertIs(None, test._log_file_name)
1254
def test_success_log_deleted(self):
1255
"""Successful tests have their log deleted"""
1257
class LogTester(TestCase):
1259
def test_success(self):
1260
self.log('this will be removed\n')
1262
sio = cStringIO.StringIO()
1263
runner = TextTestRunner(stream=sio)
1264
test = LogTester('test_success')
1265
result = self.run_test_runner(runner, test)
1267
self.assertLogDeleted(test)
1269
def test_skipped_log_deleted(self):
1270
"""Skipped tests have their log deleted"""
1272
class LogTester(TestCase):
1274
def test_skipped(self):
1275
self.log('this will be removed\n')
1276
raise tests.TestSkipped()
1278
sio = cStringIO.StringIO()
1279
runner = TextTestRunner(stream=sio)
1280
test = LogTester('test_skipped')
1281
result = self.run_test_runner(runner, test)
1283
self.assertLogDeleted(test)
1285
def test_not_aplicable_log_deleted(self):
1286
"""Not applicable tests have their log deleted"""
1288
class LogTester(TestCase):
1290
def test_not_applicable(self):
1291
self.log('this will be removed\n')
1292
raise tests.TestNotApplicable()
1294
sio = cStringIO.StringIO()
1295
runner = TextTestRunner(stream=sio)
1296
test = LogTester('test_not_applicable')
1297
result = self.run_test_runner(runner, test)
1299
self.assertLogDeleted(test)
1301
def test_known_failure_log_deleted(self):
1302
"""Know failure tests have their log deleted"""
1304
class LogTester(TestCase):
1306
def test_known_failure(self):
1307
self.log('this will be removed\n')
1308
raise tests.KnownFailure()
1310
sio = cStringIO.StringIO()
1311
runner = TextTestRunner(stream=sio)
1312
test = LogTester('test_known_failure')
1313
result = self.run_test_runner(runner, test)
1315
self.assertLogDeleted(test)
1317
def test_fail_log_kept(self):
1318
"""Failed tests have their log kept"""
1320
class LogTester(TestCase):
1322
def test_fail(self):
1323
self.log('this will be kept\n')
1324
self.fail('this test fails')
1326
sio = cStringIO.StringIO()
1327
runner = TextTestRunner(stream=sio)
1328
test = LogTester('test_fail')
1329
result = self.run_test_runner(runner, test)
1331
text = sio.getvalue()
1332
self.assertContainsRe(text, 'this will be kept')
1333
self.assertContainsRe(text, 'this test fails')
1335
log = test._get_log()
1336
self.assertContainsRe(log, 'this will be kept')
1337
self.assertEqual(log, test._log_contents)
1339
def test_error_log_kept(self):
1340
"""Tests with errors have their log kept"""
1342
class LogTester(TestCase):
1344
def test_error(self):
1345
self.log('this will be kept\n')
1346
raise ValueError('random exception raised')
1348
sio = cStringIO.StringIO()
1349
runner = TextTestRunner(stream=sio)
1350
test = LogTester('test_error')
1351
result = self.run_test_runner(runner, test)
1353
text = sio.getvalue()
1354
self.assertContainsRe(text, 'this will be kept')
1355
self.assertContainsRe(text, 'random exception raised')
1357
log = test._get_log()
1358
self.assertContainsRe(log, 'this will be kept')
1359
self.assertEqual(log, test._log_contents)
1362
class SampleTestCase(TestCase):
1225
self.assertLength(1, self._get_source_tree_calls)
1227
def test_startTestRun(self):
1228
"""run should call result.startTestRun()"""
1230
class LoggingDecorator(tests.ForwardingResult):
1231
def startTestRun(self):
1232
tests.ForwardingResult.startTestRun(self)
1233
calls.append('startTestRun')
1234
test = unittest.FunctionTestCase(lambda:None)
1236
runner = tests.TextTestRunner(stream=stream,
1237
result_decorators=[LoggingDecorator])
1238
result = self.run_test_runner(runner, test)
1239
self.assertLength(1, calls)
1241
def test_stopTestRun(self):
1242
"""run should call result.stopTestRun()"""
1244
class LoggingDecorator(tests.ForwardingResult):
1245
def stopTestRun(self):
1246
tests.ForwardingResult.stopTestRun(self)
1247
calls.append('stopTestRun')
1248
test = unittest.FunctionTestCase(lambda:None)
1250
runner = tests.TextTestRunner(stream=stream,
1251
result_decorators=[LoggingDecorator])
1252
result = self.run_test_runner(runner, test)
1253
self.assertLength(1, calls)
1256
class SampleTestCase(tests.TestCase):
1364
1258
def _test_pass(self):
1793
1779
tree = self.make_branch_and_memory_tree('a')
1794
1780
self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1797
class TestSFTPMakeBranchAndTree(TestCaseWithSFTPServer):
1799
def test_make_tree_for_sftp_branch(self):
1800
"""Transports backed by local directories create local trees."""
1782
def test_make_tree_for_local_vfs_backed_transport(self):
1783
# make_branch_and_tree has to use local branch and repositories
1784
# when the vfs transport and local disk are colocated, even if
1785
# a different transport is in use for url generation.
1786
from bzrlib.transport.fakevfat import FakeVFATServer
1787
self.transport_server = FakeVFATServer
1788
self.assertFalse(self.get_url('t1').startswith('file://'))
1802
1789
tree = self.make_branch_and_tree('t1')
1803
1790
base = tree.bzrdir.root_transport.base
1804
self.failIf(base.startswith('sftp'),
1805
'base %r is on sftp but should be local' % base)
1791
self.assertStartsWith(base, 'file://')
1806
1792
self.assertEquals(tree.bzrdir.root_transport,
1807
1793
tree.branch.bzrdir.root_transport)
1808
1794
self.assertEquals(tree.bzrdir.root_transport,
1809
1795
tree.branch.repository.bzrdir.root_transport)
1812
class TestSelftest(TestCase):
1798
class SelfTestHelper:
1800
def run_selftest(self, **kwargs):
1801
"""Run selftest returning its output."""
1803
old_transport = bzrlib.tests.default_transport
1804
old_root = tests.TestCaseWithMemoryTransport.TEST_ROOT
1805
tests.TestCaseWithMemoryTransport.TEST_ROOT = None
1807
self.assertEqual(True, tests.selftest(stream=output, **kwargs))
1809
bzrlib.tests.default_transport = old_transport
1810
tests.TestCaseWithMemoryTransport.TEST_ROOT = old_root
1815
class TestSelftest(tests.TestCase, SelfTestHelper):
1813
1816
"""Tests of bzrlib.tests.selftest."""
1815
1818
def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
1816
1819
factory_called = []
1818
1821
factory_called.append(True)
1822
return TestUtil.TestSuite()
1820
1823
out = StringIO()
1821
1824
err = StringIO()
1822
1825
self.apply_redirected(out, err, None, bzrlib.tests.selftest,
1823
1826
test_suite_factory=factory)
1824
1827
self.assertEqual([True], factory_called)
1827
class TestKnownFailure(TestCase):
1829
def test_known_failure(self):
1830
"""Check that KnownFailure is defined appropriately."""
1831
# a KnownFailure is an assertion error for compatability with unaware
1833
self.assertIsInstance(KnownFailure(""), AssertionError)
1835
def test_expect_failure(self):
1837
self.expectFailure("Doomed to failure", self.assertTrue, False)
1838
except KnownFailure, e:
1839
self.assertEqual('Doomed to failure', e.args[0])
1841
self.expectFailure("Doomed to failure", self.assertTrue, True)
1842
except AssertionError, e:
1843
self.assertEqual('Unexpected success. Should have failed:'
1844
' Doomed to failure', e.args[0])
1830
"""A test suite factory."""
1831
class Test(tests.TestCase):
1838
return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
1840
def test_list_only(self):
1841
output = self.run_selftest(test_suite_factory=self.factory,
1843
self.assertEqual(3, len(output.readlines()))
1845
def test_list_only_filtered(self):
1846
output = self.run_selftest(test_suite_factory=self.factory,
1847
list_only=True, pattern="Test.b")
1848
self.assertEndsWith(output.getvalue(), "Test.b\n")
1849
self.assertLength(1, output.readlines())
1851
def test_list_only_excludes(self):
1852
output = self.run_selftest(test_suite_factory=self.factory,
1853
list_only=True, exclude_pattern="Test.b")
1854
self.assertNotContainsRe("Test.b", output.getvalue())
1855
self.assertLength(2, output.readlines())
1857
def test_lsprof_tests(self):
1858
self.requireFeature(test_lsprof.LSProfFeature)
1861
def __call__(test, result):
1863
def run(test, result):
1864
self.assertIsInstance(result, tests.ForwardingResult)
1865
calls.append("called")
1866
def countTestCases(self):
1868
self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
1869
self.assertLength(1, calls)
1871
def test_random(self):
1872
# test randomising by listing a number of tests.
1873
output_123 = self.run_selftest(test_suite_factory=self.factory,
1874
list_only=True, random_seed="123")
1875
output_234 = self.run_selftest(test_suite_factory=self.factory,
1876
list_only=True, random_seed="234")
1877
self.assertNotEqual(output_123, output_234)
1878
# "Randominzing test order..\n\n
1879
self.assertLength(5, output_123.readlines())
1880
self.assertLength(5, output_234.readlines())
1882
def test_random_reuse_is_same_order(self):
1883
# test randomising by listing a number of tests.
1884
expected = self.run_selftest(test_suite_factory=self.factory,
1885
list_only=True, random_seed="123")
1886
repeated = self.run_selftest(test_suite_factory=self.factory,
1887
list_only=True, random_seed="123")
1888
self.assertEqual(expected.getvalue(), repeated.getvalue())
1890
def test_runner_class(self):
1891
self.requireFeature(features.subunit)
1892
from subunit import ProtocolTestCase
1893
stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
1894
test_suite_factory=self.factory)
1895
test = ProtocolTestCase(stream)
1896
result = unittest.TestResult()
1898
self.assertEqual(3, result.testsRun)
1900
def test_starting_with_single_argument(self):
1901
output = self.run_selftest(test_suite_factory=self.factory,
1902
starting_with=['bzrlib.tests.test_selftest.Test.a'],
1904
self.assertEqual('bzrlib.tests.test_selftest.Test.a\n',
1907
def test_starting_with_multiple_argument(self):
1908
output = self.run_selftest(test_suite_factory=self.factory,
1909
starting_with=['bzrlib.tests.test_selftest.Test.a',
1910
'bzrlib.tests.test_selftest.Test.b'],
1912
self.assertEqual('bzrlib.tests.test_selftest.Test.a\n'
1913
'bzrlib.tests.test_selftest.Test.b\n',
1916
def check_transport_set(self, transport_server):
1917
captured_transport = []
1918
def seen_transport(a_transport):
1919
captured_transport.append(a_transport)
1920
class Capture(tests.TestCase):
1922
seen_transport(bzrlib.tests.default_transport)
1924
return TestUtil.TestSuite([Capture("a")])
1925
self.run_selftest(transport=transport_server, test_suite_factory=factory)
1926
self.assertEqual(transport_server, captured_transport[0])
1928
def test_transport_sftp(self):
1929
self.requireFeature(features.paramiko)
1930
self.check_transport_set(bzrlib.transport.sftp.SFTPAbsoluteServer)
1932
def test_transport_memory(self):
1933
self.check_transport_set(bzrlib.transport.memory.MemoryServer)
1936
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
1937
# Does IO: reads test.list
1939
def test_load_list(self):
1940
# Provide a list with one test - this test.
1941
test_id_line = '%s\n' % self.id()
1942
self.build_tree_contents([('test.list', test_id_line)])
1943
# And generate a list of the tests in the suite.
1944
stream = self.run_selftest(load_list='test.list', list_only=True)
1945
self.assertEqual(test_id_line, stream.getvalue())
1947
def test_load_unknown(self):
1948
# Provide a list with one test - this test.
1949
# And generate a list of the tests in the suite.
1950
err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
1951
load_list='missing file name', list_only=True)
1954
class TestRunBzr(tests.TestCase):
1959
def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
1961
"""Override _run_bzr_core to test how it is invoked by run_bzr.
1963
Attempts to run bzr from inside this class don't actually run it.
1965
We test how run_bzr actually invokes bzr in another location. Here we
1966
only need to test that it passes the right parameters to run_bzr.
1968
self.argv = list(argv)
1969
self.retcode = retcode
1970
self.encoding = encoding
1972
self.working_dir = working_dir
1973
return self.retcode, self.out, self.err
1975
def test_run_bzr_error(self):
1976
self.out = "It sure does!\n"
1977
out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
1978
self.assertEqual(['rocks'], self.argv)
1979
self.assertEqual(34, self.retcode)
1980
self.assertEqual('It sure does!\n', out)
1981
self.assertEquals(out, self.out)
1982
self.assertEqual('', err)
1983
self.assertEquals(err, self.err)
1985
def test_run_bzr_error_regexes(self):
1987
self.err = "bzr: ERROR: foobarbaz is not versioned"
1988
out, err = self.run_bzr_error(
1989
["bzr: ERROR: foobarbaz is not versioned"],
1990
['file-id', 'foobarbaz'])
1992
def test_encoding(self):
1993
"""Test that run_bzr passes encoding to _run_bzr_core"""
1994
self.run_bzr('foo bar')
1995
self.assertEqual(None, self.encoding)
1996
self.assertEqual(['foo', 'bar'], self.argv)
1998
self.run_bzr('foo bar', encoding='baz')
1999
self.assertEqual('baz', self.encoding)
2000
self.assertEqual(['foo', 'bar'], self.argv)
2002
def test_retcode(self):
2003
"""Test that run_bzr passes retcode to _run_bzr_core"""
2004
# Default is retcode == 0
2005
self.run_bzr('foo bar')
2006
self.assertEqual(0, self.retcode)
2007
self.assertEqual(['foo', 'bar'], self.argv)
2009
self.run_bzr('foo bar', retcode=1)
2010
self.assertEqual(1, self.retcode)
2011
self.assertEqual(['foo', 'bar'], self.argv)
2013
self.run_bzr('foo bar', retcode=None)
2014
self.assertEqual(None, self.retcode)
2015
self.assertEqual(['foo', 'bar'], self.argv)
2017
self.run_bzr(['foo', 'bar'], retcode=3)
2018
self.assertEqual(3, self.retcode)
2019
self.assertEqual(['foo', 'bar'], self.argv)
2021
def test_stdin(self):
2022
# test that the stdin keyword to run_bzr is passed through to
2023
# _run_bzr_core as-is. We do this by overriding
2024
# _run_bzr_core in this class, and then calling run_bzr,
2025
# which is a convenience function for _run_bzr_core, so
2027
self.run_bzr('foo bar', stdin='gam')
2028
self.assertEqual('gam', self.stdin)
2029
self.assertEqual(['foo', 'bar'], self.argv)
2031
self.run_bzr('foo bar', stdin='zippy')
2032
self.assertEqual('zippy', self.stdin)
2033
self.assertEqual(['foo', 'bar'], self.argv)
2035
def test_working_dir(self):
2036
"""Test that run_bzr passes working_dir to _run_bzr_core"""
2037
self.run_bzr('foo bar')
2038
self.assertEqual(None, self.working_dir)
2039
self.assertEqual(['foo', 'bar'], self.argv)
2041
self.run_bzr('foo bar', working_dir='baz')
2042
self.assertEqual('baz', self.working_dir)
2043
self.assertEqual(['foo', 'bar'], self.argv)
2045
def test_reject_extra_keyword_arguments(self):
2046
self.assertRaises(TypeError, self.run_bzr, "foo bar",
2047
error_regex=['error message'])
2050
class TestRunBzrCaptured(tests.TestCaseWithTransport):
2051
# Does IO when testing the working_dir parameter.
2053
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
2054
a_callable=None, *args, **kwargs):
2056
self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
2057
self.factory = bzrlib.ui.ui_factory
2058
self.working_dir = osutils.getcwd()
2059
stdout.write('foo\n')
2060
stderr.write('bar\n')
2063
def test_stdin(self):
2064
# test that the stdin keyword to _run_bzr_core is passed through to
2065
# apply_redirected as a StringIO. We do this by overriding
2066
# apply_redirected in this class, and then calling _run_bzr_core,
2067
# which calls apply_redirected.
2068
self.run_bzr(['foo', 'bar'], stdin='gam')
2069
self.assertEqual('gam', self.stdin.read())
2070
self.assertTrue(self.stdin is self.factory_stdin)
2071
self.run_bzr(['foo', 'bar'], stdin='zippy')
2072
self.assertEqual('zippy', self.stdin.read())
2073
self.assertTrue(self.stdin is self.factory_stdin)
2075
def test_ui_factory(self):
2076
# each invocation of self.run_bzr should get its
2077
# own UI factory, which is an instance of TestUIFactory,
2078
# with stdin, stdout and stderr attached to the stdin,
2079
# stdout and stderr of the invoked run_bzr
2080
current_factory = bzrlib.ui.ui_factory
2081
self.run_bzr(['foo'])
2082
self.failIf(current_factory is self.factory)
2083
self.assertNotEqual(sys.stdout, self.factory.stdout)
2084
self.assertNotEqual(sys.stderr, self.factory.stderr)
2085
self.assertEqual('foo\n', self.factory.stdout.getvalue())
2086
self.assertEqual('bar\n', self.factory.stderr.getvalue())
2087
self.assertIsInstance(self.factory, tests.TestUIFactory)
2089
def test_working_dir(self):
2090
self.build_tree(['one/', 'two/'])
2091
cwd = osutils.getcwd()
2093
# Default is to work in the current directory
2094
self.run_bzr(['foo', 'bar'])
2095
self.assertEqual(cwd, self.working_dir)
2097
self.run_bzr(['foo', 'bar'], working_dir=None)
2098
self.assertEqual(cwd, self.working_dir)
2100
# The function should be run in the alternative directory
2101
# but afterwards the current working dir shouldn't be changed
2102
self.run_bzr(['foo', 'bar'], working_dir='one')
2103
self.assertNotEqual(cwd, self.working_dir)
2104
self.assertEndsWith(self.working_dir, 'one')
2105
self.assertEqual(cwd, osutils.getcwd())
2107
self.run_bzr(['foo', 'bar'], working_dir='two')
2108
self.assertNotEqual(cwd, self.working_dir)
2109
self.assertEndsWith(self.working_dir, 'two')
2110
self.assertEqual(cwd, osutils.getcwd())
2113
class StubProcess(object):
2114
"""A stub process for testing run_bzr_subprocess."""
2116
def __init__(self, out="", err="", retcode=0):
2119
self.returncode = retcode
2121
def communicate(self):
2122
return self.out, self.err
2125
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
2126
"""Base class for tests testing how we might run bzr."""
2129
tests.TestCaseWithTransport.setUp(self)
2130
self.subprocess_calls = []
2132
def start_bzr_subprocess(self, process_args, env_changes=None,
2133
skip_if_plan_to_signal=False,
2135
allow_plugins=False):
2136
"""capture what run_bzr_subprocess tries to do."""
2137
self.subprocess_calls.append({'process_args':process_args,
2138
'env_changes':env_changes,
2139
'skip_if_plan_to_signal':skip_if_plan_to_signal,
2140
'working_dir':working_dir, 'allow_plugins':allow_plugins})
2141
return self.next_subprocess
2144
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
2146
def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
2147
"""Run run_bzr_subprocess with args and kwargs using a stubbed process.
2149
Inside TestRunBzrSubprocessCommands we use a stub start_bzr_subprocess
2150
that will return static results. This assertion method populates those
2151
results and also checks the arguments run_bzr_subprocess generates.
2153
self.next_subprocess = process
2155
result = self.run_bzr_subprocess(*args, **kwargs)
2157
self.next_subprocess = None
2158
for key, expected in expected_args.iteritems():
2159
self.assertEqual(expected, self.subprocess_calls[-1][key])
1846
self.fail('Assertion not raised')
1849
class TestFeature(TestCase):
2162
self.next_subprocess = None
2163
for key, expected in expected_args.iteritems():
2164
self.assertEqual(expected, self.subprocess_calls[-1][key])
2167
def test_run_bzr_subprocess(self):
2168
"""The run_bzr_helper_external command behaves nicely."""
2169
self.assertRunBzrSubprocess({'process_args':['--version']},
2170
StubProcess(), '--version')
2171
self.assertRunBzrSubprocess({'process_args':['--version']},
2172
StubProcess(), ['--version'])
2173
# retcode=None disables retcode checking
2174
result = self.assertRunBzrSubprocess({},
2175
StubProcess(retcode=3), '--version', retcode=None)
2176
result = self.assertRunBzrSubprocess({},
2177
StubProcess(out="is free software"), '--version')
2178
self.assertContainsRe(result[0], 'is free software')
2179
# Running a subcommand that is missing errors
2180
self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
2181
{'process_args':['--versionn']}, StubProcess(retcode=3),
2183
# Unless it is told to expect the error from the subprocess
2184
result = self.assertRunBzrSubprocess({},
2185
StubProcess(retcode=3), '--versionn', retcode=3)
2186
# Or to ignore retcode checking
2187
result = self.assertRunBzrSubprocess({},
2188
StubProcess(err="unknown command", retcode=3), '--versionn',
2190
self.assertContainsRe(result[1], 'unknown command')
2192
def test_env_change_passes_through(self):
2193
self.assertRunBzrSubprocess(
2194
{'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
2196
env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
2198
def test_no_working_dir_passed_as_None(self):
2199
self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
2201
def test_no_working_dir_passed_through(self):
2202
self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
2205
def test_run_bzr_subprocess_no_plugins(self):
2206
self.assertRunBzrSubprocess({'allow_plugins': False},
2209
def test_allow_plugins(self):
2210
self.assertRunBzrSubprocess({'allow_plugins': True},
2211
StubProcess(), '', allow_plugins=True)
2214
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2216
def test_finish_bzr_subprocess_with_error(self):
2217
"""finish_bzr_subprocess allows specification of the desired exit code.
2219
process = StubProcess(err="unknown command", retcode=3)
2220
result = self.finish_bzr_subprocess(process, retcode=3)
2221
self.assertEqual('', result[0])
2222
self.assertContainsRe(result[1], 'unknown command')
2224
def test_finish_bzr_subprocess_ignoring_retcode(self):
2225
"""finish_bzr_subprocess allows the exit code to be ignored."""
2226
process = StubProcess(err="unknown command", retcode=3)
2227
result = self.finish_bzr_subprocess(process, retcode=None)
2228
self.assertEqual('', result[0])
2229
self.assertContainsRe(result[1], 'unknown command')
2231
def test_finish_subprocess_with_unexpected_retcode(self):
2232
"""finish_bzr_subprocess raises self.failureException if the retcode is
2233
not the expected one.
2235
process = StubProcess(err="unknown command", retcode=3)
2236
self.assertRaises(self.failureException, self.finish_bzr_subprocess,
2240
class _DontSpawnProcess(Exception):
2241
"""A simple exception which just allows us to skip unnecessary steps"""
2244
class TestStartBzrSubProcess(tests.TestCase):
2246
def check_popen_state(self):
2247
"""Replace to make assertions when popen is called."""
2249
def _popen(self, *args, **kwargs):
2250
"""Record the command that is run, so that we can ensure it is correct"""
2251
self.check_popen_state()
2252
self._popen_args = args
2253
self._popen_kwargs = kwargs
2254
raise _DontSpawnProcess()
2256
def test_run_bzr_subprocess_no_plugins(self):
2257
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2258
command = self._popen_args[0]
2259
self.assertEqual(sys.executable, command[0])
2260
self.assertEqual(self.get_bzr_path(), command[1])
2261
self.assertEqual(['--no-plugins'], command[2:])
2263
def test_allow_plugins(self):
2264
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2266
command = self._popen_args[0]
2267
self.assertEqual([], command[2:])
2269
def test_set_env(self):
2270
self.failIf('EXISTANT_ENV_VAR' in os.environ)
2272
def check_environment():
2273
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2274
self.check_popen_state = check_environment
2275
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2276
env_changes={'EXISTANT_ENV_VAR':'set variable'})
2277
# not set in theparent
2278
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2280
def test_run_bzr_subprocess_env_del(self):
2281
"""run_bzr_subprocess can remove environment variables too."""
2282
self.failIf('EXISTANT_ENV_VAR' in os.environ)
2283
def check_environment():
2284
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2285
os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2286
self.check_popen_state = check_environment
2287
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2288
env_changes={'EXISTANT_ENV_VAR':None})
2289
# Still set in parent
2290
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2291
del os.environ['EXISTANT_ENV_VAR']
2293
def test_env_del_missing(self):
2294
self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2295
def check_environment():
2296
self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2297
self.check_popen_state = check_environment
2298
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2299
env_changes={'NON_EXISTANT_ENV_VAR':None})
2301
def test_working_dir(self):
2302
"""Test that we can specify the working dir for the child"""
2303
orig_getcwd = osutils.getcwd
2304
orig_chdir = os.chdir
2312
osutils.getcwd = getcwd
2314
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2317
osutils.getcwd = orig_getcwd
2319
os.chdir = orig_chdir
2320
self.assertEqual(['foo', 'current'], chdirs)
2323
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2324
"""Tests that really need to do things with an external bzr."""
2326
def test_start_and_stop_bzr_subprocess_send_signal(self):
2327
"""finish_bzr_subprocess raises self.failureException if the retcode is
2328
not the expected one.
2330
self.disable_missing_extensions_warning()
2331
process = self.start_bzr_subprocess(['wait-until-signalled'],
2332
skip_if_plan_to_signal=True)
2333
self.assertEqual('running\n', process.stdout.readline())
2334
result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
2336
self.assertEqual('', result[0])
2337
self.assertEqual('bzr: interrupted\n', result[1])
2340
class TestFeature(tests.TestCase):
1851
2342
def test_caching(self):
1852
2343
"""Feature._probe is called by the feature at most once."""
1853
class InstrumentedFeature(Feature):
2344
class InstrumentedFeature(tests.Feature):
1854
2345
def __init__(self):
1855
Feature.__init__(self)
2346
super(InstrumentedFeature, self).__init__()
1856
2347
self.calls = []
1857
2348
def _probe(self):
1858
2349
self.calls.append('_probe')
added
removed
bzrlib/tests/test_selftest.py
