13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
17
"""Tests for the test framework."""
19
from cStringIO import StringIO
21
from StringIO import StringIO
29
from testtools import (
30
ExtendedToOriginalDecorator,
32
__version__ as testtools_version,
34
from testtools.content import Content
35
from testtools.content_type import ContentType
36
from testtools.matchers import (
40
import testtools.testresult.doubles
28
43
from bzrlib import (
40
from bzrlib.progress import _BaseProgressBar
41
60
from bzrlib.repofmt import (
45
63
from bzrlib.symbol_versioning import (
50
68
from bzrlib.tests import (
57
TestCaseWithMemoryTransport,
58
TestCaseWithTransport,
67
exclude_tests_by_condition,
69
filter_suite_by_condition,
74
split_suite_by_condition,
79
from bzrlib.tests.test_sftp_transport import TestCaseWithSFTPServer
80
from bzrlib.tests.TestUtil import _load_module_by_name
81
from bzrlib.trace import note
82
from bzrlib.transport.memory import MemoryServer, MemoryTransport
83
from bzrlib.version import _get_bzr_source_tree
74
from bzrlib.trace import note, mutter
75
from bzrlib.transport import memory
86
78
def _test_ids(test_suite):
87
79
"""Get the ids for the tests in a test suite."""
88
return [t.id() for t in iter_suite_tests(test_suite)]
91
class SelftestTests(TestCase):
93
def test_import_tests(self):
94
mod = _load_module_by_name('bzrlib.tests.test_selftest')
95
self.assertEqual(mod.SelftestTests, SelftestTests)
97
def test_import_test_failure(self):
98
self.assertRaises(ImportError,
100
'bzrlib.no-name-yet')
102
class MetaTestLog(TestCase):
80
return [t.id() for t in tests.iter_suite_tests(test_suite)]
83
class MetaTestLog(tests.TestCase):
104
85
def test_logging(self):
105
86
"""Test logs are captured when a test fails."""
106
87
self.log('a test message')
107
self._log_file.flush()
108
self.assertContainsRe(self._get_log(keep_log_file=True),
112
class TestUnicodeFilename(TestCase):
114
def test_probe_passes(self):
115
"""UnicodeFilename._probe passes."""
116
# We can't test much more than that because the behaviour depends
118
tests.UnicodeFilename._probe()
121
class TestTreeShape(TestCaseInTempDir):
88
details = self.getDetails()
90
self.assertThat(log.content_type, Equals(ContentType(
91
"text", "plain", {"charset": "utf8"})))
92
self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
93
self.assertThat(self.get_log(),
94
DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
97
class TestTreeShape(tests.TestCaseInTempDir):
123
99
def test_unicode_paths(self):
124
self.requireFeature(tests.UnicodeFilename)
100
self.requireFeature(features.UnicodeFilenameFeature)
126
102
filename = u'hell\u00d8'
127
103
self.build_tree_contents([(filename, 'contents of hello')])
128
self.failUnlessExists(filename)
131
class TestTransportProviderAdapter(TestCase):
104
self.assertPathExists(filename)
107
class TestClassesAvailable(tests.TestCase):
108
"""As a convenience we expose Test* classes from bzrlib.tests"""
110
def test_test_case(self):
111
from bzrlib.tests import TestCase
113
def test_test_loader(self):
114
from bzrlib.tests import TestLoader
116
def test_test_suite(self):
117
from bzrlib.tests import TestSuite
120
class TestTransportScenarios(tests.TestCase):
132
121
"""A group of tests that test the transport implementation adaption core.
134
This is a meta test that the tests are applied to all available
123
This is a meta test that the tests are applied to all available
137
This will be generalised in the future which is why it is in this
126
This will be generalised in the future which is why it is in this
138
127
test file even though it is specific to transport tests at the moment.
141
130
def test_get_transport_permutations(self):
142
131
# this checks that get_test_permutations defined by the module is
143
# called by the adapter get_transport_test_permutations method.
132
# called by the get_transport_test_permutations function.
144
133
class MockModule(object):
145
134
def get_test_permutations(self):
146
135
return sample_permutation
147
136
sample_permutation = [(1,2), (3,4)]
148
from bzrlib.tests.test_transport_implementations \
149
import TransportTestProviderAdapter
150
adapter = TransportTestProviderAdapter()
137
from bzrlib.tests.per_transport import get_transport_test_permutations
151
138
self.assertEqual(sample_permutation,
152
adapter.get_transport_test_permutations(MockModule()))
139
get_transport_test_permutations(MockModule()))
154
def test_adapter_checks_all_modules(self):
155
# this checks that the adapter returns as many permutations as there
156
# are in all the registered transport modules - we assume if this
157
# matches its probably doing the right thing especially in combination
158
# with the tests for setting the right classes below.
159
from bzrlib.tests.test_transport_implementations \
160
import TransportTestProviderAdapter
141
def test_scenarios_include_all_modules(self):
142
# this checks that the scenario generator returns as many permutations
143
# as there are in all the registered transport modules - we assume if
144
# this matches its probably doing the right thing especially in
145
# combination with the tests for setting the right classes below.
146
from bzrlib.tests.per_transport import transport_test_permutations
161
147
from bzrlib.transport import _get_transport_modules
162
148
modules = _get_transport_modules()
163
149
permutation_count = 0
164
150
for module in modules:
166
permutation_count += len(reduce(getattr,
152
permutation_count += len(reduce(getattr,
167
153
(module + ".get_test_permutations").split('.')[1:],
168
154
__import__(module))())
169
155
except errors.DependencyNotPresent:
171
input_test = TestTransportProviderAdapter(
172
"test_adapter_sets_transport_class")
173
adapter = TransportTestProviderAdapter()
174
self.assertEqual(permutation_count,
175
len(list(iter(adapter.adapt(input_test)))))
157
scenarios = transport_test_permutations()
158
self.assertEqual(permutation_count, len(scenarios))
177
def test_adapter_sets_transport_class(self):
178
# Check that the test adapter inserts a transport and server into the
160
def test_scenarios_include_transport_class(self):
181
161
# This test used to know about all the possible transports and the
182
162
# order they were returned but that seems overly brittle (mbp
184
from bzrlib.tests.test_transport_implementations \
185
import TransportTestProviderAdapter
186
scenarios = TransportTestProviderAdapter().scenarios
164
from bzrlib.tests.per_transport import transport_test_permutations
165
scenarios = transport_test_permutations()
187
166
# there are at least that many builtin transports
188
167
self.assertTrue(len(scenarios) > 6)
189
168
one_scenario = scenarios[0]
1194
def test_bench_history(self):
1195
# tests that the running the benchmark produces a history file
1196
# containing a timestamp and the revision id of the bzrlib source which
1198
workingtree = _get_bzr_source_tree()
1199
test = TestRunner('dummy_test')
1201
runner = TextTestRunner(stream=self._log_file, bench_history=output)
1202
result = self.run_test_runner(runner, test)
1203
output_string = output.getvalue()
1204
self.assertContainsRe(output_string, "--date [0-9.]+")
1205
if workingtree is not None:
1206
revision_id = workingtree.get_parent_ids()[0]
1207
self.assertEndsWith(output_string.rstrip(), revision_id)
1209
def assertLogDeleted(self, test):
1210
log = test._get_log()
1211
self.assertEqual("DELETED log file to reduce memory footprint", log)
1212
self.assertEqual('', test._log_contents)
1213
self.assertIs(None, test._log_file_name)
1215
def test_success_log_deleted(self):
1216
"""Successful tests have their log deleted"""
1218
class LogTester(TestCase):
1220
def test_success(self):
1221
self.log('this will be removed\n')
1223
sio = cStringIO.StringIO()
1224
runner = TextTestRunner(stream=sio)
1225
test = LogTester('test_success')
1226
result = self.run_test_runner(runner, test)
1228
self.assertLogDeleted(test)
1230
def test_skipped_log_deleted(self):
1231
"""Skipped tests have their log deleted"""
1233
class LogTester(TestCase):
1235
def test_skipped(self):
1236
self.log('this will be removed\n')
1237
raise tests.TestSkipped()
1239
sio = cStringIO.StringIO()
1240
runner = TextTestRunner(stream=sio)
1241
test = LogTester('test_skipped')
1242
result = self.run_test_runner(runner, test)
1244
self.assertLogDeleted(test)
1246
def test_not_aplicable_log_deleted(self):
1247
"""Not applicable tests have their log deleted"""
1249
class LogTester(TestCase):
1251
def test_not_applicable(self):
1252
self.log('this will be removed\n')
1253
raise tests.TestNotApplicable()
1255
sio = cStringIO.StringIO()
1256
runner = TextTestRunner(stream=sio)
1257
test = LogTester('test_not_applicable')
1258
result = self.run_test_runner(runner, test)
1260
self.assertLogDeleted(test)
1262
def test_known_failure_log_deleted(self):
1263
"""Know failure tests have their log deleted"""
1265
class LogTester(TestCase):
1267
def test_known_failure(self):
1268
self.log('this will be removed\n')
1269
raise tests.KnownFailure()
1271
sio = cStringIO.StringIO()
1272
runner = TextTestRunner(stream=sio)
1273
test = LogTester('test_known_failure')
1274
result = self.run_test_runner(runner, test)
1276
self.assertLogDeleted(test)
1278
def test_fail_log_kept(self):
1279
"""Failed tests have their log kept"""
1281
class LogTester(TestCase):
1283
def test_fail(self):
1284
self.log('this will be kept\n')
1285
self.fail('this test fails')
1287
sio = cStringIO.StringIO()
1288
runner = TextTestRunner(stream=sio)
1289
test = LogTester('test_fail')
1290
result = self.run_test_runner(runner, test)
1292
text = sio.getvalue()
1293
self.assertContainsRe(text, 'this will be kept')
1294
self.assertContainsRe(text, 'this test fails')
1296
log = test._get_log()
1297
self.assertContainsRe(log, 'this will be kept')
1298
self.assertEqual(log, test._log_contents)
1300
def test_error_log_kept(self):
1301
"""Tests with errors have their log kept"""
1303
class LogTester(TestCase):
1305
def test_error(self):
1306
self.log('this will be kept\n')
1307
raise ValueError('random exception raised')
1309
sio = cStringIO.StringIO()
1310
runner = TextTestRunner(stream=sio)
1311
test = LogTester('test_error')
1312
result = self.run_test_runner(runner, test)
1314
text = sio.getvalue()
1315
self.assertContainsRe(text, 'this will be kept')
1316
self.assertContainsRe(text, 'random exception raised')
1318
log = test._get_log()
1319
self.assertContainsRe(log, 'this will be kept')
1320
self.assertEqual(log, test._log_contents)
1323
class SampleTestCase(TestCase):
1203
def test_verbose_test_count(self):
1204
"""A verbose test run reports the right test count at the start"""
1205
suite = TestUtil.TestSuite([
1206
unittest.FunctionTestCase(lambda:None),
1207
unittest.FunctionTestCase(lambda:None)])
1208
self.assertEqual(suite.countTestCases(), 2)
1210
runner = tests.TextTestRunner(stream=stream, verbosity=2)
1211
# Need to use the CountingDecorator as that's what sets num_tests
1212
result = self.run_test_runner(runner, tests.CountingDecorator(suite))
1213
self.assertStartsWith(stream.getvalue(), "running 2 tests")
1215
def test_startTestRun(self):
1216
"""run should call result.startTestRun()"""
1218
class LoggingDecorator(ExtendedToOriginalDecorator):
1219
def startTestRun(self):
1220
ExtendedToOriginalDecorator.startTestRun(self)
1221
calls.append('startTestRun')
1222
test = unittest.FunctionTestCase(lambda:None)
1224
runner = tests.TextTestRunner(stream=stream,
1225
result_decorators=[LoggingDecorator])
1226
result = self.run_test_runner(runner, test)
1227
self.assertLength(1, calls)
1229
def test_stopTestRun(self):
1230
"""run should call result.stopTestRun()"""
1232
class LoggingDecorator(ExtendedToOriginalDecorator):
1233
def stopTestRun(self):
1234
ExtendedToOriginalDecorator.stopTestRun(self)
1235
calls.append('stopTestRun')
1236
test = unittest.FunctionTestCase(lambda:None)
1238
runner = tests.TextTestRunner(stream=stream,
1239
result_decorators=[LoggingDecorator])
1240
result = self.run_test_runner(runner, test)
1241
self.assertLength(1, calls)
1243
def test_unicode_test_output_on_ascii_stream(self):
1244
"""Showing results should always succeed even on an ascii console"""
1245
class FailureWithUnicode(tests.TestCase):
1246
def test_log_unicode(self):
1248
self.fail("Now print that log!")
1250
self.overrideAttr(osutils, "get_terminal_encoding",
1251
lambda trace=False: "ascii")
1252
result = self.run_test_runner(tests.TextTestRunner(stream=out),
1253
FailureWithUnicode("test_log_unicode"))
1254
if testtools_version[:3] > (0, 9, 11):
1255
self.assertContainsRe(out.getvalue(), "log: {{{\d+\.\d+ \\\\u2606}}}")
1257
self.assertContainsRe(out.getvalue(),
1258
"Text attachment: log\n"
1260
"\d+\.\d+ \\\\u2606\n"
1264
class SampleTestCase(tests.TestCase):
1325
1266
def _test_pass(self):
1545
1637
self.assertRaises(AssertionError,
1546
1638
self.assertListRaises, _TestException, success_generator)
1549
@symbol_versioning.deprecated_function(zero_eleven)
1640
def test_overrideAttr_without_value(self):
1641
self.test_attr = 'original' # Define a test attribute
1642
obj = self # Make 'obj' visible to the embedded test
1643
class Test(tests.TestCase):
1646
tests.TestCase.setUp(self)
1647
self.orig = self.overrideAttr(obj, 'test_attr')
1649
def test_value(self):
1650
self.assertEqual('original', self.orig)
1651
self.assertEqual('original', obj.test_attr)
1652
obj.test_attr = 'modified'
1653
self.assertEqual('modified', obj.test_attr)
1655
test = Test('test_value')
1656
test.run(unittest.TestResult())
1657
self.assertEqual('original', obj.test_attr)
1659
def test_overrideAttr_with_value(self):
1660
self.test_attr = 'original' # Define a test attribute
1661
obj = self # Make 'obj' visible to the embedded test
1662
class Test(tests.TestCase):
1665
tests.TestCase.setUp(self)
1666
self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1668
def test_value(self):
1669
self.assertEqual('original', self.orig)
1670
self.assertEqual('modified', obj.test_attr)
1672
test = Test('test_value')
1673
test.run(unittest.TestResult())
1674
self.assertEqual('original', obj.test_attr)
1676
def test_recordCalls(self):
1677
from bzrlib.tests import test_selftest
1678
calls = self.recordCalls(
1679
test_selftest, '_add_numbers')
1680
self.assertEqual(test_selftest._add_numbers(2, 10),
1682
self.assertEquals(calls, [((2, 10), {})])
1685
def _add_numbers(a, b):
1689
class _MissingFeature(features.Feature):
1692
missing_feature = _MissingFeature()
1695
def _get_test(name):
1696
"""Get an instance of a specific example test.
1698
We protect this in a function so that they don't auto-run in the test
1702
class ExampleTests(tests.TestCase):
1704
def test_fail(self):
1705
mutter('this was a failing test')
1706
self.fail('this test will fail')
1708
def test_error(self):
1709
mutter('this test errored')
1710
raise RuntimeError('gotcha')
1712
def test_missing_feature(self):
1713
mutter('missing the feature')
1714
self.requireFeature(missing_feature)
1716
def test_skip(self):
1717
mutter('this test will be skipped')
1718
raise tests.TestSkipped('reason')
1720
def test_success(self):
1721
mutter('this test succeeds')
1723
def test_xfail(self):
1724
mutter('test with expected failure')
1725
self.knownFailure('this_fails')
1727
def test_unexpected_success(self):
1728
mutter('test with unexpected success')
1729
self.expectFailure('should_fail', lambda: None)
1731
return ExampleTests(name)
1734
class TestTestCaseLogDetails(tests.TestCase):
1736
def _run_test(self, test_name):
1737
test = _get_test(test_name)
1738
result = testtools.TestResult()
1742
def test_fail_has_log(self):
1743
result = self._run_test('test_fail')
1744
self.assertEqual(1, len(result.failures))
1745
result_content = result.failures[0][1]
1746
if testtools_version < (0, 9, 12):
1747
self.assertContainsRe(result_content, 'Text attachment: log')
1748
self.assertContainsRe(result_content, 'this was a failing test')
1750
def test_error_has_log(self):
1751
result = self._run_test('test_error')
1752
self.assertEqual(1, len(result.errors))
1753
result_content = result.errors[0][1]
1754
if testtools_version < (0, 9, 12):
1755
self.assertContainsRe(result_content, 'Text attachment: log')
1756
self.assertContainsRe(result_content, 'this test errored')
1758
def test_skip_has_no_log(self):
1759
result = self._run_test('test_skip')
1760
self.assertEqual(['reason'], result.skip_reasons.keys())
1761
skips = result.skip_reasons['reason']
1762
self.assertEqual(1, len(skips))
1764
self.assertFalse('log' in test.getDetails())
1766
def test_missing_feature_has_no_log(self):
1767
# testtools doesn't know about addNotSupported, so it just gets
1768
# considered as a skip
1769
result = self._run_test('test_missing_feature')
1770
self.assertEqual([missing_feature], result.skip_reasons.keys())
1771
skips = result.skip_reasons[missing_feature]
1772
self.assertEqual(1, len(skips))
1774
self.assertFalse('log' in test.getDetails())
1776
def test_xfail_has_no_log(self):
1777
result = self._run_test('test_xfail')
1778
self.assertEqual(1, len(result.expectedFailures))
1779
result_content = result.expectedFailures[0][1]
1780
self.assertNotContainsRe(result_content, 'Text attachment: log')
1781
self.assertNotContainsRe(result_content, 'test with expected failure')
1783
def test_unexpected_success_has_log(self):
1784
result = self._run_test('test_unexpected_success')
1785
self.assertEqual(1, len(result.unexpectedSuccesses))
1786
# Inconsistency, unexpectedSuccesses is a list of tests,
1787
# expectedFailures is a list of reasons?
1788
test = result.unexpectedSuccesses[0]
1789
details = test.getDetails()
1790
self.assertTrue('log' in details)
1793
class TestTestCloning(tests.TestCase):
1794
"""Tests that test cloning of TestCases (as used by multiply_tests)."""
1796
def test_cloned_testcase_does_not_share_details(self):
1797
"""A TestCase cloned with clone_test does not share mutable attributes
1798
such as details or cleanups.
1800
class Test(tests.TestCase):
1802
self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1803
orig_test = Test('test_foo')
1804
cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1805
orig_test.run(unittest.TestResult())
1806
self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1807
self.assertEqual(None, cloned_test.getDetails().get('foo'))
1809
def test_double_apply_scenario_preserves_first_scenario(self):
1810
"""Applying two levels of scenarios to a test preserves the attributes
1811
added by both scenarios.
1813
class Test(tests.TestCase):
1816
test = Test('test_foo')
1817
scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1818
scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1819
suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1820
suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1821
all_tests = list(tests.iter_suite_tests(suite))
1822
self.assertLength(4, all_tests)
1823
all_xys = sorted((t.x, t.y) for t in all_tests)
1824
self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1827
# NB: Don't delete this; it's not actually from 0.11!
1828
@deprecated_function(deprecated_in((0, 11, 0)))
1550
1829
def sample_deprecated_function():
1551
1830
"""A deprecated function to test applyDeprecated with."""
1662
1965
tree = self.make_branch_and_memory_tree('a')
1663
1966
self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1666
class TestSFTPMakeBranchAndTree(TestCaseWithSFTPServer):
1668
def test_make_tree_for_sftp_branch(self):
1669
"""Transports backed by local directories create local trees."""
1968
def test_make_tree_for_local_vfs_backed_transport(self):
1969
# make_branch_and_tree has to use local branch and repositories
1970
# when the vfs transport and local disk are colocated, even if
1971
# a different transport is in use for url generation.
1972
self.transport_server = test_server.FakeVFATServer
1973
self.assertFalse(self.get_url('t1').startswith('file://'))
1671
1974
tree = self.make_branch_and_tree('t1')
1672
1975
base = tree.bzrdir.root_transport.base
1673
self.failIf(base.startswith('sftp'),
1674
'base %r is on sftp but should be local' % base)
1976
self.assertStartsWith(base, 'file://')
1675
1977
self.assertEquals(tree.bzrdir.root_transport,
1676
1978
tree.branch.bzrdir.root_transport)
1677
1979
self.assertEquals(tree.bzrdir.root_transport,
1678
1980
tree.branch.repository.bzrdir.root_transport)
1681
class TestSelftest(TestCase):
1983
class SelfTestHelper(object):
1985
def run_selftest(self, **kwargs):
1986
"""Run selftest returning its output."""
1988
old_transport = bzrlib.tests.default_transport
1989
old_root = tests.TestCaseWithMemoryTransport.TEST_ROOT
1990
tests.TestCaseWithMemoryTransport.TEST_ROOT = None
1992
self.assertEqual(True, tests.selftest(stream=output, **kwargs))
1994
bzrlib.tests.default_transport = old_transport
1995
tests.TestCaseWithMemoryTransport.TEST_ROOT = old_root
2000
class TestSelftest(tests.TestCase, SelfTestHelper):
1682
2001
"""Tests of bzrlib.tests.selftest."""
1684
2003
def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
1685
2004
factory_called = []
1687
2006
factory_called.append(True)
2007
return TestUtil.TestSuite()
1689
2008
out = StringIO()
1690
2009
err = StringIO()
1691
self.apply_redirected(out, err, None, bzrlib.tests.selftest,
2010
self.apply_redirected(out, err, None, bzrlib.tests.selftest,
1692
2011
test_suite_factory=factory)
1693
2012
self.assertEqual([True], factory_called)
1696
class TestKnownFailure(TestCase):
1698
def test_known_failure(self):
1699
"""Check that KnownFailure is defined appropriately."""
1700
# a KnownFailure is an assertion error for compatability with unaware
1702
self.assertIsInstance(KnownFailure(""), AssertionError)
1704
def test_expect_failure(self):
1706
self.expectFailure("Doomed to failure", self.assertTrue, False)
1707
except KnownFailure, e:
1708
self.assertEqual('Doomed to failure', e.args[0])
1710
self.expectFailure("Doomed to failure", self.assertTrue, True)
1711
except AssertionError, e:
1712
self.assertEqual('Unexpected success. Should have failed:'
1713
' Doomed to failure', e.args[0])
2015
"""A test suite factory."""
2016
class Test(tests.TestCase):
2023
return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
2025
def test_list_only(self):
2026
output = self.run_selftest(test_suite_factory=self.factory,
2028
self.assertEqual(3, len(output.readlines()))
2030
def test_list_only_filtered(self):
2031
output = self.run_selftest(test_suite_factory=self.factory,
2032
list_only=True, pattern="Test.b")
2033
self.assertEndsWith(output.getvalue(), "Test.b\n")
2034
self.assertLength(1, output.readlines())
2036
def test_list_only_excludes(self):
2037
output = self.run_selftest(test_suite_factory=self.factory,
2038
list_only=True, exclude_pattern="Test.b")
2039
self.assertNotContainsRe("Test.b", output.getvalue())
2040
self.assertLength(2, output.readlines())
2042
def test_lsprof_tests(self):
2043
self.requireFeature(features.lsprof_feature)
2046
def __call__(test, result):
2048
def run(test, result):
2049
results.append(result)
2050
def countTestCases(self):
2052
self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
2053
self.assertLength(1, results)
2054
self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
2056
def test_random(self):
2057
# test randomising by listing a number of tests.
2058
output_123 = self.run_selftest(test_suite_factory=self.factory,
2059
list_only=True, random_seed="123")
2060
output_234 = self.run_selftest(test_suite_factory=self.factory,
2061
list_only=True, random_seed="234")
2062
self.assertNotEqual(output_123, output_234)
2063
# "Randominzing test order..\n\n
2064
self.assertLength(5, output_123.readlines())
2065
self.assertLength(5, output_234.readlines())
2067
def test_random_reuse_is_same_order(self):
2068
# test randomising by listing a number of tests.
2069
expected = self.run_selftest(test_suite_factory=self.factory,
2070
list_only=True, random_seed="123")
2071
repeated = self.run_selftest(test_suite_factory=self.factory,
2072
list_only=True, random_seed="123")
2073
self.assertEqual(expected.getvalue(), repeated.getvalue())
2075
def test_runner_class(self):
2076
self.requireFeature(features.subunit)
2077
from subunit import ProtocolTestCase
2078
stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2079
test_suite_factory=self.factory)
2080
test = ProtocolTestCase(stream)
2081
result = unittest.TestResult()
2083
self.assertEqual(3, result.testsRun)
2085
def test_starting_with_single_argument(self):
2086
output = self.run_selftest(test_suite_factory=self.factory,
2087
starting_with=['bzrlib.tests.test_selftest.Test.a'],
2089
self.assertEqual('bzrlib.tests.test_selftest.Test.a\n',
2092
def test_starting_with_multiple_argument(self):
2093
output = self.run_selftest(test_suite_factory=self.factory,
2094
starting_with=['bzrlib.tests.test_selftest.Test.a',
2095
'bzrlib.tests.test_selftest.Test.b'],
2097
self.assertEqual('bzrlib.tests.test_selftest.Test.a\n'
2098
'bzrlib.tests.test_selftest.Test.b\n',
2101
def check_transport_set(self, transport_server):
2102
captured_transport = []
2103
def seen_transport(a_transport):
2104
captured_transport.append(a_transport)
2105
class Capture(tests.TestCase):
2107
seen_transport(bzrlib.tests.default_transport)
2109
return TestUtil.TestSuite([Capture("a")])
2110
self.run_selftest(transport=transport_server, test_suite_factory=factory)
2111
self.assertEqual(transport_server, captured_transport[0])
2113
def test_transport_sftp(self):
2114
self.requireFeature(features.paramiko)
2115
from bzrlib.tests import stub_sftp
2116
self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
2118
def test_transport_memory(self):
2119
self.check_transport_set(memory.MemoryServer)
2122
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
2123
# Does IO: reads test.list
2125
def test_load_list(self):
2126
# Provide a list with one test - this test.
2127
test_id_line = '%s\n' % self.id()
2128
self.build_tree_contents([('test.list', test_id_line)])
2129
# And generate a list of the tests in the suite.
2130
stream = self.run_selftest(load_list='test.list', list_only=True)
2131
self.assertEqual(test_id_line, stream.getvalue())
2133
def test_load_unknown(self):
2134
# Provide a list with one test - this test.
2135
# And generate a list of the tests in the suite.
2136
err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
2137
load_list='missing file name', list_only=True)
2140
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2142
_test_needs_features = [features.subunit]
2144
def run_subunit_stream(self, test_name):
2145
from subunit import ProtocolTestCase
2147
return TestUtil.TestSuite([_get_test(test_name)])
2148
stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2149
test_suite_factory=factory)
2150
test = ProtocolTestCase(stream)
2151
result = testtools.TestResult()
2153
content = stream.getvalue()
2154
return content, result
2156
def test_fail_has_log(self):
2157
content, result = self.run_subunit_stream('test_fail')
2158
self.assertEqual(1, len(result.failures))
2159
self.assertContainsRe(content, '(?m)^log$')
2160
self.assertContainsRe(content, 'this test will fail')
2162
def test_error_has_log(self):
2163
content, result = self.run_subunit_stream('test_error')
2164
self.assertContainsRe(content, '(?m)^log$')
2165
self.assertContainsRe(content, 'this test errored')
2167
def test_skip_has_no_log(self):
2168
content, result = self.run_subunit_stream('test_skip')
2169
self.assertNotContainsRe(content, '(?m)^log$')
2170
self.assertNotContainsRe(content, 'this test will be skipped')
2171
self.assertEqual(['reason'], result.skip_reasons.keys())
2172
skips = result.skip_reasons['reason']
2173
self.assertEqual(1, len(skips))
2175
# RemotedTestCase doesn't preserve the "details"
2176
## self.assertFalse('log' in test.getDetails())
2178
def test_missing_feature_has_no_log(self):
2179
content, result = self.run_subunit_stream('test_missing_feature')
2180
self.assertNotContainsRe(content, '(?m)^log$')
2181
self.assertNotContainsRe(content, 'missing the feature')
2182
self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
2183
skips = result.skip_reasons['_MissingFeature\n']
2184
self.assertEqual(1, len(skips))
2186
# RemotedTestCase doesn't preserve the "details"
2187
## self.assertFalse('log' in test.getDetails())
2189
def test_xfail_has_no_log(self):
2190
content, result = self.run_subunit_stream('test_xfail')
2191
self.assertNotContainsRe(content, '(?m)^log$')
2192
self.assertNotContainsRe(content, 'test with expected failure')
2193
self.assertEqual(1, len(result.expectedFailures))
2194
result_content = result.expectedFailures[0][1]
2195
self.assertNotContainsRe(result_content, 'Text attachment: log')
2196
self.assertNotContainsRe(result_content, 'test with expected failure')
2198
def test_unexpected_success_has_log(self):
2199
content, result = self.run_subunit_stream('test_unexpected_success')
2200
self.assertContainsRe(content, '(?m)^log$')
2201
self.assertContainsRe(content, 'test with unexpected success')
2202
# GZ 2011-05-18: Old versions of subunit treat unexpected success as a
2203
# success, if a min version check is added remove this
2204
from subunit import TestProtocolClient as _Client
2205
if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
2206
self.expectFailure('subunit treats "unexpectedSuccess"'
2207
' as a plain success',
2208
self.assertEqual, 1, len(result.unexpectedSuccesses))
2209
self.assertEqual(1, len(result.unexpectedSuccesses))
2210
test = result.unexpectedSuccesses[0]
2211
# RemotedTestCase doesn't preserve the "details"
2212
## self.assertTrue('log' in test.getDetails())
2214
def test_success_has_no_log(self):
2215
content, result = self.run_subunit_stream('test_success')
2216
self.assertEqual(1, result.testsRun)
2217
self.assertNotContainsRe(content, '(?m)^log$')
2218
self.assertNotContainsRe(content, 'this test succeeds')
2221
class TestRunBzr(tests.TestCase):
2226
def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
2228
"""Override _run_bzr_core to test how it is invoked by run_bzr.
2230
Attempts to run bzr from inside this class don't actually run it.
2232
We test how run_bzr actually invokes bzr in another location. Here we
2233
only need to test that it passes the right parameters to run_bzr.
2235
self.argv = list(argv)
2236
self.retcode = retcode
2237
self.encoding = encoding
2239
self.working_dir = working_dir
2240
return self.retcode, self.out, self.err
2242
def test_run_bzr_error(self):
2243
self.out = "It sure does!\n"
2244
out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
2245
self.assertEqual(['rocks'], self.argv)
2246
self.assertEqual(34, self.retcode)
2247
self.assertEqual('It sure does!\n', out)
2248
self.assertEquals(out, self.out)
2249
self.assertEqual('', err)
2250
self.assertEquals(err, self.err)
2252
def test_run_bzr_error_regexes(self):
2254
self.err = "bzr: ERROR: foobarbaz is not versioned"
2255
out, err = self.run_bzr_error(
2256
["bzr: ERROR: foobarbaz is not versioned"],
2257
['file-id', 'foobarbaz'])
2259
def test_encoding(self):
2260
"""Test that run_bzr passes encoding to _run_bzr_core"""
2261
self.run_bzr('foo bar')
2262
self.assertEqual(None, self.encoding)
2263
self.assertEqual(['foo', 'bar'], self.argv)
2265
self.run_bzr('foo bar', encoding='baz')
2266
self.assertEqual('baz', self.encoding)
2267
self.assertEqual(['foo', 'bar'], self.argv)
2269
def test_retcode(self):
2270
"""Test that run_bzr passes retcode to _run_bzr_core"""
2271
# Default is retcode == 0
2272
self.run_bzr('foo bar')
2273
self.assertEqual(0, self.retcode)
2274
self.assertEqual(['foo', 'bar'], self.argv)
2276
self.run_bzr('foo bar', retcode=1)
2277
self.assertEqual(1, self.retcode)
2278
self.assertEqual(['foo', 'bar'], self.argv)
2280
self.run_bzr('foo bar', retcode=None)
2281
self.assertEqual(None, self.retcode)
2282
self.assertEqual(['foo', 'bar'], self.argv)
2284
self.run_bzr(['foo', 'bar'], retcode=3)
2285
self.assertEqual(3, self.retcode)
2286
self.assertEqual(['foo', 'bar'], self.argv)
2288
def test_stdin(self):
2289
# test that the stdin keyword to run_bzr is passed through to
2290
# _run_bzr_core as-is. We do this by overriding
2291
# _run_bzr_core in this class, and then calling run_bzr,
2292
# which is a convenience function for _run_bzr_core, so
2294
self.run_bzr('foo bar', stdin='gam')
2295
self.assertEqual('gam', self.stdin)
2296
self.assertEqual(['foo', 'bar'], self.argv)
2298
self.run_bzr('foo bar', stdin='zippy')
2299
self.assertEqual('zippy', self.stdin)
2300
self.assertEqual(['foo', 'bar'], self.argv)
2302
def test_working_dir(self):
2303
"""Test that run_bzr passes working_dir to _run_bzr_core"""
2304
self.run_bzr('foo bar')
2305
self.assertEqual(None, self.working_dir)
2306
self.assertEqual(['foo', 'bar'], self.argv)
2308
self.run_bzr('foo bar', working_dir='baz')
2309
self.assertEqual('baz', self.working_dir)
2310
self.assertEqual(['foo', 'bar'], self.argv)
2312
def test_reject_extra_keyword_arguments(self):
2313
self.assertRaises(TypeError, self.run_bzr, "foo bar",
2314
error_regex=['error message'])
2317
class TestRunBzrCaptured(tests.TestCaseWithTransport):
2318
# Does IO when testing the working_dir parameter.
2320
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
2321
a_callable=None, *args, **kwargs):
2323
self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
2324
self.factory = bzrlib.ui.ui_factory
2325
self.working_dir = osutils.getcwd()
2326
stdout.write('foo\n')
2327
stderr.write('bar\n')
2330
def test_stdin(self):
2331
# test that the stdin keyword to _run_bzr_core is passed through to
2332
# apply_redirected as a StringIO. We do this by overriding
2333
# apply_redirected in this class, and then calling _run_bzr_core,
2334
# which calls apply_redirected.
2335
self.run_bzr(['foo', 'bar'], stdin='gam')
2336
self.assertEqual('gam', self.stdin.read())
2337
self.assertTrue(self.stdin is self.factory_stdin)
2338
self.run_bzr(['foo', 'bar'], stdin='zippy')
2339
self.assertEqual('zippy', self.stdin.read())
2340
self.assertTrue(self.stdin is self.factory_stdin)
2342
def test_ui_factory(self):
2343
# each invocation of self.run_bzr should get its
2344
# own UI factory, which is an instance of TestUIFactory,
2345
# with stdin, stdout and stderr attached to the stdin,
2346
# stdout and stderr of the invoked run_bzr
2347
current_factory = bzrlib.ui.ui_factory
2348
self.run_bzr(['foo'])
2349
self.assertFalse(current_factory is self.factory)
2350
self.assertNotEqual(sys.stdout, self.factory.stdout)
2351
self.assertNotEqual(sys.stderr, self.factory.stderr)
2352
self.assertEqual('foo\n', self.factory.stdout.getvalue())
2353
self.assertEqual('bar\n', self.factory.stderr.getvalue())
2354
self.assertIsInstance(self.factory, tests.TestUIFactory)
2356
def test_working_dir(self):
2357
self.build_tree(['one/', 'two/'])
2358
cwd = osutils.getcwd()
2360
# Default is to work in the current directory
2361
self.run_bzr(['foo', 'bar'])
2362
self.assertEqual(cwd, self.working_dir)
2364
self.run_bzr(['foo', 'bar'], working_dir=None)
2365
self.assertEqual(cwd, self.working_dir)
2367
# The function should be run in the alternative directory
2368
# but afterwards the current working dir shouldn't be changed
2369
self.run_bzr(['foo', 'bar'], working_dir='one')
2370
self.assertNotEqual(cwd, self.working_dir)
2371
self.assertEndsWith(self.working_dir, 'one')
2372
self.assertEqual(cwd, osutils.getcwd())
2374
self.run_bzr(['foo', 'bar'], working_dir='two')
2375
self.assertNotEqual(cwd, self.working_dir)
2376
self.assertEndsWith(self.working_dir, 'two')
2377
self.assertEqual(cwd, osutils.getcwd())
2380
class StubProcess(object):
2381
"""A stub process for testing run_bzr_subprocess."""
2383
def __init__(self, out="", err="", retcode=0):
2386
self.returncode = retcode
2388
def communicate(self):
2389
return self.out, self.err
2392
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
2393
"""Base class for tests testing how we might run bzr."""
2396
tests.TestCaseWithTransport.setUp(self)
2397
self.subprocess_calls = []
2399
def start_bzr_subprocess(self, process_args, env_changes=None,
2400
skip_if_plan_to_signal=False,
2402
allow_plugins=False):
2403
"""capture what run_bzr_subprocess tries to do."""
2404
self.subprocess_calls.append({'process_args':process_args,
2405
'env_changes':env_changes,
2406
'skip_if_plan_to_signal':skip_if_plan_to_signal,
2407
'working_dir':working_dir, 'allow_plugins':allow_plugins})
2408
return self.next_subprocess
2411
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
2413
def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
2414
"""Run run_bzr_subprocess with args and kwargs using a stubbed process.
2416
Inside TestRunBzrSubprocessCommands we use a stub start_bzr_subprocess
2417
that will return static results. This assertion method populates those
2418
results and also checks the arguments run_bzr_subprocess generates.
2420
self.next_subprocess = process
2422
result = self.run_bzr_subprocess(*args, **kwargs)
2424
self.next_subprocess = None
2425
for key, expected in expected_args.iteritems():
2426
self.assertEqual(expected, self.subprocess_calls[-1][key])
1715
self.fail('Assertion not raised')
1718
class TestFeature(TestCase):
1720
def test_caching(self):
1721
"""Feature._probe is called by the feature at most once."""
1722
class InstrumentedFeature(Feature):
1724
Feature.__init__(self)
1727
self.calls.append('_probe')
1729
feature = InstrumentedFeature()
1731
self.assertEqual(['_probe'], feature.calls)
1733
self.assertEqual(['_probe'], feature.calls)
1735
def test_named_str(self):
1736
"""Feature.__str__ should thunk to feature_name()."""
1737
class NamedFeature(Feature):
1738
def feature_name(self):
1740
feature = NamedFeature()
1741
self.assertEqual('symlinks', str(feature))
1743
def test_default_str(self):
1744
"""Feature.__str__ should default to __class__.__name__."""
1745
class NamedFeature(Feature):
1747
feature = NamedFeature()
1748
self.assertEqual('NamedFeature', str(feature))
1751
class TestUnavailableFeature(TestCase):
1753
def test_access_feature(self):
1755
exception = UnavailableFeature(feature)
1756
self.assertIs(feature, exception.args[0])
1759
class TestSelftestFiltering(TestCase):
2429
self.next_subprocess = None
2430
for key, expected in expected_args.iteritems():
2431
self.assertEqual(expected, self.subprocess_calls[-1][key])
2434
def test_run_bzr_subprocess(self):
2435
"""The run_bzr_helper_external command behaves nicely."""
2436
self.assertRunBzrSubprocess({'process_args':['--version']},
2437
StubProcess(), '--version')
2438
self.assertRunBzrSubprocess({'process_args':['--version']},
2439
StubProcess(), ['--version'])
2440
# retcode=None disables retcode checking
2441
result = self.assertRunBzrSubprocess({},
2442
StubProcess(retcode=3), '--version', retcode=None)
2443
result = self.assertRunBzrSubprocess({},
2444
StubProcess(out="is free software"), '--version')
2445
self.assertContainsRe(result[0], 'is free software')
2446
# Running a subcommand that is missing errors
2447
self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
2448
{'process_args':['--versionn']}, StubProcess(retcode=3),
2450
# Unless it is told to expect the error from the subprocess
2451
result = self.assertRunBzrSubprocess({},
2452
StubProcess(retcode=3), '--versionn', retcode=3)
2453
# Or to ignore retcode checking
2454
result = self.assertRunBzrSubprocess({},
2455
StubProcess(err="unknown command", retcode=3), '--versionn',
2457
self.assertContainsRe(result[1], 'unknown command')
2459
def test_env_change_passes_through(self):
2460
self.assertRunBzrSubprocess(
2461
{'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
2463
env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
2465
def test_no_working_dir_passed_as_None(self):
2466
self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
2468
def test_no_working_dir_passed_through(self):
2469
self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
2472
def test_run_bzr_subprocess_no_plugins(self):
2473
self.assertRunBzrSubprocess({'allow_plugins': False},
2476
def test_allow_plugins(self):
2477
self.assertRunBzrSubprocess({'allow_plugins': True},
2478
StubProcess(), '', allow_plugins=True)
2481
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2483
def test_finish_bzr_subprocess_with_error(self):
2484
"""finish_bzr_subprocess allows specification of the desired exit code.
2486
process = StubProcess(err="unknown command", retcode=3)
2487
result = self.finish_bzr_subprocess(process, retcode=3)
2488
self.assertEqual('', result[0])
2489
self.assertContainsRe(result[1], 'unknown command')
2491
def test_finish_bzr_subprocess_ignoring_retcode(self):
2492
"""finish_bzr_subprocess allows the exit code to be ignored."""
2493
process = StubProcess(err="unknown command", retcode=3)
2494
result = self.finish_bzr_subprocess(process, retcode=None)
2495
self.assertEqual('', result[0])
2496
self.assertContainsRe(result[1], 'unknown command')
2498
def test_finish_subprocess_with_unexpected_retcode(self):
2499
"""finish_bzr_subprocess raises self.failureException if the retcode is
2500
not the expected one.
2502
process = StubProcess(err="unknown command", retcode=3)
2503
self.assertRaises(self.failureException, self.finish_bzr_subprocess,
2507
class _DontSpawnProcess(Exception):
2508
"""A simple exception which just allows us to skip unnecessary steps"""
2511
class TestStartBzrSubProcess(tests.TestCase):
2512
"""Stub test start_bzr_subprocess."""
2514
def _subprocess_log_cleanup(self):
2515
"""Inhibits the base version as we don't produce a log file."""
2517
def _popen(self, *args, **kwargs):
2518
"""Override the base version to record the command that is run.
2520
From there we can ensure it is correct without spawning a real process.
2522
self.check_popen_state()
2523
self._popen_args = args
2524
self._popen_kwargs = kwargs
2525
raise _DontSpawnProcess()
2527
def check_popen_state(self):
2528
"""Replace to make assertions when popen is called."""
2530
def test_run_bzr_subprocess_no_plugins(self):
2531
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2532
command = self._popen_args[0]
2533
self.assertEqual(sys.executable, command[0])
2534
self.assertEqual(self.get_bzr_path(), command[1])
2535
self.assertEqual(['--no-plugins'], command[2:])
2537
def test_allow_plugins(self):
2538
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2540
command = self._popen_args[0]
2541
self.assertEqual([], command[2:])
2543
def test_set_env(self):
2544
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2546
def check_environment():
2547
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2548
self.check_popen_state = check_environment
2549
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2550
env_changes={'EXISTANT_ENV_VAR':'set variable'})
2551
# not set in theparent
2552
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2554
def test_run_bzr_subprocess_env_del(self):
2555
"""run_bzr_subprocess can remove environment variables too."""
2556
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2557
def check_environment():
2558
self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2559
os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2560
self.check_popen_state = check_environment
2561
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2562
env_changes={'EXISTANT_ENV_VAR':None})
2563
# Still set in parent
2564
self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2565
del os.environ['EXISTANT_ENV_VAR']
2567
def test_env_del_missing(self):
2568
self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2569
def check_environment():
2570
self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2571
self.check_popen_state = check_environment
2572
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2573
env_changes={'NON_EXISTANT_ENV_VAR':None})
2575
def test_working_dir(self):
2576
"""Test that we can specify the working dir for the child"""
2577
orig_getcwd = osutils.getcwd
2578
orig_chdir = os.chdir
2582
self.overrideAttr(os, 'chdir', chdir)
2585
self.overrideAttr(osutils, 'getcwd', getcwd)
2586
self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2588
self.assertEqual(['foo', 'current'], chdirs)
2590
def test_get_bzr_path_with_cwd_bzrlib(self):
2591
self.get_source_path = lambda: ""
2592
self.overrideAttr(os.path, "isfile", lambda path: True)
2593
self.assertEqual(self.get_bzr_path(), "bzr")
2596
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2597
"""Tests that really need to do things with an external bzr."""
2599
def test_start_and_stop_bzr_subprocess_send_signal(self):
2600
"""finish_bzr_subprocess raises self.failureException if the retcode is
2601
not the expected one.
2603
self.disable_missing_extensions_warning()
2604
process = self.start_bzr_subprocess(['wait-until-signalled'],
2605
skip_if_plan_to_signal=True)
2606
self.assertEqual('running\n', process.stdout.readline())
2607
result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
2609
self.assertEqual('', result[0])
2610
self.assertEqual('bzr: interrupted\n', result[1])
2613
class TestSelftestFiltering(tests.TestCase):
1761
2615
def setUp(self):
2616
tests.TestCase.setUp(self)
1762
2617
self.suite = TestUtil.TestSuite()
1763
2618
self.loader = TestUtil.TestLoader()
1764
self.suite.addTest(self.loader.loadTestsFromModuleNames([
1765
'bzrlib.tests.test_selftest']))
2619
self.suite.addTest(self.loader.loadTestsFromModule(
2620
sys.modules['bzrlib.tests.test_selftest']))
1766
2621
self.all_names = _test_ids(self.suite)
1768
2623
def test_condition_id_re(self):
1769
2624
test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
1770
2625
'test_condition_id_re')
1771
filtered_suite = filter_suite_by_condition(self.suite,
1772
condition_id_re('test_condition_id_re'))
2626
filtered_suite = tests.filter_suite_by_condition(
2627
self.suite, tests.condition_id_re('test_condition_id_re'))
1773
2628
self.assertEqual([test_name], _test_ids(filtered_suite))
1775
2630
def test_condition_id_in_list(self):
1776
2631
test_names = ['bzrlib.tests.test_selftest.TestSelftestFiltering.'
1777
2632
'test_condition_id_in_list']
1778
2633
id_list = tests.TestIdList(test_names)
1779
filtered_suite = filter_suite_by_condition(
2634
filtered_suite = tests.filter_suite_by_condition(
1780
2635
self.suite, tests.condition_id_in_list(id_list))
1781
2636
my_pattern = 'TestSelftestFiltering.*test_condition_id_in_list'
1782
re_filtered = filter_suite_by_re(self.suite, my_pattern)
2637
re_filtered = tests.filter_suite_by_re(self.suite, my_pattern)
1783
2638
self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
1785
2640
def test_condition_id_startswith(self):
2217
3138
self.assertEquals('bzrlib.tests', tpr.resolve_alias('bt'))
2218
3139
self.assertEquals('bzrlib.tests.blackbox', tpr.resolve_alias('bb'))
2219
3140
self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3143
class TestThreadLeakDetection(tests.TestCase):
3144
"""Ensure when tests leak threads we detect and report it"""
3146
class LeakRecordingResult(tests.ExtendedTestResult):
3148
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3150
def _report_thread_leak(self, test, leaks, alive):
3151
self.leaks.append((test, leaks))
3153
def test_testcase_without_addCleanups(self):
3154
"""Check old TestCase instances don't break with leak detection"""
3155
class Test(unittest.TestCase):
3158
result = self.LeakRecordingResult()
3160
result.startTestRun()
3162
result.stopTestRun()
3163
self.assertEqual(result._tests_leaking_threads_count, 0)
3164
self.assertEqual(result.leaks, [])
3166
def test_thread_leak(self):
3167
"""Ensure a thread that outlives the running of a test is reported
3169
Uses a thread that blocks on an event, and is started by the inner
3170
test case. As the thread outlives the inner case's run, it should be
3171
detected as a leak, but the event is then set so that the thread can
3172
be safely joined in cleanup so it's not leaked for real.
3174
event = threading.Event()
3175
thread = threading.Thread(name="Leaker", target=event.wait)
3176
class Test(tests.TestCase):
3177
def test_leak(self):
3179
result = self.LeakRecordingResult()
3180
test = Test("test_leak")
3181
self.addCleanup(thread.join)
3182
self.addCleanup(event.set)
3183
result.startTestRun()
3185
result.stopTestRun()
3186
self.assertEqual(result._tests_leaking_threads_count, 1)
3187
self.assertEqual(result._first_thread_leaker_id, test.id())
3188
self.assertEqual(result.leaks, [(test, set([thread]))])
3189
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3191
def test_multiple_leaks(self):
3192
"""Check multiple leaks are blamed on the test cases at fault
3194
Same concept as the previous test, but has one inner test method that
3195
leaks two threads, and one that doesn't leak at all.
3197
event = threading.Event()
3198
thread_a = threading.Thread(name="LeakerA", target=event.wait)
3199
thread_b = threading.Thread(name="LeakerB", target=event.wait)
3200
thread_c = threading.Thread(name="LeakerC", target=event.wait)
3201
class Test(tests.TestCase):
3202
def test_first_leak(self):
3204
def test_second_no_leak(self):
3206
def test_third_leak(self):
3209
result = self.LeakRecordingResult()
3210
first_test = Test("test_first_leak")
3211
third_test = Test("test_third_leak")
3212
self.addCleanup(thread_a.join)
3213
self.addCleanup(thread_b.join)
3214
self.addCleanup(thread_c.join)
3215
self.addCleanup(event.set)
3216
result.startTestRun()
3218
[first_test, Test("test_second_no_leak"), third_test]
3220
result.stopTestRun()
3221
self.assertEqual(result._tests_leaking_threads_count, 2)
3222
self.assertEqual(result._first_thread_leaker_id, first_test.id())
3223
self.assertEqual(result.leaks, [
3224
(first_test, set([thread_b])),
3225
(third_test, set([thread_a, thread_c]))])
3226
self.assertContainsString(result.stream.getvalue(), "leaking threads")
3229
class TestPostMortemDebugging(tests.TestCase):
3230
"""Check post mortem debugging works when tests fail or error"""
3232
class TracebackRecordingResult(tests.ExtendedTestResult):
3234
tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3235
self.postcode = None
3236
def _post_mortem(self, tb=None):
3237
"""Record the code object at the end of the current traceback"""
3238
tb = tb or sys.exc_info()[2]
3241
while next is not None:
3244
self.postcode = tb.tb_frame.f_code
3245
def report_error(self, test, err):
3247
def report_failure(self, test, err):
3250
def test_location_unittest_error(self):
3251
"""Needs right post mortem traceback with erroring unittest case"""
3252
class Test(unittest.TestCase):
3255
result = self.TracebackRecordingResult()
3257
self.assertEqual(result.postcode, Test.runTest.func_code)
3259
def test_location_unittest_failure(self):
3260
"""Needs right post mortem traceback with failing unittest case"""
3261
class Test(unittest.TestCase):
3263
raise self.failureException
3264
result = self.TracebackRecordingResult()
3266
self.assertEqual(result.postcode, Test.runTest.func_code)
3268
def test_location_bt_error(self):
3269
"""Needs right post mortem traceback with erroring bzrlib.tests case"""
3270
class Test(tests.TestCase):
3271
def test_error(self):
3273
result = self.TracebackRecordingResult()
3274
Test("test_error").run(result)
3275
self.assertEqual(result.postcode, Test.test_error.func_code)
3277
def test_location_bt_failure(self):
3278
"""Needs right post mortem traceback with failing bzrlib.tests case"""
3279
class Test(tests.TestCase):
3280
def test_failure(self):
3281
raise self.failureException
3282
result = self.TracebackRecordingResult()
3283
Test("test_failure").run(result)
3284
self.assertEqual(result.postcode, Test.test_failure.func_code)
3286
def test_env_var_triggers_post_mortem(self):
3287
"""Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
3289
result = tests.ExtendedTestResult(StringIO(), 0, 1)
3290
post_mortem_calls = []
3291
self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3292
self.overrideEnv('BZR_TEST_PDB', None)
3293
result._post_mortem(1)
3294
self.overrideEnv('BZR_TEST_PDB', 'on')
3295
result._post_mortem(2)
3296
self.assertEqual([2], post_mortem_calls)
3299
class TestRunSuite(tests.TestCase):
3301
def test_runner_class(self):
3302
"""run_suite accepts and uses a runner_class keyword argument."""
3303
class Stub(tests.TestCase):
3306
suite = Stub("test_foo")
3308
class MyRunner(tests.TextTestRunner):
3309
def run(self, test):
3311
return tests.ExtendedTestResult(self.stream, self.descriptions,
3313
tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
3314
self.assertLength(1, calls)
3317
class TestEnvironHandling(tests.TestCase):
3319
def test_overrideEnv_None_called_twice_doesnt_leak(self):
3320
self.assertFalse('MYVAR' in os.environ)
3321
self.overrideEnv('MYVAR', '42')
3322
# We use an embedded test to make sure we fix the _captureVar bug
3323
class Test(tests.TestCase):
3325
# The first call save the 42 value
3326
self.overrideEnv('MYVAR', None)
3327
self.assertEquals(None, os.environ.get('MYVAR'))
3328
# Make sure we can call it twice
3329
self.overrideEnv('MYVAR', None)
3330
self.assertEquals(None, os.environ.get('MYVAR'))
3332
result = tests.TextTestResult(output, 0, 1)
3333
Test('test_me').run(result)
3334
if not result.wasStrictlySuccessful():
3335
self.fail(output.getvalue())
3336
# We get our value back
3337
self.assertEquals('42', os.environ.get('MYVAR'))
3340
class TestIsolatedEnv(tests.TestCase):
3341
"""Test isolating tests from os.environ.
3343
Since we use tests that are already isolated from os.environ a bit of care
3344
should be taken when designing the tests to avoid bootstrap side-effects.
3345
The tests start an already clean os.environ which allow doing valid
3346
assertions about which variables are present or not and design tests around
3350
class ScratchMonkey(tests.TestCase):
3355
def test_basics(self):
3356
# Make sure we know the definition of BZR_HOME: not part of os.environ
3357
# for tests.TestCase.
3358
self.assertTrue('BZR_HOME' in tests.isolated_environ)
3359
self.assertEquals(None, tests.isolated_environ['BZR_HOME'])
3360
# Being part of isolated_environ, BZR_HOME should not appear here
3361
self.assertFalse('BZR_HOME' in os.environ)
3362
# Make sure we know the definition of LINES: part of os.environ for
3364
self.assertTrue('LINES' in tests.isolated_environ)
3365
self.assertEquals('25', tests.isolated_environ['LINES'])
3366
self.assertEquals('25', os.environ['LINES'])
3368
def test_injecting_unknown_variable(self):
3369
# BZR_HOME is known to be absent from os.environ
3370
test = self.ScratchMonkey('test_me')
3371
tests.override_os_environ(test, {'BZR_HOME': 'foo'})
3372
self.assertEquals('foo', os.environ['BZR_HOME'])
3373
tests.restore_os_environ(test)
3374
self.assertFalse('BZR_HOME' in os.environ)
3376
def test_injecting_known_variable(self):
3377
test = self.ScratchMonkey('test_me')
3378
# LINES is known to be present in os.environ
3379
tests.override_os_environ(test, {'LINES': '42'})
3380
self.assertEquals('42', os.environ['LINES'])
3381
tests.restore_os_environ(test)
3382
self.assertEquals('25', os.environ['LINES'])
3384
def test_deleting_variable(self):
3385
test = self.ScratchMonkey('test_me')
3386
# LINES is known to be present in os.environ
3387
tests.override_os_environ(test, {'LINES': None})
3388
self.assertTrue('LINES' not in os.environ)
3389
tests.restore_os_environ(test)
3390
self.assertEquals('25', os.environ['LINES'])
3393
class TestDocTestSuiteIsolation(tests.TestCase):
3394
"""Test that `tests.DocTestSuite` isolates doc tests from os.environ.
3396
Since tests.TestCase alreay provides an isolation from os.environ, we use
3397
the clean environment as a base for testing. To precisely capture the
3398
isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
3401
We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
3402
not `os.environ` so each test overrides it to suit its needs.
3406
def get_doctest_suite_for_string(self, klass, string):
3407
class Finder(doctest.DocTestFinder):
3409
def find(*args, **kwargs):
3410
test = doctest.DocTestParser().get_doctest(
3411
string, {}, 'foo', 'foo.py', 0)
3414
suite = klass(test_finder=Finder())
3417
def run_doctest_suite_for_string(self, klass, string):
3418
suite = self.get_doctest_suite_for_string(klass, string)
3420
result = tests.TextTestResult(output, 0, 1)
3422
return result, output
3424
def assertDocTestStringSucceds(self, klass, string):
3425
result, output = self.run_doctest_suite_for_string(klass, string)
3426
if not result.wasStrictlySuccessful():
3427
self.fail(output.getvalue())
3429
def assertDocTestStringFails(self, klass, string):
3430
result, output = self.run_doctest_suite_for_string(klass, string)
3431
if result.wasStrictlySuccessful():
3432
self.fail(output.getvalue())
3434
def test_injected_variable(self):
3435
self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
3438
>>> os.environ['LINES']
3441
# doctest.DocTestSuite fails as it sees '25'
3442
self.assertDocTestStringFails(doctest.DocTestSuite, test)
3443
# tests.DocTestSuite sees '42'
3444
self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3446
def test_deleted_variable(self):
3447
self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
3450
>>> os.environ.get('LINES')
3452
# doctest.DocTestSuite fails as it sees '25'
3453
self.assertDocTestStringFails(doctest.DocTestSuite, test)
3454
# tests.DocTestSuite sees None
3455
self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3458
class TestSelftestExcludePatterns(tests.TestCase):
3461
super(TestSelftestExcludePatterns, self).setUp()
3462
self.overrideAttr(tests, 'test_suite', self.suite_factory)
3464
def suite_factory(self, keep_only=None, starting_with=None):
3465
"""A test suite factory with only a few tests."""
3466
class Test(tests.TestCase):
3468
# We don't need the full class path
3469
return self._testMethodName
3476
return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
3478
def assertTestList(self, expected, *selftest_args):
3479
# We rely on setUp installing the right test suite factory so we can
3480
# test at the command level without loading the whole test suite
3481
out, err = self.run_bzr(('selftest', '--list') + selftest_args)
3482
actual = out.splitlines()
3483
self.assertEquals(expected, actual)
3485
def test_full_list(self):
3486
self.assertTestList(['a', 'b', 'c'])
3488
def test_single_exclude(self):
3489
self.assertTestList(['b', 'c'], '-x', 'a')
3491
def test_mutiple_excludes(self):
3492
self.assertTestList(['c'], '-x', 'a', '-x', 'b')
3495
class TestCounterHooks(tests.TestCase, SelfTestHelper):
3497
_test_needs_features = [features.subunit]
3500
super(TestCounterHooks, self).setUp()
3501
class Test(tests.TestCase):
3504
super(Test, self).setUp()
3505
self.hooks = hooks.Hooks()
3506
self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
3507
self.install_counter_hook(self.hooks, 'myhook')
3512
def run_hook_once(self):
3513
for hook in self.hooks['myhook']:
3516
self.test_class = Test
3518
def assertHookCalls(self, expected_calls, test_name):
3519
test = self.test_class(test_name)
3520
result = unittest.TestResult()
3522
self.assertTrue(hasattr(test, '_counters'))
3523
self.assertTrue(test._counters.has_key('myhook'))
3524
self.assertEquals(expected_calls, test._counters['myhook'])
3526
def test_no_hook(self):
3527
self.assertHookCalls(0, 'no_hook')
3529
def test_run_hook_once(self):
3530
tt = features.testtools
3531
if tt.module.__version__ < (0, 9, 8):
3532
raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
3533
self.assertHookCalls(1, 'run_hook_once')