~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: John Arbash Meinel
  • Author(s): Mark Hammond
  • Date: 2008-09-09 17:02:21 UTC
  • mto: This revision was merged to the branch mainline in revision 3697.
  • Revision ID: john@arbash-meinel.com-20080909170221-svim3jw2mrz0amp3
An updated transparent icon for bzr.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2013, 2016 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007, 2008 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
12
12
#
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., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
16
16
 
17
17
"""Tests for the test framework."""
18
18
 
19
 
from cStringIO import StringIO
20
 
import gc
21
 
import doctest
 
19
import cStringIO
22
20
import os
23
 
import signal
 
21
from StringIO import StringIO
24
22
import sys
25
 
import threading
26
23
import time
27
24
import unittest
28
25
import warnings
29
26
 
30
 
from testtools import (
31
 
    ExtendedToOriginalDecorator,
32
 
    MultiTestResult,
33
 
    )
34
 
from testtools.content import Content
35
 
from testtools.content_type import ContentType
36
 
from testtools.matchers import (
37
 
    DocTestMatches,
38
 
    Equals,
39
 
    )
40
 
import testtools.testresult.doubles
41
 
 
42
27
import bzrlib
43
28
from bzrlib import (
44
29
    branchbuilder,
45
30
    bzrdir,
46
 
    controldir,
47
31
    errors,
48
 
    hooks,
49
 
    lockdir,
50
32
    memorytree,
51
33
    osutils,
52
34
    remote,
53
35
    repository,
54
36
    symbol_versioning,
55
37
    tests,
56
 
    transport,
57
38
    workingtree,
58
 
    workingtree_3,
59
 
    workingtree_4,
60
39
    )
 
40
from bzrlib.progress import _BaseProgressBar
61
41
from bzrlib.repofmt import (
62
 
    groupcompress_repo,
 
42
    pack_repo,
 
43
    weaverepo,
63
44
    )
64
45
from bzrlib.symbol_versioning import (
65
 
    deprecated_function,
66
 
    deprecated_in,
67
 
    deprecated_method,
 
46
    one_zero,
 
47
    zero_eleven,
 
48
    zero_ten,
68
49
    )
69
50
from bzrlib.tests import (
70
 
    features,
71
 
    test_lsprof,
72
 
    test_server,
73
 
    TestUtil,
74
 
    )
75
 
from bzrlib.trace import note, mutter
76
 
from bzrlib.transport import memory
 
51
                          ChrootedTestCase,
 
52
                          ExtendedTestResult,
 
53
                          Feature,
 
54
                          KnownFailure,
 
55
                          TestCase,
 
56
                          TestCaseInTempDir,
 
57
                          TestCaseWithMemoryTransport,
 
58
                          TestCaseWithTransport,
 
59
                          TestNotApplicable,
 
60
                          TestSkipped,
 
61
                          TestSuite,
 
62
                          TestUtil,
 
63
                          TextTestRunner,
 
64
                          UnavailableFeature,
 
65
                          condition_id_re,
 
66
                          condition_isinstance,
 
67
                          exclude_tests_by_condition,
 
68
                          exclude_tests_by_re,
 
69
                          filter_suite_by_condition,
 
70
                          filter_suite_by_re,
 
71
                          iter_suite_tests,
 
72
                          preserve_input,
 
73
                          randomize_suite,
 
74
                          split_suite_by_condition,
 
75
                          split_suite_by_re,
 
76
                          test_lsprof,
 
77
                          test_suite,
 
78
                          )
 
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
77
84
 
78
85
 
79
86
def _test_ids(test_suite):
80
87
    """Get the ids for the tests in a test suite."""
81
 
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
82
 
 
83
 
 
84
 
class MetaTestLog(tests.TestCase):
 
88
    return [t.id() for t in iter_suite_tests(test_suite)]
 
89
 
 
90
 
 
91
class SelftestTests(TestCase):
 
92
 
 
93
    def test_import_tests(self):
 
94
        mod = _load_module_by_name('bzrlib.tests.test_selftest')
 
95
        self.assertEqual(mod.SelftestTests, SelftestTests)
 
96
 
 
97
    def test_import_test_failure(self):
 
98
        self.assertRaises(ImportError,
 
99
                          _load_module_by_name,
 
100
                          'bzrlib.no-name-yet')
 
101
 
 
102
class MetaTestLog(TestCase):
85
103
 
86
104
    def test_logging(self):
87
105
        """Test logs are captured when a test fails."""
88
106
        self.log('a test message')
89
 
        details = self.getDetails()
90
 
        log = details['log']
91
 
        self.assertThat(log.content_type, Equals(ContentType(
92
 
            "text", "plain", {"charset": "utf8"})))
93
 
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
94
 
        self.assertThat(self.get_log(),
95
 
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
96
 
 
97
 
 
98
 
class TestTreeShape(tests.TestCaseInTempDir):
 
107
        self._log_file.flush()
 
108
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
109
                              'a test message\n')
 
110
 
 
111
 
 
112
class TestUnicodeFilename(TestCase):
 
113
 
 
114
    def test_probe_passes(self):
 
115
        """UnicodeFilename._probe passes."""
 
116
        # We can't test much more than that because the behaviour depends
 
117
        # on the platform.
 
118
        tests.UnicodeFilename._probe()
 
119
 
 
120
 
 
121
class TestTreeShape(TestCaseInTempDir):
99
122
 
100
123
    def test_unicode_paths(self):
101
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
124
        self.requireFeature(tests.UnicodeFilename)
102
125
 
103
126
        filename = u'hell\u00d8'
104
127
        self.build_tree_contents([(filename, 'contents of hello')])
105
 
        self.assertPathExists(filename)
106
 
 
107
 
 
108
 
class TestClassesAvailable(tests.TestCase):
109
 
    """As a convenience we expose Test* classes from bzrlib.tests"""
110
 
 
111
 
    def test_test_case(self):
112
 
        from bzrlib.tests import TestCase
113
 
 
114
 
    def test_test_loader(self):
115
 
        from bzrlib.tests import TestLoader
116
 
 
117
 
    def test_test_suite(self):
118
 
        from bzrlib.tests import TestSuite
119
 
 
120
 
 
121
 
class TestTransportScenarios(tests.TestCase):
 
128
        self.failUnlessExists(filename)
 
129
 
 
130
 
 
131
class TestTransportProviderAdapter(TestCase):
122
132
    """A group of tests that test the transport implementation adaption core.
123
133
 
124
 
    This is a meta test that the tests are applied to all available
 
134
    This is a meta test that the tests are applied to all available 
125
135
    transports.
126
136
 
127
 
    This will be generalised in the future which is why it is in this
 
137
    This will be generalised in the future which is why it is in this 
128
138
    test file even though it is specific to transport tests at the moment.
129
139
    """
130
140
 
131
141
    def test_get_transport_permutations(self):
132
142
        # this checks that get_test_permutations defined by the module is
133
 
        # called by the get_transport_test_permutations function.
 
143
        # called by the adapter get_transport_test_permutations method.
134
144
        class MockModule(object):
135
145
            def get_test_permutations(self):
136
146
                return sample_permutation
137
147
        sample_permutation = [(1,2), (3,4)]
138
 
        from bzrlib.tests.per_transport import get_transport_test_permutations
 
148
        from bzrlib.tests.test_transport_implementations \
 
149
            import TransportTestProviderAdapter
 
150
        adapter = TransportTestProviderAdapter()
139
151
        self.assertEqual(sample_permutation,
140
 
                         get_transport_test_permutations(MockModule()))
 
152
                         adapter.get_transport_test_permutations(MockModule()))
141
153
 
142
 
    def test_scenarios_include_all_modules(self):
143
 
        # this checks that the scenario generator returns as many permutations
144
 
        # as there are in all the registered transport modules - we assume if
145
 
        # this matches its probably doing the right thing especially in
146
 
        # combination with the tests for setting the right classes below.
147
 
        from bzrlib.tests.per_transport import transport_test_permutations
 
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
148
161
        from bzrlib.transport import _get_transport_modules
149
162
        modules = _get_transport_modules()
150
163
        permutation_count = 0
151
164
        for module in modules:
152
165
            try:
153
 
                permutation_count += len(reduce(getattr,
 
166
                permutation_count += len(reduce(getattr, 
154
167
                    (module + ".get_test_permutations").split('.')[1:],
155
168
                     __import__(module))())
156
169
            except errors.DependencyNotPresent:
157
170
                pass
158
 
        scenarios = transport_test_permutations()
159
 
        self.assertEqual(permutation_count, len(scenarios))
 
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)))))
160
176
 
161
 
    def test_scenarios_include_transport_class(self):
 
177
    def test_adapter_sets_transport_class(self):
 
178
        # Check that the test adapter inserts a transport and server into the
 
179
        # generated test.
 
180
        #
162
181
        # This test used to know about all the possible transports and the
163
182
        # order they were returned but that seems overly brittle (mbp
164
183
        # 20060307)
165
 
        from bzrlib.tests.per_transport import transport_test_permutations
166
 
        scenarios = transport_test_permutations()
 
184
        from bzrlib.tests.test_transport_implementations \
 
185
            import TransportTestProviderAdapter
 
186
        scenarios = TransportTestProviderAdapter().scenarios
167
187
        # there are at least that many builtin transports
168
188
        self.assertTrue(len(scenarios) > 6)
169
189
        one_scenario = scenarios[0]
174
194
                                   bzrlib.transport.Server))
175
195
 
176
196
 
177
 
class TestBranchScenarios(tests.TestCase):
 
197
class TestBranchProviderAdapter(TestCase):
 
198
    """A group of tests that test the branch implementation test adapter."""
178
199
 
179
 
    def test_scenarios(self):
 
200
    def test_constructor(self):
180
201
        # check that constructor parameters are passed through to the adapted
181
202
        # test.
182
 
        from bzrlib.tests.per_branch import make_scenarios
 
203
        from bzrlib.tests.branch_implementations import BranchTestProviderAdapter
183
204
        server1 = "a"
184
205
        server2 = "b"
185
206
        formats = [("c", "C"), ("d", "D")]
186
 
        scenarios = make_scenarios(server1, server2, formats)
187
 
        self.assertEqual(2, len(scenarios))
 
207
        adapter = BranchTestProviderAdapter(server1, server2, formats)
 
208
        self.assertEqual(2, len(adapter.scenarios))
188
209
        self.assertEqual([
189
210
            ('str',
190
211
             {'branch_format': 'c',
196
217
              'bzrdir_format': 'D',
197
218
              'transport_readonly_server': 'b',
198
219
              'transport_server': 'a'})],
199
 
            scenarios)
200
 
 
201
 
 
202
 
class TestBzrDirScenarios(tests.TestCase):
203
 
 
204
 
    def test_scenarios(self):
 
220
            adapter.scenarios)
 
221
 
 
222
 
 
223
class TestBzrDirProviderAdapter(TestCase):
 
224
    """A group of tests that test the bzr dir implementation test adapter."""
 
225
 
 
226
    def test_adapted_tests(self):
205
227
        # check that constructor parameters are passed through to the adapted
206
228
        # test.
207
 
        from bzrlib.tests.per_controldir import make_scenarios
 
229
        from bzrlib.tests.bzrdir_implementations import BzrDirTestProviderAdapter
208
230
        vfs_factory = "v"
209
231
        server1 = "a"
210
232
        server2 = "b"
211
233
        formats = ["c", "d"]
212
 
        scenarios = make_scenarios(vfs_factory, server1, server2, formats)
 
234
        adapter = BzrDirTestProviderAdapter(vfs_factory,
 
235
            server1, server2, formats)
213
236
        self.assertEqual([
214
237
            ('str',
215
238
             {'bzrdir_format': 'c',
221
244
              'transport_readonly_server': 'b',
222
245
              'transport_server': 'a',
223
246
              'vfs_transport_factory': 'v'})],
224
 
            scenarios)
225
 
 
226
 
 
227
 
class TestRepositoryScenarios(tests.TestCase):
 
247
            adapter.scenarios)
 
248
 
 
249
 
 
250
class TestRepositoryParameterisation(TestCase):
 
251
    """A group of tests that test the repository implementation test adapter."""
228
252
 
229
253
    def test_formats_to_scenarios(self):
230
 
        from bzrlib.tests.per_repository import formats_to_scenarios
 
254
        """The adapter can generate all the scenarios needed."""
 
255
        from bzrlib.tests.repository_implementations import formats_to_scenarios
231
256
        formats = [("(c)", remote.RemoteRepositoryFormat()),
232
257
                   ("(d)", repository.format_registry.get(
233
 
                    'Bazaar repository format 2a (needs bzr 1.16 or later)\n'))]
 
258
                        'Bazaar pack repository format 1 (needs bzr 0.92)\n'))]
234
259
        no_vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
235
260
            None)
236
261
        vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
237
262
            vfs_transport_factory="vfs")
238
263
        # no_vfs generate scenarios without vfs_transport_factory
239
 
        expected = [
 
264
        self.assertEqual([
240
265
            ('RemoteRepositoryFormat(c)',
241
266
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
242
267
              'repository_format': remote.RemoteRepositoryFormat(),
243
268
              'transport_readonly_server': 'readonly',
244
269
              'transport_server': 'server'}),
245
 
            ('RepositoryFormat2a(d)',
 
270
            ('RepositoryFormatKnitPack1(d)',
246
271
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
247
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
272
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
248
273
              'transport_readonly_server': 'readonly',
249
 
              'transport_server': 'server'})]
250
 
        self.assertEqual(expected, no_vfs_scenarios)
 
274
              'transport_server': 'server'})],
 
275
            no_vfs_scenarios)
251
276
        self.assertEqual([
252
277
            ('RemoteRepositoryFormat(c)',
253
278
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
255
280
              'transport_readonly_server': 'readonly',
256
281
              'transport_server': 'server',
257
282
              'vfs_transport_factory': 'vfs'}),
258
 
            ('RepositoryFormat2a(d)',
 
283
            ('RepositoryFormatKnitPack1(d)',
259
284
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
260
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
285
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
261
286
              'transport_readonly_server': 'readonly',
262
287
              'transport_server': 'server',
263
288
              'vfs_transport_factory': 'vfs'})],
264
289
            vfs_scenarios)
265
290
 
266
291
 
267
 
class TestTestScenarioApplication(tests.TestCase):
 
292
class TestTestScenarioApplier(TestCase):
268
293
    """Tests for the test adaption facilities."""
269
294
 
270
 
    def test_apply_scenario(self):
271
 
        from bzrlib.tests import apply_scenario
272
 
        input_test = TestTestScenarioApplication("test_apply_scenario")
 
295
    def test_adapt_applies_scenarios(self):
 
296
        from bzrlib.tests.repository_implementations import TestScenarioApplier
 
297
        input_test = TestTestScenarioApplier("test_adapt_test_to_scenario")
 
298
        adapter = TestScenarioApplier()
 
299
        adapter.scenarios = [("1", "dict"), ("2", "settings")]
 
300
        calls = []
 
301
        def capture_call(test, scenario):
 
302
            calls.append((test, scenario))
 
303
            return test
 
304
        adapter.adapt_test_to_scenario = capture_call
 
305
        adapter.adapt(input_test)
 
306
        self.assertEqual([(input_test, ("1", "dict")),
 
307
            (input_test, ("2", "settings"))], calls)
 
308
 
 
309
    def test_adapt_test_to_scenario(self):
 
310
        from bzrlib.tests.repository_implementations import TestScenarioApplier
 
311
        input_test = TestTestScenarioApplier("test_adapt_test_to_scenario")
 
312
        adapter = TestScenarioApplier()
273
313
        # setup two adapted tests
274
 
        adapted_test1 = apply_scenario(input_test,
 
314
        adapted_test1 = adapter.adapt_test_to_scenario(input_test,
275
315
            ("new id",
276
316
            {"bzrdir_format":"bzr_format",
277
317
             "repository_format":"repo_fmt",
278
318
             "transport_server":"transport_server",
279
319
             "transport_readonly_server":"readonly-server"}))
280
 
        adapted_test2 = apply_scenario(input_test,
 
320
        adapted_test2 = adapter.adapt_test_to_scenario(input_test,
281
321
            ("new id 2", {"bzrdir_format":None}))
282
322
        # input_test should have been altered.
283
323
        self.assertRaises(AttributeError, getattr, input_test, "bzrdir_format")
284
 
        # the new tests are mutually incompatible, ensuring it has
 
324
        # the new tests are mutually incompatible, ensuring it has 
285
325
        # made new ones, and unspecified elements in the scenario
286
326
        # should not have been altered.
287
327
        self.assertEqual("bzr_format", adapted_test1.bzrdir_format)
290
330
        self.assertEqual("readonly-server",
291
331
            adapted_test1.transport_readonly_server)
292
332
        self.assertEqual(
293
 
            "bzrlib.tests.test_selftest.TestTestScenarioApplication."
294
 
            "test_apply_scenario(new id)",
 
333
            "bzrlib.tests.test_selftest.TestTestScenarioApplier."
 
334
            "test_adapt_test_to_scenario(new id)",
295
335
            adapted_test1.id())
296
336
        self.assertEqual(None, adapted_test2.bzrdir_format)
297
337
        self.assertEqual(
298
 
            "bzrlib.tests.test_selftest.TestTestScenarioApplication."
299
 
            "test_apply_scenario(new id 2)",
 
338
            "bzrlib.tests.test_selftest.TestTestScenarioApplier."
 
339
            "test_adapt_test_to_scenario(new id 2)",
300
340
            adapted_test2.id())
301
341
 
302
342
 
303
 
class TestInterRepositoryScenarios(tests.TestCase):
 
343
class TestInterRepositoryProviderAdapter(TestCase):
 
344
    """A group of tests that test the InterRepository test adapter."""
304
345
 
305
 
    def test_scenarios(self):
 
346
    def test_adapted_tests(self):
306
347
        # check that constructor parameters are passed through to the adapted
307
348
        # test.
308
 
        from bzrlib.tests.per_interrepository import make_scenarios
 
349
        from bzrlib.tests.interrepository_implementations import \
 
350
            InterRepositoryTestProviderAdapter
309
351
        server1 = "a"
310
352
        server2 = "b"
311
 
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
312
 
        scenarios = make_scenarios(server1, server2, formats)
 
353
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
 
354
        adapter = InterRepositoryTestProviderAdapter(server1, server2, formats)
313
355
        self.assertEqual([
314
 
            ('C0,str,str',
315
 
             {'repository_format': 'C1',
 
356
            ('str,str,str',
 
357
             {'interrepo_class': str,
 
358
              'repository_format': 'C1',
316
359
              'repository_format_to': 'C2',
317
360
              'transport_readonly_server': 'b',
318
 
              'transport_server': 'a',
319
 
              'extra_setup': 'C3'}),
320
 
            ('D0,str,str',
321
 
             {'repository_format': 'D1',
 
361
              'transport_server': 'a'}),
 
362
            ('int,str,str',
 
363
             {'interrepo_class': int,
 
364
              'repository_format': 'D1',
322
365
              'repository_format_to': 'D2',
323
366
              'transport_readonly_server': 'b',
324
 
              'transport_server': 'a',
325
 
              'extra_setup': 'D3'})],
326
 
            scenarios)
327
 
 
328
 
 
329
 
class TestWorkingTreeScenarios(tests.TestCase):
 
367
              'transport_server': 'a'})],
 
368
            adapter.formats_to_scenarios(formats))
 
369
 
 
370
 
 
371
class TestWorkingTreeProviderAdapter(TestCase):
 
372
    """A group of tests that test the workingtree implementation test adapter."""
330
373
 
331
374
    def test_scenarios(self):
332
375
        # check that constructor parameters are passed through to the adapted
333
376
        # test.
334
 
        from bzrlib.tests.per_workingtree import make_scenarios
 
377
        from bzrlib.tests.workingtree_implementations \
 
378
            import WorkingTreeTestProviderAdapter
335
379
        server1 = "a"
336
380
        server2 = "b"
337
 
        formats = [workingtree_4.WorkingTreeFormat4(),
338
 
                   workingtree_3.WorkingTreeFormat3(),
339
 
                   workingtree_4.WorkingTreeFormat6()]
340
 
        scenarios = make_scenarios(server1, server2, formats,
341
 
            remote_server='c', remote_readonly_server='d',
342
 
            remote_backing_server='e')
 
381
        formats = [workingtree.WorkingTreeFormat2(),
 
382
                   workingtree.WorkingTreeFormat3(),]
 
383
        adapter = WorkingTreeTestProviderAdapter(server1, server2, formats)
343
384
        self.assertEqual([
344
 
            ('WorkingTreeFormat4',
 
385
            ('WorkingTreeFormat2',
345
386
             {'bzrdir_format': formats[0]._matchingbzrdir,
346
387
              'transport_readonly_server': 'b',
347
388
              'transport_server': 'a',
350
391
             {'bzrdir_format': formats[1]._matchingbzrdir,
351
392
              'transport_readonly_server': 'b',
352
393
              'transport_server': 'a',
353
 
              'workingtree_format': formats[1]}),
354
 
            ('WorkingTreeFormat6',
355
 
             {'bzrdir_format': formats[2]._matchingbzrdir,
356
 
              'transport_readonly_server': 'b',
357
 
              'transport_server': 'a',
358
 
              'workingtree_format': formats[2]}),
359
 
            ('WorkingTreeFormat6,remote',
360
 
             {'bzrdir_format': formats[2]._matchingbzrdir,
361
 
              'repo_is_remote': True,
362
 
              'transport_readonly_server': 'd',
363
 
              'transport_server': 'c',
364
 
              'vfs_transport_factory': 'e',
365
 
              'workingtree_format': formats[2]}),
366
 
            ], scenarios)
367
 
 
368
 
 
369
 
class TestTreeScenarios(tests.TestCase):
370
 
 
371
 
    def test_scenarios(self):
372
 
        # the tree implementation scenario generator is meant to setup one
373
 
        # instance for each working tree format, one additional instance
374
 
        # that will use the default wt format, but create a revision tree for
375
 
        # the tests, and one more that uses the default wt format as a
376
 
        # lightweight checkout of a remote repository.  This means that the wt
377
 
        # ones should have the workingtree_to_test_tree attribute set to
378
 
        # 'return_parameter' and the revision one set to
 
394
              'workingtree_format': formats[1]})],
 
395
            adapter.scenarios)
 
396
 
 
397
 
 
398
class TestTreeProviderAdapter(TestCase):
 
399
    """Test the setup of tree_implementation tests."""
 
400
 
 
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
379
407
        # revision_tree_from_workingtree.
380
408
 
381
 
        from bzrlib.tests.per_tree import (
382
 
            _dirstate_tree_from_workingtree,
383
 
            make_scenarios,
384
 
            preview_tree_pre,
385
 
            preview_tree_post,
 
409
        from bzrlib.tests.tree_implementations import (
 
410
            TreeTestProviderAdapter,
386
411
            return_parameter,
387
412
            revision_tree_from_workingtree
388
413
            )
 
414
        input_test = TestTreeProviderAdapter(
 
415
            "test_adapted_tests")
389
416
        server1 = "a"
390
417
        server2 = "b"
391
 
        smart_server = test_server.SmartTCPServer_for_testing
392
 
        smart_readonly_server = test_server.ReadonlySmartTCPServer_for_testing
393
 
        mem_server = memory.MemoryServer
394
 
        formats = [workingtree_4.WorkingTreeFormat4(),
395
 
                   workingtree_3.WorkingTreeFormat3(),]
396
 
        scenarios = make_scenarios(server1, server2, formats)
397
 
        self.assertEqual(8, len(scenarios))
398
 
        default_wt_format = workingtree.format_registry.get_default()
399
 
        wt4_format = workingtree_4.WorkingTreeFormat4()
400
 
        wt5_format = workingtree_4.WorkingTreeFormat5()
401
 
        wt6_format = workingtree_4.WorkingTreeFormat6()
402
 
        expected_scenarios = [
403
 
            ('WorkingTreeFormat4',
404
 
             {'bzrdir_format': formats[0]._matchingbzrdir,
405
 
              'transport_readonly_server': 'b',
406
 
              'transport_server': 'a',
407
 
              'workingtree_format': formats[0],
408
 
              '_workingtree_to_test_tree': return_parameter,
409
 
              }),
410
 
            ('WorkingTreeFormat3',
411
 
             {'bzrdir_format': formats[1]._matchingbzrdir,
412
 
              'transport_readonly_server': 'b',
413
 
              'transport_server': 'a',
414
 
              'workingtree_format': formats[1],
415
 
              '_workingtree_to_test_tree': return_parameter,
416
 
             }),
417
 
            ('WorkingTreeFormat6,remote',
418
 
             {'bzrdir_format': wt6_format._matchingbzrdir,
419
 
              'repo_is_remote': True,
420
 
              'transport_readonly_server': smart_readonly_server,
421
 
              'transport_server': smart_server,
422
 
              'vfs_transport_factory': mem_server,
423
 
              'workingtree_format': wt6_format,
424
 
              '_workingtree_to_test_tree': return_parameter,
425
 
             }),
426
 
            ('RevisionTree',
427
 
             {'_workingtree_to_test_tree': revision_tree_from_workingtree,
428
 
              'bzrdir_format': default_wt_format._matchingbzrdir,
429
 
              'transport_readonly_server': 'b',
430
 
              'transport_server': 'a',
431
 
              'workingtree_format': default_wt_format,
432
 
             }),
433
 
            ('DirStateRevisionTree,WT4',
434
 
             {'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
435
 
              'bzrdir_format': wt4_format._matchingbzrdir,
436
 
              'transport_readonly_server': 'b',
437
 
              'transport_server': 'a',
438
 
              'workingtree_format': wt4_format,
439
 
             }),
440
 
            ('DirStateRevisionTree,WT5',
441
 
             {'_workingtree_to_test_tree': _dirstate_tree_from_workingtree,
442
 
              'bzrdir_format': wt5_format._matchingbzrdir,
443
 
              'transport_readonly_server': 'b',
444
 
              'transport_server': 'a',
445
 
              'workingtree_format': wt5_format,
446
 
             }),
447
 
            ('PreviewTree',
448
 
             {'_workingtree_to_test_tree': preview_tree_pre,
449
 
              'bzrdir_format': default_wt_format._matchingbzrdir,
450
 
              'transport_readonly_server': 'b',
451
 
              'transport_server': 'a',
452
 
              'workingtree_format': default_wt_format}),
453
 
            ('PreviewTreePost',
454
 
             {'_workingtree_to_test_tree': preview_tree_post,
455
 
              'bzrdir_format': default_wt_format._matchingbzrdir,
456
 
              'transport_readonly_server': 'b',
457
 
              'transport_server': 'a',
458
 
              'workingtree_format': default_wt_format}),
459
 
             ]
460
 
        self.assertEqual(expected_scenarios, scenarios)
461
 
 
462
 
 
463
 
class TestInterTreeScenarios(tests.TestCase):
 
418
        formats = [workingtree.WorkingTreeFormat2(),
 
419
                   workingtree.WorkingTreeFormat3(),]
 
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
 
424
        # abentley 20080412
 
425
        self.assertEqual(6, 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)
 
446
 
 
447
 
 
448
class TestInterTreeProviderAdapter(TestCase):
464
449
    """A group of tests that test the InterTreeTestAdapter."""
465
450
 
466
 
    def test_scenarios(self):
 
451
    def test_adapted_tests(self):
467
452
        # check that constructor parameters are passed through to the adapted
468
453
        # test.
469
454
        # for InterTree tests we want the machinery to bring up two trees in
473
458
        # unlike the TestProviderAdapter we dont want to automatically add a
474
459
        # parameterized one for WorkingTree - the optimisers will tell us what
475
460
        # ones to add.
476
 
        from bzrlib.tests.per_tree import (
 
461
        from bzrlib.tests.tree_implementations import (
477
462
            return_parameter,
478
 
            )
479
 
        from bzrlib.tests.per_intertree import (
480
 
            make_scenarios,
481
 
            )
482
 
        from bzrlib.workingtree_3 import WorkingTreeFormat3
483
 
        from bzrlib.workingtree_4 import WorkingTreeFormat4
484
 
        input_test = TestInterTreeScenarios(
485
 
            "test_scenarios")
 
463
            revision_tree_from_workingtree
 
464
            )
 
465
        from bzrlib.tests.intertree_implementations import (
 
466
            InterTreeTestProviderAdapter,
 
467
            )
 
468
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
 
469
        input_test = TestInterTreeProviderAdapter(
 
470
            "test_adapted_tests")
486
471
        server1 = "a"
487
472
        server2 = "b"
488
 
        format1 = WorkingTreeFormat4()
 
473
        format1 = WorkingTreeFormat2()
489
474
        format2 = WorkingTreeFormat3()
490
 
        formats = [("1", str, format1, format2, "converter1"),
491
 
            ("2", int, format2, format1, "converter2")]
492
 
        scenarios = make_scenarios(server1, server2, formats)
493
 
        self.assertEqual(2, len(scenarios))
494
 
        expected_scenarios = [
495
 
            ("1", {
496
 
                "bzrdir_format": format1._matchingbzrdir,
497
 
                "intertree_class": formats[0][1],
498
 
                "workingtree_format": formats[0][2],
499
 
                "workingtree_format_to": formats[0][3],
500
 
                "mutable_trees_to_test_trees": formats[0][4],
501
 
                "_workingtree_to_test_tree": return_parameter,
502
 
                "transport_server": server1,
503
 
                "transport_readonly_server": server2,
504
 
                }),
505
 
            ("2", {
506
 
                "bzrdir_format": format2._matchingbzrdir,
507
 
                "intertree_class": formats[1][1],
508
 
                "workingtree_format": formats[1][2],
509
 
                "workingtree_format_to": formats[1][3],
510
 
                "mutable_trees_to_test_trees": formats[1][4],
511
 
                "_workingtree_to_test_tree": return_parameter,
512
 
                "transport_server": server1,
513
 
                "transport_readonly_server": server2,
514
 
                }),
515
 
            ]
516
 
        self.assertEqual(scenarios, expected_scenarios)
517
 
 
518
 
 
519
 
class TestTestCaseInTempDir(tests.TestCaseInTempDir):
 
475
        formats = [(str, format1, format2, "converter1"),
 
476
            (int, format2, format1, "converter2")]
 
477
        adapter = InterTreeTestProviderAdapter(server1, server2, formats)
 
478
        suite = adapter.adapt(input_test)
 
479
        tests = list(iter(suite))
 
480
        self.assertEqual(2, len(tests))
 
481
        self.assertEqual(tests[0].intertree_class, formats[0][0])
 
482
        self.assertEqual(tests[0].workingtree_format, formats[0][1])
 
483
        self.assertEqual(tests[0].workingtree_format_to, formats[0][2])
 
484
        self.assertEqual(tests[0].mutable_trees_to_test_trees, formats[0][3])
 
485
        self.assertEqual(tests[0]._workingtree_to_test_tree, return_parameter)
 
486
        self.assertEqual(tests[0].transport_server, server1)
 
487
        self.assertEqual(tests[0].transport_readonly_server, server2)
 
488
        self.assertEqual(tests[1].intertree_class, formats[1][0])
 
489
        self.assertEqual(tests[1].workingtree_format, formats[1][1])
 
490
        self.assertEqual(tests[1].workingtree_format_to, formats[1][2])
 
491
        self.assertEqual(tests[1].mutable_trees_to_test_trees, formats[1][3])
 
492
        self.assertEqual(tests[1]._workingtree_to_test_tree, return_parameter)
 
493
        self.assertEqual(tests[1].transport_server, server1)
 
494
        self.assertEqual(tests[1].transport_readonly_server, server2)
 
495
 
 
496
 
 
497
class TestTestCaseInTempDir(TestCaseInTempDir):
520
498
 
521
499
    def test_home_is_not_working(self):
522
500
        self.assertNotEqual(self.test_dir, self.test_home_dir)
524
502
        self.assertIsSameRealPath(self.test_dir, cwd)
525
503
        self.assertIsSameRealPath(self.test_home_dir, os.environ['HOME'])
526
504
 
527
 
    def test_assertEqualStat_equal(self):
528
 
        from bzrlib.tests.test_dirstate import _FakeStat
529
 
        self.build_tree(["foo"])
530
 
        real = os.lstat("foo")
531
 
        fake = _FakeStat(real.st_size, real.st_mtime, real.st_ctime,
532
 
            real.st_dev, real.st_ino, real.st_mode)
533
 
        self.assertEqualStat(real, fake)
534
 
 
535
 
    def test_assertEqualStat_notequal(self):
536
 
        self.build_tree(["foo", "longname"])
537
 
        self.assertRaises(AssertionError, self.assertEqualStat,
538
 
            os.lstat("foo"), os.lstat("longname"))
539
 
 
540
 
    def test_failUnlessExists(self):
541
 
        """Deprecated failUnlessExists and failIfExists"""
542
 
        self.applyDeprecated(
543
 
            deprecated_in((2, 4)),
544
 
            self.failUnlessExists, '.')
545
 
        self.build_tree(['foo/', 'foo/bar'])
546
 
        self.applyDeprecated(
547
 
            deprecated_in((2, 4)),
548
 
            self.failUnlessExists, 'foo/bar')
549
 
        self.applyDeprecated(
550
 
            deprecated_in((2, 4)),
551
 
            self.failIfExists, 'foo/foo')
552
 
 
553
 
    def test_assertPathExists(self):
554
 
        self.assertPathExists('.')
555
 
        self.build_tree(['foo/', 'foo/bar'])
556
 
        self.assertPathExists('foo/bar')
557
 
        self.assertPathDoesNotExist('foo/foo')
558
 
 
559
 
 
560
 
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
 
505
 
 
506
class TestTestCaseWithMemoryTransport(TestCaseWithMemoryTransport):
561
507
 
562
508
    def test_home_is_non_existant_dir_under_root(self):
563
509
        """The test_home_dir for TestCaseWithMemoryTransport is missing.
564
510
 
565
511
        This is because TestCaseWithMemoryTransport is for tests that do not
566
 
        need any disk resources: they should be hooked into bzrlib in such a
567
 
        way that no global settings are being changed by the test (only a
 
512
        need any disk resources: they should be hooked into bzrlib in such a 
 
513
        way that no global settings are being changed by the test (only a 
568
514
        few tests should need to do that), and having a missing dir as home is
569
515
        an effective way to ensure that this is the case.
570
516
        """
572
518
            self.TEST_ROOT + "/MemoryTransportMissingHomeDir",
573
519
            self.test_home_dir)
574
520
        self.assertIsSameRealPath(self.test_home_dir, os.environ['HOME'])
575
 
 
 
521
        
576
522
    def test_cwd_is_TEST_ROOT(self):
577
523
        self.assertIsSameRealPath(self.test_dir, self.TEST_ROOT)
578
524
        cwd = osutils.getcwd()
579
525
        self.assertIsSameRealPath(self.test_dir, cwd)
580
526
 
581
 
    def test_BZR_HOME_and_HOME_are_bytestrings(self):
582
 
        """The $BZR_HOME and $HOME environment variables should not be unicode.
583
 
 
584
 
        See https://bugs.launchpad.net/bzr/+bug/464174
585
 
        """
586
 
        self.assertIsInstance(os.environ['BZR_HOME'], str)
587
 
        self.assertIsInstance(os.environ['HOME'], str)
588
 
 
589
527
    def test_make_branch_and_memory_tree(self):
590
528
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
591
529
 
595
533
        tree = self.make_branch_and_memory_tree('dir')
596
534
        # Guard against regression into MemoryTransport leaking
597
535
        # files to disk instead of keeping them in memory.
598
 
        self.assertFalse(osutils.lexists('dir'))
 
536
        self.failIf(osutils.lexists('dir'))
599
537
        self.assertIsInstance(tree, memorytree.MemoryTree)
600
538
 
601
539
    def test_make_branch_and_memory_tree_with_format(self):
602
540
        """make_branch_and_memory_tree should accept a format option."""
603
541
        format = bzrdir.BzrDirMetaFormat1()
604
 
        format.repository_format = repository.format_registry.get_default()
 
542
        format.repository_format = weaverepo.RepositoryFormat7()
605
543
        tree = self.make_branch_and_memory_tree('dir', format=format)
606
544
        # Guard against regression into MemoryTransport leaking
607
545
        # files to disk instead of keeping them in memory.
608
 
        self.assertFalse(osutils.lexists('dir'))
 
546
        self.failIf(osutils.lexists('dir'))
609
547
        self.assertIsInstance(tree, memorytree.MemoryTree)
610
548
        self.assertEqual(format.repository_format.__class__,
611
549
            tree.branch.repository._format.__class__)
615
553
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
616
554
        # Guard against regression into MemoryTransport leaking
617
555
        # files to disk instead of keeping them in memory.
618
 
        self.assertFalse(osutils.lexists('dir'))
 
556
        self.failIf(osutils.lexists('dir'))
619
557
 
620
558
    def test_make_branch_builder_with_format(self):
621
559
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
622
560
        # that the format objects are used.
623
561
        format = bzrdir.BzrDirMetaFormat1()
624
 
        repo_format = repository.format_registry.get_default()
 
562
        repo_format = weaverepo.RepositoryFormat7()
625
563
        format.repository_format = repo_format
626
564
        builder = self.make_branch_builder('dir', format=format)
627
565
        the_branch = builder.get_branch()
628
566
        # Guard against regression into MemoryTransport leaking
629
567
        # files to disk instead of keeping them in memory.
630
 
        self.assertFalse(osutils.lexists('dir'))
 
568
        self.failIf(osutils.lexists('dir'))
631
569
        self.assertEqual(format.repository_format.__class__,
632
570
                         the_branch.repository._format.__class__)
633
571
        self.assertEqual(repo_format.get_format_string(),
639
577
        the_branch = builder.get_branch()
640
578
        # Guard against regression into MemoryTransport leaking
641
579
        # files to disk instead of keeping them in memory.
642
 
        self.assertFalse(osutils.lexists('dir'))
643
 
        dir_format = controldir.format_registry.make_bzrdir('knit')
 
580
        self.failIf(osutils.lexists('dir'))
 
581
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
644
582
        self.assertEqual(dir_format.repository_format.__class__,
645
583
                         the_branch.repository._format.__class__)
646
584
        self.assertEqual('Bazaar-NG Knit Repository Format 1',
647
585
                         self.get_transport().get_bytes(
648
586
                            'dir/.bzr/repository/format'))
649
587
 
650
 
    def test_dangling_locks_cause_failures(self):
651
 
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
652
 
            def test_function(self):
653
 
                t = self.get_transport_from_path('.')
654
 
                l = lockdir.LockDir(t, 'lock')
655
 
                l.create()
656
 
                l.attempt_lock()
657
 
        test = TestDanglingLock('test_function')
658
 
        result = test.run()
659
 
        total_failures = result.errors + result.failures
660
 
        if self._lock_check_thorough:
661
 
            self.assertEqual(1, len(total_failures))
662
 
        else:
663
 
            # When _lock_check_thorough is disabled, then we don't trigger a
664
 
            # failure
665
 
            self.assertEqual(0, len(total_failures))
666
 
 
667
 
 
668
 
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
 
588
    def test_safety_net(self):
 
589
        """No test should modify the safety .bzr directory.
 
590
 
 
591
        We just test that the _check_safety_net private method raises
 
592
        AssertionError, it's easier than building a test suite with the same
 
593
        test.
 
594
        """
 
595
        # Oops, a commit in the current directory (i.e. without local .bzr
 
596
        # directory) will crawl up the hierarchy to find a .bzr directory.
 
597
        self.run_bzr(['commit', '-mfoo', '--unchanged'])
 
598
        # But we have a safety net in place.
 
599
        self.assertRaises(AssertionError, self._check_safety_net)
 
600
 
 
601
 
 
602
class TestTestCaseWithTransport(TestCaseWithTransport):
669
603
    """Tests for the convenience functions TestCaseWithTransport introduces."""
670
604
 
671
605
    def test_get_readonly_url_none(self):
 
606
        from bzrlib.transport import get_transport
 
607
        from bzrlib.transport.memory import MemoryServer
672
608
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
673
 
        self.vfs_transport_factory = memory.MemoryServer
 
609
        self.vfs_transport_factory = MemoryServer
674
610
        self.transport_readonly_server = None
675
611
        # calling get_readonly_transport() constructs a decorator on the url
676
612
        # for the server
677
613
        url = self.get_readonly_url()
678
614
        url2 = self.get_readonly_url('foo/bar')
679
 
        t = transport.get_transport_from_url(url)
680
 
        t2 = transport.get_transport_from_url(url2)
681
 
        self.assertIsInstance(t, ReadonlyTransportDecorator)
682
 
        self.assertIsInstance(t2, ReadonlyTransportDecorator)
 
615
        t = get_transport(url)
 
616
        t2 = get_transport(url2)
 
617
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
 
618
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
683
619
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
684
620
 
685
621
    def test_get_readonly_url_http(self):
686
622
        from bzrlib.tests.http_server import HttpServer
 
623
        from bzrlib.transport import get_transport
 
624
        from bzrlib.transport.local import LocalURLServer
687
625
        from bzrlib.transport.http import HttpTransportBase
688
 
        self.transport_server = test_server.LocalURLServer
 
626
        self.transport_server = LocalURLServer
689
627
        self.transport_readonly_server = HttpServer
690
628
        # calling get_readonly_transport() gives us a HTTP server instance.
691
629
        url = self.get_readonly_url()
692
630
        url2 = self.get_readonly_url('foo/bar')
693
631
        # the transport returned may be any HttpTransportBase subclass
694
 
        t = transport.get_transport_from_url(url)
695
 
        t2 = transport.get_transport_from_url(url2)
696
 
        self.assertIsInstance(t, HttpTransportBase)
697
 
        self.assertIsInstance(t2, HttpTransportBase)
 
632
        t = get_transport(url)
 
633
        t2 = get_transport(url2)
 
634
        self.failUnless(isinstance(t, HttpTransportBase))
 
635
        self.failUnless(isinstance(t2, HttpTransportBase))
698
636
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
699
637
 
700
638
    def test_is_directory(self):
708
646
    def test_make_branch_builder(self):
709
647
        builder = self.make_branch_builder('dir')
710
648
        rev_id = builder.build_commit()
711
 
        self.assertPathExists('dir')
712
 
        a_dir = controldir.ControlDir.open('dir')
 
649
        self.failUnlessExists('dir')
 
650
        a_dir = bzrdir.BzrDir.open('dir')
713
651
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
714
652
        a_branch = a_dir.open_branch()
715
653
        builder_branch = builder.get_branch()
718
656
        self.assertEqual((1, rev_id), a_branch.last_revision_info())
719
657
 
720
658
 
721
 
class TestTestCaseTransports(tests.TestCaseWithTransport):
 
659
class TestTestCaseTransports(TestCaseWithTransport):
722
660
 
723
661
    def setUp(self):
724
662
        super(TestTestCaseTransports, self).setUp()
725
 
        self.vfs_transport_factory = memory.MemoryServer
 
663
        self.vfs_transport_factory = MemoryServer
726
664
 
727
665
    def test_make_bzrdir_preserves_transport(self):
728
666
        t = self.get_transport()
729
667
        result_bzrdir = self.make_bzrdir('subdir')
730
 
        self.assertIsInstance(result_bzrdir.transport,
731
 
                              memory.MemoryTransport)
 
668
        self.assertIsInstance(result_bzrdir.transport, 
 
669
                              MemoryTransport)
732
670
        # should not be on disk, should only be in memory
733
 
        self.assertPathDoesNotExist('subdir')
734
 
 
735
 
 
736
 
class TestChrootedTest(tests.ChrootedTestCase):
 
671
        self.failIfExists('subdir')
 
672
 
 
673
 
 
674
class TestChrootedTest(ChrootedTestCase):
737
675
 
738
676
    def test_root_is_root(self):
739
 
        t = transport.get_transport_from_url(self.get_readonly_url())
 
677
        from bzrlib.transport import get_transport
 
678
        t = get_transport(self.get_readonly_url())
740
679
        url = t.base
741
680
        self.assertEqual(url, t.clone('..').base)
742
681
 
743
682
 
744
 
class TestProfileResult(tests.TestCase):
745
 
 
746
 
    def test_profiles_tests(self):
747
 
        self.requireFeature(features.lsprof_feature)
748
 
        terminal = testtools.testresult.doubles.ExtendedTestResult()
749
 
        result = tests.ProfileResult(terminal)
750
 
        class Sample(tests.TestCase):
751
 
            def a(self):
752
 
                self.sample_function()
753
 
            def sample_function(self):
754
 
                pass
755
 
        test = Sample("a")
756
 
        test.run(result)
757
 
        case = terminal._events[0][1]
758
 
        self.assertLength(1, case._benchcalls)
759
 
        # We must be able to unpack it as the test reporting code wants
760
 
        (_, _, _), stats = case._benchcalls[0]
761
 
        self.assertTrue(callable(stats.pprint))
762
 
 
763
 
 
764
 
class TestTestResult(tests.TestCase):
 
683
class MockProgress(_BaseProgressBar):
 
684
    """Progress-bar standin that records calls.
 
685
 
 
686
    Useful for testing pb using code.
 
687
    """
 
688
 
 
689
    def __init__(self):
 
690
        _BaseProgressBar.__init__(self)
 
691
        self.calls = []
 
692
 
 
693
    def tick(self):
 
694
        self.calls.append(('tick',))
 
695
 
 
696
    def update(self, msg=None, current=None, total=None):
 
697
        self.calls.append(('update', msg, current, total))
 
698
 
 
699
    def clear(self):
 
700
        self.calls.append(('clear',))
 
701
 
 
702
    def note(self, msg, *args):
 
703
        self.calls.append(('note', msg, args))
 
704
 
 
705
 
 
706
class TestTestResult(TestCase):
765
707
 
766
708
    def check_timing(self, test_case, expected_re):
767
709
        result = bzrlib.tests.TextTestResult(self._log_file,
768
710
                descriptions=0,
769
711
                verbosity=1,
770
712
                )
771
 
        capture = testtools.testresult.doubles.ExtendedTestResult()
772
 
        test_case.run(MultiTestResult(result, capture))
773
 
        run_case = capture._events[0][1]
774
 
        timed_string = result._testTimeString(run_case)
 
713
        test_case.run(result)
 
714
        timed_string = result._testTimeString(test_case)
775
715
        self.assertContainsRe(timed_string, expected_re)
776
716
 
777
717
    def test_test_reporting(self):
778
 
        class ShortDelayTestCase(tests.TestCase):
 
718
        class ShortDelayTestCase(TestCase):
779
719
            def test_short_delay(self):
780
720
                time.sleep(0.003)
781
721
            def test_short_benchmark(self):
782
722
                self.time(time.sleep, 0.003)
783
723
        self.check_timing(ShortDelayTestCase('test_short_delay'),
784
724
                          r"^ +[0-9]+ms$")
785
 
        # if a benchmark time is given, we now show just that time followed by
786
 
        # a star
 
725
        # if a benchmark time is given, we want a x of y style result.
787
726
        self.check_timing(ShortDelayTestCase('test_short_benchmark'),
788
 
                          r"^ +[0-9]+ms\*$")
 
727
                          r"^ +[0-9]+ms/ +[0-9]+ms$")
789
728
 
790
729
    def test_unittest_reporting_unittest_class(self):
791
730
        # getting the time from a non-bzrlib test works ok
794
733
                time.sleep(0.003)
795
734
        self.check_timing(ShortDelayTestCase('test_short_delay'),
796
735
                          r"^ +[0-9]+ms$")
797
 
 
 
736
        
 
737
    def test_assigned_benchmark_file_stores_date(self):
 
738
        output = StringIO()
 
739
        result = bzrlib.tests.TextTestResult(self._log_file,
 
740
                                        descriptions=0,
 
741
                                        verbosity=1,
 
742
                                        bench_history=output
 
743
                                        )
 
744
        output_string = output.getvalue()
 
745
        # if you are wondering about the regexp please read the comment in
 
746
        # test_bench_history (bzrlib.tests.test_selftest.TestRunner)
 
747
        # XXX: what comment?  -- Andrew Bennetts
 
748
        self.assertContainsRe(output_string, "--date [0-9.]+")
 
749
 
 
750
    def test_benchhistory_records_test_times(self):
 
751
        result_stream = StringIO()
 
752
        result = bzrlib.tests.TextTestResult(
 
753
            self._log_file,
 
754
            descriptions=0,
 
755
            verbosity=1,
 
756
            bench_history=result_stream
 
757
            )
 
758
 
 
759
        # we want profile a call and check that its test duration is recorded
 
760
        # make a new test instance that when run will generate a benchmark
 
761
        example_test_case = TestTestResult("_time_hello_world_encoding")
 
762
        # execute the test, which should succeed and record times
 
763
        example_test_case.run(result)
 
764
        lines = result_stream.getvalue().splitlines()
 
765
        self.assertEqual(2, len(lines))
 
766
        self.assertContainsRe(lines[1],
 
767
            " *[0-9]+ms bzrlib.tests.test_selftest.TestTestResult"
 
768
            "._time_hello_world_encoding")
 
769
 
798
770
    def _time_hello_world_encoding(self):
799
771
        """Profile two sleep calls
800
 
 
 
772
        
801
773
        This is used to exercise the test framework.
802
774
        """
803
775
        self.time(unicode, 'hello', errors='replace')
805
777
 
806
778
    def test_lsprofiling(self):
807
779
        """Verbose test result prints lsprof statistics from test cases."""
808
 
        self.requireFeature(features.lsprof_feature)
 
780
        self.requireFeature(test_lsprof.LSProfFeature)
809
781
        result_stream = StringIO()
810
782
        result = bzrlib.tests.VerboseTestResult(
811
 
            result_stream,
 
783
            unittest._WritelnDecorator(result_stream),
812
784
            descriptions=0,
813
785
            verbosity=2,
814
786
            )
821
793
        # execute the test, which should succeed and record profiles
822
794
        example_test_case.run(result)
823
795
        # lsprofile_something()
824
 
        # if this worked we want
 
796
        # if this worked we want 
825
797
        # LSProf output for <built in function unicode> (['hello'], {'errors': 'replace'})
826
798
        #    CallCount    Recursive    Total(ms)   Inline(ms) module:lineno(function)
827
799
        # (the lsprof header)
828
800
        # ... an arbitrary number of lines
829
801
        # and the function call which is time.sleep.
830
 
        #           1        0            ???         ???       ???(sleep)
 
802
        #           1        0            ???         ???       ???(sleep) 
831
803
        # and then repeated but with 'world', rather than 'hello'.
832
804
        # this should appear in the output stream of our test result.
833
805
        output = result_stream.getvalue()
840
812
        self.assertContainsRe(output,
841
813
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
842
814
 
843
 
    def test_uses_time_from_testtools(self):
844
 
        """Test case timings in verbose results should use testtools times"""
845
 
        import datetime
846
 
        class TimeAddedVerboseTestResult(tests.VerboseTestResult):
847
 
            def startTest(self, test):
848
 
                self.time(datetime.datetime.utcfromtimestamp(1.145))
849
 
                super(TimeAddedVerboseTestResult, self).startTest(test)
850
 
            def addSuccess(self, test):
851
 
                self.time(datetime.datetime.utcfromtimestamp(51.147))
852
 
                super(TimeAddedVerboseTestResult, self).addSuccess(test)
853
 
            def report_tests_starting(self): pass
854
 
        sio = StringIO()
855
 
        self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
856
 
        self.assertEndsWith(sio.getvalue(), "OK    50002ms\n")
857
 
 
858
815
    def test_known_failure(self):
859
 
        """Using knownFailure should trigger several result actions."""
860
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
861
 
            def stopTestRun(self): pass
862
 
            def report_tests_starting(self): pass
863
 
            def report_known_failure(self, test, err=None, details=None):
864
 
                self._call = test, 'known failure'
 
816
        """A KnownFailure being raised should trigger several result actions."""
 
817
        class InstrumentedTestResult(ExtendedTestResult):
 
818
 
 
819
            def report_test_start(self, test): pass
 
820
            def report_known_failure(self, test, err):
 
821
                self._call = test, err
865
822
        result = InstrumentedTestResult(None, None, None, None)
866
 
        class Test(tests.TestCase):
867
 
            def test_function(self):
868
 
                self.knownFailure('failed!')
869
 
        test = Test("test_function")
 
823
        def test_function():
 
824
            raise KnownFailure('failed!')
 
825
        test = unittest.FunctionTestCase(test_function)
870
826
        test.run(result)
871
827
        # it should invoke 'report_known_failure'.
872
828
        self.assertEqual(2, len(result._call))
873
 
        self.assertEqual(test.id(), result._call[0].id())
874
 
        self.assertEqual('known failure', result._call[1])
 
829
        self.assertEqual(test, result._call[0])
 
830
        self.assertEqual(KnownFailure, result._call[1][0])
 
831
        self.assertIsInstance(result._call[1][1], KnownFailure)
875
832
        # we dont introspec the traceback, if the rest is ok, it would be
876
833
        # exceptional for it not to be.
877
834
        # it should update the known_failure_count on the object.
883
840
        # verbose test output formatting
884
841
        result_stream = StringIO()
885
842
        result = bzrlib.tests.VerboseTestResult(
886
 
            result_stream,
 
843
            unittest._WritelnDecorator(result_stream),
887
844
            descriptions=0,
888
845
            verbosity=2,
889
846
            )
890
 
        _get_test("test_xfail").run(result)
891
 
        self.assertContainsRe(result_stream.getvalue(),
892
 
            "\n\\S+\\.test_xfail\\s+XFAIL\\s+\\d+ms\n"
893
 
            "\\s*(?:Text attachment: )?reason"
894
 
            "(?:\n-+\n|: {{{)"
895
 
            "this_fails"
896
 
            "(?:\n-+\n|}}}\n)")
 
847
        test = self.get_passing_test()
 
848
        result.startTest(test)
 
849
        prefix = len(result_stream.getvalue())
 
850
        # the err parameter has the shape:
 
851
        # (class, exception object, traceback)
 
852
        # KnownFailures dont get their tracebacks shown though, so we
 
853
        # can skip that.
 
854
        err = (KnownFailure, KnownFailure('foo'), None)
 
855
        result.report_known_failure(test, err)
 
856
        output = result_stream.getvalue()[prefix:]
 
857
        lines = output.splitlines()
 
858
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
 
859
        self.assertEqual(lines[1], '    foo')
 
860
        self.assertEqual(2, len(lines))
 
861
 
 
862
    def test_text_report_known_failure(self):
 
863
        # text test output formatting
 
864
        pb = MockProgress()
 
865
        result = bzrlib.tests.TextTestResult(
 
866
            None,
 
867
            descriptions=0,
 
868
            verbosity=1,
 
869
            pb=pb,
 
870
            )
 
871
        test = self.get_passing_test()
 
872
        # this seeds the state to handle reporting the test.
 
873
        result.startTest(test)
 
874
        # the err parameter has the shape:
 
875
        # (class, exception object, traceback)
 
876
        # KnownFailures dont get their tracebacks shown though, so we
 
877
        # can skip that.
 
878
        err = (KnownFailure, KnownFailure('foo'), None)
 
879
        result.report_known_failure(test, err)
 
880
        self.assertEqual(
 
881
            [
 
882
            ('update', '[1 in 0s] passing_test', None, None),
 
883
            ('note', 'XFAIL: %s\n%s\n', ('passing_test', err[1]))
 
884
            ],
 
885
            pb.calls)
 
886
        # known_failures should be printed in the summary, so if we run a test
 
887
        # after there are some known failures, the update prefix should match
 
888
        # this.
 
889
        result.known_failure_count = 3
 
890
        test.run(result)
 
891
        self.assertEqual(
 
892
            [
 
893
            ('update', '[2 in 0s] passing_test', None, None),
 
894
            ],
 
895
            pb.calls[2:])
897
896
 
898
897
    def get_passing_test(self):
899
898
        """Return a test object that can't be run usefully."""
903
902
 
904
903
    def test_add_not_supported(self):
905
904
        """Test the behaviour of invoking addNotSupported."""
906
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
907
 
            def stopTestRun(self): pass
908
 
            def report_tests_starting(self): pass
 
905
        class InstrumentedTestResult(ExtendedTestResult):
 
906
            def report_test_start(self, test): pass
909
907
            def report_unsupported(self, test, feature):
910
908
                self._call = test, feature
911
909
        result = InstrumentedTestResult(None, None, None, None)
912
910
        test = SampleTestCase('_test_pass')
913
 
        feature = features.Feature()
 
911
        feature = Feature()
914
912
        result.startTest(test)
915
913
        result.addNotSupported(test, feature)
916
914
        # it should invoke 'report_unsupported'.
930
928
        # verbose test output formatting
931
929
        result_stream = StringIO()
932
930
        result = bzrlib.tests.VerboseTestResult(
933
 
            result_stream,
 
931
            unittest._WritelnDecorator(result_stream),
934
932
            descriptions=0,
935
933
            verbosity=2,
936
934
            )
937
935
        test = self.get_passing_test()
938
 
        feature = features.Feature()
 
936
        feature = Feature()
939
937
        result.startTest(test)
940
938
        prefix = len(result_stream.getvalue())
941
939
        result.report_unsupported(test, feature)
942
940
        output = result_stream.getvalue()[prefix:]
943
941
        lines = output.splitlines()
944
 
        # We don't check for the final '0ms' since it may fail on slow hosts
945
 
        self.assertStartsWith(lines[0], 'NODEP')
946
 
        self.assertEqual(lines[1],
947
 
                         "    The feature 'Feature' is not available.")
948
 
 
 
942
        self.assertEqual(lines, ['NODEP                   0ms', "    The feature 'Feature' is not available."])
 
943
    
 
944
    def test_text_report_unsupported(self):
 
945
        # text test output formatting
 
946
        pb = MockProgress()
 
947
        result = bzrlib.tests.TextTestResult(
 
948
            None,
 
949
            descriptions=0,
 
950
            verbosity=1,
 
951
            pb=pb,
 
952
            )
 
953
        test = self.get_passing_test()
 
954
        feature = Feature()
 
955
        # this seeds the state to handle reporting the test.
 
956
        result.startTest(test)
 
957
        result.report_unsupported(test, feature)
 
958
        # no output on unsupported features
 
959
        self.assertEqual(
 
960
            [('update', '[1 in 0s] passing_test', None, None)
 
961
            ],
 
962
            pb.calls)
 
963
        # the number of missing features should be printed in the progress
 
964
        # summary, so check for that.
 
965
        result.unsupported = {'foo':0, 'bar':0}
 
966
        test.run(result)
 
967
        self.assertEqual(
 
968
            [
 
969
            ('update', '[2 in 0s, 2 missing] passing_test', None, None),
 
970
            ],
 
971
            pb.calls[1:])
 
972
    
949
973
    def test_unavailable_exception(self):
950
974
        """An UnavailableFeature being raised should invoke addNotSupported."""
951
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
952
 
            def stopTestRun(self): pass
953
 
            def report_tests_starting(self): pass
 
975
        class InstrumentedTestResult(ExtendedTestResult):
 
976
 
 
977
            def report_test_start(self, test): pass
954
978
            def addNotSupported(self, test, feature):
955
979
                self._call = test, feature
956
980
        result = InstrumentedTestResult(None, None, None, None)
957
 
        feature = features.Feature()
958
 
        class Test(tests.TestCase):
959
 
            def test_function(self):
960
 
                raise tests.UnavailableFeature(feature)
961
 
        test = Test("test_function")
 
981
        feature = Feature()
 
982
        def test_function():
 
983
            raise UnavailableFeature(feature)
 
984
        test = unittest.FunctionTestCase(test_function)
962
985
        test.run(result)
963
986
        # it should invoke 'addNotSupported'.
964
987
        self.assertEqual(2, len(result._call))
965
 
        self.assertEqual(test.id(), result._call[0].id())
 
988
        self.assertEqual(test, result._call[0])
966
989
        self.assertEqual(feature, result._call[1])
967
990
        # and not count as an error
968
991
        self.assertEqual(0, result.error_count)
975
998
        result.addNotSupported(test, feature)
976
999
        self.assertFalse(result.wasStrictlySuccessful())
977
1000
        self.assertEqual(None, result._extractBenchmarkTime(test))
978
 
 
 
1001
 
979
1002
    def test_strict_with_known_failure(self):
980
1003
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
981
1004
                                             verbosity=1)
982
 
        test = _get_test("test_xfail")
983
 
        test.run(result)
 
1005
        test = self.get_passing_test()
 
1006
        err = (KnownFailure, KnownFailure('foo'), None)
 
1007
        result._addKnownFailure(test, err)
984
1008
        self.assertFalse(result.wasStrictlySuccessful())
985
1009
        self.assertEqual(None, result._extractBenchmarkTime(test))
986
1010
 
992
1016
        self.assertTrue(result.wasStrictlySuccessful())
993
1017
        self.assertEqual(None, result._extractBenchmarkTime(test))
994
1018
 
995
 
    def test_startTests(self):
996
 
        """Starting the first test should trigger startTests."""
997
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
998
 
            calls = 0
999
 
            def startTests(self): self.calls += 1
1000
 
        result = InstrumentedTestResult(None, None, None, None)
1001
 
        def test_function():
1002
 
            pass
1003
 
        test = unittest.FunctionTestCase(test_function)
1004
 
        test.run(result)
1005
 
        self.assertEqual(1, result.calls)
1006
 
 
1007
 
    def test_startTests_only_once(self):
1008
 
        """With multiple tests startTests should still only be called once"""
1009
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
1010
 
            calls = 0
1011
 
            def startTests(self): self.calls += 1
1012
 
        result = InstrumentedTestResult(None, None, None, None)
1013
 
        suite = unittest.TestSuite([
1014
 
            unittest.FunctionTestCase(lambda: None),
1015
 
            unittest.FunctionTestCase(lambda: None)])
1016
 
        suite.run(result)
1017
 
        self.assertEqual(1, result.calls)
1018
 
        self.assertEqual(2, result.count)
1019
 
 
1020
 
 
1021
 
class TestRunner(tests.TestCase):
 
1019
 
 
1020
class TestUnicodeFilenameFeature(TestCase):
 
1021
 
 
1022
    def test_probe_passes(self):
 
1023
        """UnicodeFilenameFeature._probe passes."""
 
1024
        # We can't test much more than that because the behaviour depends
 
1025
        # on the platform.
 
1026
        tests.UnicodeFilenameFeature._probe()
 
1027
 
 
1028
 
 
1029
class TestRunner(TestCase):
1022
1030
 
1023
1031
    def dummy_test(self):
1024
1032
        pass
1028
1036
 
1029
1037
        This current saves and restores:
1030
1038
        TestCaseInTempDir.TEST_ROOT
1031
 
 
1032
 
        There should be no tests in this file that use
1033
 
        bzrlib.tests.TextTestRunner without using this convenience method,
1034
 
        because of our use of global state.
 
1039
        
 
1040
        There should be no tests in this file that use bzrlib.tests.TextTestRunner
 
1041
        without using this convenience method, because of our use of global state.
1035
1042
        """
1036
 
        old_root = tests.TestCaseInTempDir.TEST_ROOT
 
1043
        old_root = TestCaseInTempDir.TEST_ROOT
1037
1044
        try:
1038
 
            tests.TestCaseInTempDir.TEST_ROOT = None
 
1045
            TestCaseInTempDir.TEST_ROOT = None
1039
1046
            return testrunner.run(test)
1040
1047
        finally:
1041
 
            tests.TestCaseInTempDir.TEST_ROOT = old_root
 
1048
            TestCaseInTempDir.TEST_ROOT = old_root
1042
1049
 
1043
1050
    def test_known_failure_failed_run(self):
1044
1051
        # run a test that generates a known failure which should be printed in
1045
1052
        # the final output when real failures occur.
1046
 
        class Test(tests.TestCase):
1047
 
            def known_failure_test(self):
1048
 
                self.expectFailure('failed', self.assertTrue, False)
 
1053
        def known_failure_test():
 
1054
            raise KnownFailure('failed')
1049
1055
        test = unittest.TestSuite()
1050
 
        test.addTest(Test("known_failure_test"))
 
1056
        test.addTest(unittest.FunctionTestCase(known_failure_test))
1051
1057
        def failing_test():
1052
1058
            raise AssertionError('foo')
1053
1059
        test.addTest(unittest.FunctionTestCase(failing_test))
1054
1060
        stream = StringIO()
1055
 
        runner = tests.TextTestRunner(stream=stream)
 
1061
        runner = TextTestRunner(stream=stream)
1056
1062
        result = self.run_test_runner(runner, test)
1057
1063
        lines = stream.getvalue().splitlines()
1058
 
        self.assertContainsRe(stream.getvalue(),
1059
 
            '(?sm)^bzr selftest.*$'
1060
 
            '.*'
1061
 
            '^======================================================================\n'
1062
 
            '^FAIL: failing_test\n'
1063
 
            '^----------------------------------------------------------------------\n'
1064
 
            'Traceback \\(most recent call last\\):\n'
1065
 
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1066
 
            '    raise AssertionError\\(\'foo\'\\)\n'
1067
 
            '.*'
1068
 
            '^----------------------------------------------------------------------\n'
1069
 
            '.*'
1070
 
            'FAILED \\(failures=1, known_failure_count=1\\)'
1071
 
            )
 
1064
        self.assertEqual([
 
1065
            '',
 
1066
            '======================================================================',
 
1067
            'FAIL: unittest.FunctionTestCase (failing_test)',
 
1068
            '----------------------------------------------------------------------',
 
1069
            'Traceback (most recent call last):',
 
1070
            '    raise AssertionError(\'foo\')',
 
1071
            'AssertionError: foo',
 
1072
            '',
 
1073
            '----------------------------------------------------------------------',
 
1074
            '',
 
1075
            'FAILED (failures=1, known_failure_count=1)'],
 
1076
            lines[0:5] + lines[6:10] + lines[11:])
1072
1077
 
1073
1078
    def test_known_failure_ok_run(self):
1074
 
        # run a test that generates a known failure which should be printed in
1075
 
        # the final output.
1076
 
        class Test(tests.TestCase):
1077
 
            def known_failure_test(self):
1078
 
                self.knownFailure("Never works...")
1079
 
        test = Test("known_failure_test")
 
1079
        # run a test that generates a known failure which should be printed in the final output.
 
1080
        def known_failure_test():
 
1081
            raise KnownFailure('failed')
 
1082
        test = unittest.FunctionTestCase(known_failure_test)
1080
1083
        stream = StringIO()
1081
 
        runner = tests.TextTestRunner(stream=stream)
 
1084
        runner = TextTestRunner(stream=stream)
1082
1085
        result = self.run_test_runner(runner, test)
1083
1086
        self.assertContainsRe(stream.getvalue(),
1084
1087
            '\n'
1087
1090
            '\n'
1088
1091
            'OK \\(known_failures=1\\)\n')
1089
1092
 
1090
 
    def test_unexpected_success_bad(self):
1091
 
        class Test(tests.TestCase):
1092
 
            def test_truth(self):
1093
 
                self.expectFailure("No absolute truth", self.assertTrue, True)
1094
 
        runner = tests.TextTestRunner(stream=StringIO())
1095
 
        result = self.run_test_runner(runner, Test("test_truth"))
1096
 
        self.assertContainsRe(runner.stream.getvalue(),
1097
 
            "=+\n"
1098
 
            "FAIL: \\S+\.test_truth\n"
1099
 
            "-+\n"
1100
 
            "(?:.*\n)*"
1101
 
            "\\s*(?:Text attachment: )?reason"
1102
 
            "(?:\n-+\n|: {{{)"
1103
 
            "No absolute truth"
1104
 
            "(?:\n-+\n|}}}\n)"
1105
 
            "(?:.*\n)*"
1106
 
            "-+\n"
1107
 
            "Ran 1 test in .*\n"
1108
 
            "\n"
1109
 
            "FAILED \\(failures=1\\)\n\\Z")
1110
 
 
1111
 
    def test_result_decorator(self):
1112
 
        # decorate results
1113
 
        calls = []
1114
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
1115
 
            def startTest(self, test):
1116
 
                ExtendedToOriginalDecorator.startTest(self, test)
1117
 
                calls.append('start')
1118
 
        test = unittest.FunctionTestCase(lambda:None)
1119
 
        stream = StringIO()
1120
 
        runner = tests.TextTestRunner(stream=stream,
1121
 
            result_decorators=[LoggingDecorator])
1122
 
        result = self.run_test_runner(runner, test)
1123
 
        self.assertLength(1, calls)
1124
 
 
1125
1093
    def test_skipped_test(self):
1126
1094
        # run a test that is skipped, and check the suite as a whole still
1127
1095
        # succeeds.
1128
1096
        # skipping_test must be hidden in here so it's not run as a real test
1129
 
        class SkippingTest(tests.TestCase):
1130
 
            def skipping_test(self):
1131
 
                raise tests.TestSkipped('test intentionally skipped')
1132
 
        runner = tests.TextTestRunner(stream=self._log_file)
1133
 
        test = SkippingTest("skipping_test")
 
1097
        def skipping_test():
 
1098
            raise TestSkipped('test intentionally skipped')
 
1099
 
 
1100
        runner = TextTestRunner(stream=self._log_file)
 
1101
        test = unittest.FunctionTestCase(skipping_test)
1134
1102
        result = self.run_test_runner(runner, test)
1135
1103
        self.assertTrue(result.wasSuccessful())
1136
1104
 
1137
1105
    def test_skipped_from_setup(self):
1138
1106
        calls = []
1139
 
        class SkippedSetupTest(tests.TestCase):
 
1107
        class SkippedSetupTest(TestCase):
1140
1108
 
1141
1109
            def setUp(self):
1142
1110
                calls.append('setUp')
1143
1111
                self.addCleanup(self.cleanup)
1144
 
                raise tests.TestSkipped('skipped setup')
 
1112
                raise TestSkipped('skipped setup')
1145
1113
 
1146
1114
            def test_skip(self):
1147
1115
                self.fail('test reached')
1149
1117
            def cleanup(self):
1150
1118
                calls.append('cleanup')
1151
1119
 
1152
 
        runner = tests.TextTestRunner(stream=self._log_file)
 
1120
        runner = TextTestRunner(stream=self._log_file)
1153
1121
        test = SkippedSetupTest('test_skip')
1154
1122
        result = self.run_test_runner(runner, test)
1155
1123
        self.assertTrue(result.wasSuccessful())
1158
1126
 
1159
1127
    def test_skipped_from_test(self):
1160
1128
        calls = []
1161
 
        class SkippedTest(tests.TestCase):
 
1129
        class SkippedTest(TestCase):
1162
1130
 
1163
1131
            def setUp(self):
1164
 
                super(SkippedTest, self).setUp()
1165
1132
                calls.append('setUp')
1166
1133
                self.addCleanup(self.cleanup)
1167
1134
 
1168
1135
            def test_skip(self):
1169
 
                raise tests.TestSkipped('skipped test')
 
1136
                raise TestSkipped('skipped test')
1170
1137
 
1171
1138
            def cleanup(self):
1172
1139
                calls.append('cleanup')
1173
1140
 
1174
 
        runner = tests.TextTestRunner(stream=self._log_file)
 
1141
        runner = TextTestRunner(stream=self._log_file)
1175
1142
        test = SkippedTest('test_skip')
1176
1143
        result = self.run_test_runner(runner, test)
1177
1144
        self.assertTrue(result.wasSuccessful())
1180
1147
 
1181
1148
    def test_not_applicable(self):
1182
1149
        # run a test that is skipped because it's not applicable
1183
 
        class Test(tests.TestCase):
1184
 
            def not_applicable_test(self):
1185
 
                raise tests.TestNotApplicable('this test never runs')
 
1150
        def not_applicable_test():
 
1151
            from bzrlib.tests import TestNotApplicable
 
1152
            raise TestNotApplicable('this test never runs')
1186
1153
        out = StringIO()
1187
 
        runner = tests.TextTestRunner(stream=out, verbosity=2)
1188
 
        test = Test("not_applicable_test")
 
1154
        runner = TextTestRunner(stream=out, verbosity=2)
 
1155
        test = unittest.FunctionTestCase(not_applicable_test)
1189
1156
        result = self.run_test_runner(runner, test)
1190
1157
        self._log_file.write(out.getvalue())
1191
1158
        self.assertTrue(result.wasSuccessful())
1195
1162
        self.assertContainsRe(out.getvalue(),
1196
1163
                r'(?m)^    this test never runs')
1197
1164
 
 
1165
    def test_not_applicable_demo(self):
 
1166
        # just so you can see it in the test output
 
1167
        raise TestNotApplicable('this test is just a demonstation')
 
1168
 
1198
1169
    def test_unsupported_features_listed(self):
1199
1170
        """When unsupported features are encountered they are detailed."""
1200
 
        class Feature1(features.Feature):
 
1171
        class Feature1(Feature):
1201
1172
            def _probe(self): return False
1202
 
        class Feature2(features.Feature):
 
1173
        class Feature2(Feature):
1203
1174
            def _probe(self): return False
1204
1175
        # create sample tests
1205
1176
        test1 = SampleTestCase('_test_pass')
1210
1181
        test.addTest(test1)
1211
1182
        test.addTest(test2)
1212
1183
        stream = StringIO()
1213
 
        runner = tests.TextTestRunner(stream=stream)
 
1184
        runner = TextTestRunner(stream=stream)
1214
1185
        result = self.run_test_runner(runner, test)
1215
1186
        lines = stream.getvalue().splitlines()
1216
1187
        self.assertEqual([
1220
1191
            ],
1221
1192
            lines[-3:])
1222
1193
 
1223
 
    def test_verbose_test_count(self):
1224
 
        """A verbose test run reports the right test count at the start"""
1225
 
        suite = TestUtil.TestSuite([
1226
 
            unittest.FunctionTestCase(lambda:None),
1227
 
            unittest.FunctionTestCase(lambda:None)])
1228
 
        self.assertEqual(suite.countTestCases(), 2)
1229
 
        stream = StringIO()
1230
 
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
1231
 
        # Need to use the CountingDecorator as that's what sets num_tests
1232
 
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
1233
 
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
1234
 
 
1235
 
    def test_startTestRun(self):
1236
 
        """run should call result.startTestRun()"""
1237
 
        calls = []
1238
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
1239
 
            def startTestRun(self):
1240
 
                ExtendedToOriginalDecorator.startTestRun(self)
1241
 
                calls.append('startTestRun')
1242
 
        test = unittest.FunctionTestCase(lambda:None)
1243
 
        stream = StringIO()
1244
 
        runner = tests.TextTestRunner(stream=stream,
1245
 
            result_decorators=[LoggingDecorator])
1246
 
        result = self.run_test_runner(runner, test)
1247
 
        self.assertLength(1, calls)
1248
 
 
1249
 
    def test_stopTestRun(self):
1250
 
        """run should call result.stopTestRun()"""
1251
 
        calls = []
1252
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
1253
 
            def stopTestRun(self):
1254
 
                ExtendedToOriginalDecorator.stopTestRun(self)
1255
 
                calls.append('stopTestRun')
1256
 
        test = unittest.FunctionTestCase(lambda:None)
1257
 
        stream = StringIO()
1258
 
        runner = tests.TextTestRunner(stream=stream,
1259
 
            result_decorators=[LoggingDecorator])
1260
 
        result = self.run_test_runner(runner, test)
1261
 
        self.assertLength(1, calls)
1262
 
 
1263
 
    def test_unicode_test_output_on_ascii_stream(self):
1264
 
        """Showing results should always succeed even on an ascii console"""
1265
 
        class FailureWithUnicode(tests.TestCase):
1266
 
            def test_log_unicode(self):
1267
 
                self.log(u"\u2606")
1268
 
                self.fail("Now print that log!")
1269
 
        out = StringIO()
1270
 
        self.overrideAttr(osutils, "get_terminal_encoding",
1271
 
            lambda trace=False: "ascii")
1272
 
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
1273
 
            FailureWithUnicode("test_log_unicode"))
1274
 
        self.assertContainsRe(out.getvalue(),
1275
 
            "(?:Text attachment: )?log"
1276
 
            "(?:\n-+\n|: {{{)"
1277
 
            "\d+\.\d+  \\\\u2606"
1278
 
            "(?:\n-+\n|}}}\n)")
1279
 
 
1280
 
 
1281
 
class SampleTestCase(tests.TestCase):
 
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
 
1197
        # was tested.
 
1198
        workingtree = _get_bzr_source_tree()
 
1199
        test = TestRunner('dummy_test')
 
1200
        output = StringIO()
 
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)
 
1208
 
 
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)
 
1214
 
 
1215
    def test_success_log_deleted(self):
 
1216
        """Successful tests have their log deleted"""
 
1217
 
 
1218
        class LogTester(TestCase):
 
1219
 
 
1220
            def test_success(self):
 
1221
                self.log('this will be removed\n')
 
1222
 
 
1223
        sio = cStringIO.StringIO()
 
1224
        runner = TextTestRunner(stream=sio)
 
1225
        test = LogTester('test_success')
 
1226
        result = self.run_test_runner(runner, test)
 
1227
 
 
1228
        self.assertLogDeleted(test)
 
1229
 
 
1230
    def test_skipped_log_deleted(self):
 
1231
        """Skipped tests have their log deleted"""
 
1232
 
 
1233
        class LogTester(TestCase):
 
1234
 
 
1235
            def test_skipped(self):
 
1236
                self.log('this will be removed\n')
 
1237
                raise tests.TestSkipped()
 
1238
 
 
1239
        sio = cStringIO.StringIO()
 
1240
        runner = TextTestRunner(stream=sio)
 
1241
        test = LogTester('test_skipped')
 
1242
        result = self.run_test_runner(runner, test)
 
1243
 
 
1244
        self.assertLogDeleted(test)
 
1245
 
 
1246
    def test_not_aplicable_log_deleted(self):
 
1247
        """Not applicable tests have their log deleted"""
 
1248
 
 
1249
        class LogTester(TestCase):
 
1250
 
 
1251
            def test_not_applicable(self):
 
1252
                self.log('this will be removed\n')
 
1253
                raise tests.TestNotApplicable()
 
1254
 
 
1255
        sio = cStringIO.StringIO()
 
1256
        runner = TextTestRunner(stream=sio)
 
1257
        test = LogTester('test_not_applicable')
 
1258
        result = self.run_test_runner(runner, test)
 
1259
 
 
1260
        self.assertLogDeleted(test)
 
1261
 
 
1262
    def test_known_failure_log_deleted(self):
 
1263
        """Know failure tests have their log deleted"""
 
1264
 
 
1265
        class LogTester(TestCase):
 
1266
 
 
1267
            def test_known_failure(self):
 
1268
                self.log('this will be removed\n')
 
1269
                raise tests.KnownFailure()
 
1270
 
 
1271
        sio = cStringIO.StringIO()
 
1272
        runner = TextTestRunner(stream=sio)
 
1273
        test = LogTester('test_known_failure')
 
1274
        result = self.run_test_runner(runner, test)
 
1275
 
 
1276
        self.assertLogDeleted(test)
 
1277
 
 
1278
    def test_fail_log_kept(self):
 
1279
        """Failed tests have their log kept"""
 
1280
 
 
1281
        class LogTester(TestCase):
 
1282
 
 
1283
            def test_fail(self):
 
1284
                self.log('this will be kept\n')
 
1285
                self.fail('this test fails')
 
1286
 
 
1287
        sio = cStringIO.StringIO()
 
1288
        runner = TextTestRunner(stream=sio)
 
1289
        test = LogTester('test_fail')
 
1290
        result = self.run_test_runner(runner, test)
 
1291
 
 
1292
        text = sio.getvalue()
 
1293
        self.assertContainsRe(text, 'this will be kept')
 
1294
        self.assertContainsRe(text, 'this test fails')
 
1295
 
 
1296
        log = test._get_log()
 
1297
        self.assertContainsRe(log, 'this will be kept')
 
1298
        self.assertEqual(log, test._log_contents)
 
1299
 
 
1300
    def test_error_log_kept(self):
 
1301
        """Tests with errors have their log kept"""
 
1302
 
 
1303
        class LogTester(TestCase):
 
1304
 
 
1305
            def test_error(self):
 
1306
                self.log('this will be kept\n')
 
1307
                raise ValueError('random exception raised')
 
1308
 
 
1309
        sio = cStringIO.StringIO()
 
1310
        runner = TextTestRunner(stream=sio)
 
1311
        test = LogTester('test_error')
 
1312
        result = self.run_test_runner(runner, test)
 
1313
 
 
1314
        text = sio.getvalue()
 
1315
        self.assertContainsRe(text, 'this will be kept')
 
1316
        self.assertContainsRe(text, 'random exception raised')
 
1317
 
 
1318
        log = test._get_log()
 
1319
        self.assertContainsRe(log, 'this will be kept')
 
1320
        self.assertEqual(log, test._log_contents)
 
1321
 
 
1322
 
 
1323
class SampleTestCase(TestCase):
1282
1324
 
1283
1325
    def _test_pass(self):
1284
1326
        pass
1286
1328
class _TestException(Exception):
1287
1329
    pass
1288
1330
 
1289
 
 
1290
 
class TestTestCase(tests.TestCase):
 
1331
class TestTestCase(TestCase):
1291
1332
    """Tests that test the core bzrlib TestCase."""
1292
1333
 
1293
 
    def test_assertLength_matches_empty(self):
1294
 
        a_list = []
1295
 
        self.assertLength(0, a_list)
1296
 
 
1297
 
    def test_assertLength_matches_nonempty(self):
1298
 
        a_list = [1, 2, 3]
1299
 
        self.assertLength(3, a_list)
1300
 
 
1301
 
    def test_assertLength_fails_different(self):
1302
 
        a_list = []
1303
 
        self.assertRaises(AssertionError, self.assertLength, 1, a_list)
1304
 
 
1305
 
    def test_assertLength_shows_sequence_in_failure(self):
1306
 
        a_list = [1, 2, 3]
1307
 
        exception = self.assertRaises(AssertionError, self.assertLength, 2,
1308
 
            a_list)
1309
 
        self.assertEqual('Incorrect length: wanted 2, got 3 for [1, 2, 3]',
1310
 
            exception.args[0])
1311
 
 
1312
 
    def test_base_setUp_not_called_causes_failure(self):
1313
 
        class TestCaseWithBrokenSetUp(tests.TestCase):
1314
 
            def setUp(self):
1315
 
                pass # does not call TestCase.setUp
1316
 
            def test_foo(self):
1317
 
                pass
1318
 
        test = TestCaseWithBrokenSetUp('test_foo')
1319
 
        result = unittest.TestResult()
1320
 
        test.run(result)
1321
 
        self.assertFalse(result.wasSuccessful())
1322
 
        self.assertEqual(1, result.testsRun)
1323
 
 
1324
 
    def test_base_tearDown_not_called_causes_failure(self):
1325
 
        class TestCaseWithBrokenTearDown(tests.TestCase):
1326
 
            def tearDown(self):
1327
 
                pass # does not call TestCase.tearDown
1328
 
            def test_foo(self):
1329
 
                pass
1330
 
        test = TestCaseWithBrokenTearDown('test_foo')
1331
 
        result = unittest.TestResult()
1332
 
        test.run(result)
1333
 
        self.assertFalse(result.wasSuccessful())
1334
 
        self.assertEqual(1, result.testsRun)
1335
 
 
1336
1334
    def test_debug_flags_sanitised(self):
1337
1335
        """The bzrlib debug flags should be sanitised by setUp."""
1338
 
        if 'allow_debug' in tests.selftest_debug_flags:
1339
 
            raise tests.TestNotApplicable(
1340
 
                '-Eallow_debug option prevents debug flag sanitisation')
1341
1336
        # we could set something and run a test that will check
1342
1337
        # it gets santised, but this is probably sufficient for now:
1343
1338
        # if someone runs the test with -Dsomething it will error.
1344
 
        flags = set()
1345
 
        if self._lock_check_thorough:
1346
 
            flags.add('strict_locks')
1347
 
        self.assertEqual(flags, bzrlib.debug.debug_flags)
1348
 
 
1349
 
    def change_selftest_debug_flags(self, new_flags):
1350
 
        self.overrideAttr(tests, 'selftest_debug_flags', set(new_flags))
1351
 
 
1352
 
    def test_allow_debug_flag(self):
1353
 
        """The -Eallow_debug flag prevents bzrlib.debug.debug_flags from being
1354
 
        sanitised (i.e. cleared) before running a test.
1355
 
        """
1356
 
        self.change_selftest_debug_flags(set(['allow_debug']))
1357
 
        bzrlib.debug.debug_flags = set(['a-flag'])
1358
 
        class TestThatRecordsFlags(tests.TestCase):
1359
 
            def test_foo(nested_self):
1360
 
                self.flags = set(bzrlib.debug.debug_flags)
1361
 
        test = TestThatRecordsFlags('test_foo')
1362
 
        test.run(self.make_test_result())
1363
 
        flags = set(['a-flag'])
1364
 
        if 'disable_lock_checks' not in tests.selftest_debug_flags:
1365
 
            flags.add('strict_locks')
1366
 
        self.assertEqual(flags, self.flags)
1367
 
 
1368
 
    def test_disable_lock_checks(self):
1369
 
        """The -Edisable_lock_checks flag disables thorough checks."""
1370
 
        class TestThatRecordsFlags(tests.TestCase):
1371
 
            def test_foo(nested_self):
1372
 
                self.flags = set(bzrlib.debug.debug_flags)
1373
 
                self.test_lock_check_thorough = nested_self._lock_check_thorough
1374
 
        self.change_selftest_debug_flags(set())
1375
 
        test = TestThatRecordsFlags('test_foo')
1376
 
        test.run(self.make_test_result())
1377
 
        # By default we do strict lock checking and thorough lock/unlock
1378
 
        # tracking.
1379
 
        self.assertTrue(self.test_lock_check_thorough)
1380
 
        self.assertEqual(set(['strict_locks']), self.flags)
1381
 
        # Now set the disable_lock_checks flag, and show that this changed.
1382
 
        self.change_selftest_debug_flags(set(['disable_lock_checks']))
1383
 
        test = TestThatRecordsFlags('test_foo')
1384
 
        test.run(self.make_test_result())
1385
 
        self.assertFalse(self.test_lock_check_thorough)
1386
 
        self.assertEqual(set(), self.flags)
1387
 
 
1388
 
    def test_this_fails_strict_lock_check(self):
1389
 
        class TestThatRecordsFlags(tests.TestCase):
1390
 
            def test_foo(nested_self):
1391
 
                self.flags1 = set(bzrlib.debug.debug_flags)
1392
 
                self.thisFailsStrictLockCheck()
1393
 
                self.flags2 = set(bzrlib.debug.debug_flags)
1394
 
        # Make sure lock checking is active
1395
 
        self.change_selftest_debug_flags(set())
1396
 
        test = TestThatRecordsFlags('test_foo')
1397
 
        test.run(self.make_test_result())
1398
 
        self.assertEqual(set(['strict_locks']), self.flags1)
1399
 
        self.assertEqual(set(), self.flags2)
1400
 
 
1401
 
    def test_debug_flags_restored(self):
1402
 
        """The bzrlib debug flags should be restored to their original state
1403
 
        after the test was run, even if allow_debug is set.
1404
 
        """
1405
 
        self.change_selftest_debug_flags(set(['allow_debug']))
1406
 
        # Now run a test that modifies debug.debug_flags.
1407
 
        bzrlib.debug.debug_flags = set(['original-state'])
1408
 
        class TestThatModifiesFlags(tests.TestCase):
1409
 
            def test_foo(self):
1410
 
                bzrlib.debug.debug_flags = set(['modified'])
1411
 
        test = TestThatModifiesFlags('test_foo')
1412
 
        test.run(self.make_test_result())
1413
 
        self.assertEqual(set(['original-state']), bzrlib.debug.debug_flags)
1414
 
 
1415
 
    def make_test_result(self):
1416
 
        """Get a test result that writes to the test log file."""
1417
 
        return tests.TextTestResult(self._log_file, descriptions=0, verbosity=1)
 
1339
        self.assertEqual(set(), bzrlib.debug.debug_flags)
1418
1340
 
1419
1341
    def inner_test(self):
1420
1342
        # the inner child test
1424
1346
        # the outer child test
1425
1347
        note("outer_start")
1426
1348
        self.inner_test = TestTestCase("inner_child")
1427
 
        result = self.make_test_result()
 
1349
        result = bzrlib.tests.TextTestResult(self._log_file,
 
1350
                                        descriptions=0,
 
1351
                                        verbosity=1)
1428
1352
        self.inner_test.run(result)
1429
1353
        note("outer finish")
1430
 
        self.addCleanup(osutils.delete_any, self._log_file_name)
1431
1354
 
1432
1355
    def test_trace_nesting(self):
1433
1356
        # this tests that each test case nests its trace facility correctly.
1435
1358
        # should setup a new log, log content to it, setup a child case (B),
1436
1359
        # which should log independently, then case (A) should log a trailer
1437
1360
        # and return.
1438
 
        # we do two nested children so that we can verify the state of the
 
1361
        # we do two nested children so that we can verify the state of the 
1439
1362
        # logs after the outer child finishes is correct, which a bad clean
1440
1363
        # up routine in tearDown might trigger a fault in our test with only
1441
1364
        # one child, we should instead see the bad result inside our test with
1443
1366
        # the outer child test
1444
1367
        original_trace = bzrlib.trace._trace_file
1445
1368
        outer_test = TestTestCase("outer_child")
1446
 
        result = self.make_test_result()
 
1369
        result = bzrlib.tests.TextTestResult(self._log_file,
 
1370
                                        descriptions=0,
 
1371
                                        verbosity=1)
1447
1372
        outer_test.run(result)
1448
1373
        self.assertEqual(original_trace, bzrlib.trace._trace_file)
1449
1374
 
1457
1382
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1458
1383
        output_stream = StringIO()
1459
1384
        result = bzrlib.tests.VerboseTestResult(
1460
 
            output_stream,
 
1385
            unittest._WritelnDecorator(output_stream),
1461
1386
            descriptions=0,
1462
 
            verbosity=2)
 
1387
            verbosity=2,
 
1388
            num_tests=sample_test.countTestCases())
1463
1389
        sample_test.run(result)
1464
1390
        self.assertContainsRe(
1465
1391
            output_stream.getvalue(),
1466
 
            r"\d+ms\*\n$")
 
1392
            r"\d+ms/ +\d+ms\n$")
1467
1393
 
1468
1394
    def test_hooks_sanitised(self):
1469
1395
        """The bzrlib hooks should be sanitised by setUp."""
1470
 
        # Note this test won't fail with hooks that the core library doesn't
1471
 
        # use - but it trigger with a plugin that adds hooks, so its still a
1472
 
        # useful warning in that case.
1473
 
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
1474
 
        self.assertEqual(
1475
 
            bzrlib.smart.server.SmartServerHooks(),
 
1396
        self.assertEqual(bzrlib.branch.BranchHooks(),
 
1397
            bzrlib.branch.Branch.hooks)
 
1398
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
1476
1399
            bzrlib.smart.server.SmartTCPServer.hooks)
1477
 
        self.assertEqual(
1478
 
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
1479
1400
 
1480
1401
    def test__gather_lsprof_in_benchmarks(self):
1481
1402
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1482
 
 
 
1403
        
1483
1404
        Each self.time() call is individually and separately profiled.
1484
1405
        """
1485
 
        self.requireFeature(features.lsprof_feature)
1486
 
        # overrides the class member with an instance member so no cleanup
 
1406
        self.requireFeature(test_lsprof.LSProfFeature)
 
1407
        # overrides the class member with an instance member so no cleanup 
1487
1408
        # needed.
1488
1409
        self._gather_lsprof_in_benchmarks = True
1489
1410
        self.time(time.sleep, 0.000)
1493
1414
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1494
1415
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1495
1416
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
1496
 
        del self._benchcalls[:]
1497
1417
 
1498
1418
    def test_knownFailure(self):
1499
1419
        """Self.knownFailure() should raise a KnownFailure exception."""
1500
 
        self.assertRaises(tests.KnownFailure, self.knownFailure, "A Failure")
1501
 
 
1502
 
    def test_open_bzrdir_safe_roots(self):
1503
 
        # even a memory transport should fail to open when its url isn't 
1504
 
        # permitted.
1505
 
        # Manually set one up (TestCase doesn't and shouldn't provide magic
1506
 
        # machinery)
1507
 
        transport_server = memory.MemoryServer()
1508
 
        transport_server.start_server()
1509
 
        self.addCleanup(transport_server.stop_server)
1510
 
        t = transport.get_transport_from_url(transport_server.get_url())
1511
 
        controldir.ControlDir.create(t.base)
1512
 
        self.assertRaises(errors.BzrError,
1513
 
            controldir.ControlDir.open_from_transport, t)
1514
 
        # But if we declare this as safe, we can open the bzrdir.
1515
 
        self.permit_url(t.base)
1516
 
        self._bzr_selftest_roots.append(t.base)
1517
 
        controldir.ControlDir.open_from_transport(t)
 
1420
        self.assertRaises(KnownFailure, self.knownFailure, "A Failure")
1518
1421
 
1519
1422
    def test_requireFeature_available(self):
1520
1423
        """self.requireFeature(available) is a no-op."""
1521
 
        class Available(features.Feature):
 
1424
        class Available(Feature):
1522
1425
            def _probe(self):return True
1523
1426
        feature = Available()
1524
1427
        self.requireFeature(feature)
1525
1428
 
1526
1429
    def test_requireFeature_unavailable(self):
1527
1430
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1528
 
        class Unavailable(features.Feature):
 
1431
        class Unavailable(Feature):
1529
1432
            def _probe(self):return False
1530
1433
        feature = Unavailable()
1531
 
        self.assertRaises(tests.UnavailableFeature,
1532
 
                          self.requireFeature, feature)
 
1434
        self.assertRaises(UnavailableFeature, self.requireFeature, feature)
1533
1435
 
1534
1436
    def test_run_no_parameters(self):
1535
1437
        test = SampleTestCase('_test_pass')
1536
1438
        test.run()
1537
 
 
 
1439
    
1538
1440
    def test_run_enabled_unittest_result(self):
1539
1441
        """Test we revert to regular behaviour when the test is enabled."""
1540
1442
        test = SampleTestCase('_test_pass')
1565
1467
        """Test disabled tests behaviour with support aware results."""
1566
1468
        test = SampleTestCase('_test_pass')
1567
1469
        class DisabledFeature(object):
1568
 
            def __eq__(self, other):
1569
 
                return isinstance(other, DisabledFeature)
1570
1470
            def available(self):
1571
1471
                return False
1572
1472
        the_feature = DisabledFeature()
1583
1483
                self.calls.append(('addNotSupported', test, feature))
1584
1484
        result = InstrumentedTestResult()
1585
1485
        test.run(result)
1586
 
        case = result.calls[0][1]
1587
1486
        self.assertEqual([
1588
 
            ('startTest', case),
1589
 
            ('addNotSupported', case, the_feature),
1590
 
            ('stopTest', case),
 
1487
            ('startTest', test),
 
1488
            ('addNotSupported', test, the_feature),
 
1489
            ('stopTest', test),
1591
1490
            ],
1592
1491
            result.calls)
1593
1492
 
1594
 
    def test_start_server_registers_url(self):
1595
 
        transport_server = memory.MemoryServer()
1596
 
        # A little strict, but unlikely to be changed soon.
1597
 
        self.assertEqual([], self._bzr_selftest_roots)
1598
 
        self.start_server(transport_server)
1599
 
        self.assertSubset([transport_server.get_url()],
1600
 
            self._bzr_selftest_roots)
1601
 
 
1602
1493
    def test_assert_list_raises_on_generator(self):
1603
1494
        def generator_which_will_raise():
1604
1495
            # This will not raise until after the first yield
1654
1545
        self.assertRaises(AssertionError,
1655
1546
            self.assertListRaises, _TestException, success_generator)
1656
1547
 
1657
 
    def _run_successful_test(self, test):
1658
 
        result = testtools.TestResult()
1659
 
        test.run(result)
1660
 
        self.assertTrue(result.wasSuccessful())
1661
 
        return result
1662
 
 
1663
 
    def test_overrideAttr_without_value(self):
1664
 
        self.test_attr = 'original' # Define a test attribute
1665
 
        obj = self # Make 'obj' visible to the embedded test
1666
 
        class Test(tests.TestCase):
1667
 
 
1668
 
            def setUp(self):
1669
 
                super(Test, self).setUp()
1670
 
                self.orig = self.overrideAttr(obj, 'test_attr')
1671
 
 
1672
 
            def test_value(self):
1673
 
                self.assertEqual('original', self.orig)
1674
 
                self.assertEqual('original', obj.test_attr)
1675
 
                obj.test_attr = 'modified'
1676
 
                self.assertEqual('modified', obj.test_attr)
1677
 
 
1678
 
        self._run_successful_test(Test('test_value'))
1679
 
        self.assertEqual('original', obj.test_attr)
1680
 
 
1681
 
    def test_overrideAttr_with_value(self):
1682
 
        self.test_attr = 'original' # Define a test attribute
1683
 
        obj = self # Make 'obj' visible to the embedded test
1684
 
        class Test(tests.TestCase):
1685
 
 
1686
 
            def setUp(self):
1687
 
                super(Test, self).setUp()
1688
 
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1689
 
 
1690
 
            def test_value(self):
1691
 
                self.assertEqual('original', self.orig)
1692
 
                self.assertEqual('modified', obj.test_attr)
1693
 
 
1694
 
        self._run_successful_test(Test('test_value'))
1695
 
        self.assertEqual('original', obj.test_attr)
1696
 
 
1697
 
    def test_overrideAttr_with_no_existing_value_and_value(self):
1698
 
        # Do not define the test_attribute
1699
 
        obj = self # Make 'obj' visible to the embedded test
1700
 
        class Test(tests.TestCase):
1701
 
 
1702
 
            def setUp(self):
1703
 
                tests.TestCase.setUp(self)
1704
 
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1705
 
 
1706
 
            def test_value(self):
1707
 
                self.assertEqual(tests._unitialized_attr, self.orig)
1708
 
                self.assertEqual('modified', obj.test_attr)
1709
 
 
1710
 
        self._run_successful_test(Test('test_value'))
1711
 
        self.assertRaises(AttributeError, getattr, obj, 'test_attr')
1712
 
 
1713
 
    def test_overrideAttr_with_no_existing_value_and_no_value(self):
1714
 
        # Do not define the test_attribute
1715
 
        obj = self # Make 'obj' visible to the embedded test
1716
 
        class Test(tests.TestCase):
1717
 
 
1718
 
            def setUp(self):
1719
 
                tests.TestCase.setUp(self)
1720
 
                self.orig = self.overrideAttr(obj, 'test_attr')
1721
 
 
1722
 
            def test_value(self):
1723
 
                self.assertEqual(tests._unitialized_attr, self.orig)
1724
 
                self.assertRaises(AttributeError, getattr, obj, 'test_attr')
1725
 
 
1726
 
        self._run_successful_test(Test('test_value'))
1727
 
        self.assertRaises(AttributeError, getattr, obj, 'test_attr')
1728
 
 
1729
 
    def test_recordCalls(self):
1730
 
        from bzrlib.tests import test_selftest
1731
 
        calls = self.recordCalls(
1732
 
            test_selftest, '_add_numbers')
1733
 
        self.assertEqual(test_selftest._add_numbers(2, 10),
1734
 
            12)
1735
 
        self.assertEqual(calls, [((2, 10), {})])
1736
 
 
1737
 
 
1738
 
def _add_numbers(a, b):
1739
 
    return a + b
1740
 
 
1741
 
 
1742
 
class _MissingFeature(features.Feature):
1743
 
    def _probe(self):
1744
 
        return False
1745
 
missing_feature = _MissingFeature()
1746
 
 
1747
 
 
1748
 
def _get_test(name):
1749
 
    """Get an instance of a specific example test.
1750
 
 
1751
 
    We protect this in a function so that they don't auto-run in the test
1752
 
    suite.
1753
 
    """
1754
 
 
1755
 
    class ExampleTests(tests.TestCase):
1756
 
 
1757
 
        def test_fail(self):
1758
 
            mutter('this was a failing test')
1759
 
            self.fail('this test will fail')
1760
 
 
1761
 
        def test_error(self):
1762
 
            mutter('this test errored')
1763
 
            raise RuntimeError('gotcha')
1764
 
 
1765
 
        def test_missing_feature(self):
1766
 
            mutter('missing the feature')
1767
 
            self.requireFeature(missing_feature)
1768
 
 
1769
 
        def test_skip(self):
1770
 
            mutter('this test will be skipped')
1771
 
            raise tests.TestSkipped('reason')
1772
 
 
1773
 
        def test_success(self):
1774
 
            mutter('this test succeeds')
1775
 
 
1776
 
        def test_xfail(self):
1777
 
            mutter('test with expected failure')
1778
 
            self.knownFailure('this_fails')
1779
 
 
1780
 
        def test_unexpected_success(self):
1781
 
            mutter('test with unexpected success')
1782
 
            self.expectFailure('should_fail', lambda: None)
1783
 
 
1784
 
    return ExampleTests(name)
1785
 
 
1786
 
 
1787
 
class TestTestCaseLogDetails(tests.TestCase):
1788
 
 
1789
 
    def _run_test(self, test_name):
1790
 
        test = _get_test(test_name)
1791
 
        result = testtools.TestResult()
1792
 
        test.run(result)
1793
 
        return result
1794
 
 
1795
 
    def test_fail_has_log(self):
1796
 
        result = self._run_test('test_fail')
1797
 
        self.assertEqual(1, len(result.failures))
1798
 
        result_content = result.failures[0][1]
1799
 
        self.assertContainsRe(result_content,
1800
 
            '(?m)^(?:Text attachment: )?log(?:$|: )')
1801
 
        self.assertContainsRe(result_content, 'this was a failing test')
1802
 
 
1803
 
    def test_error_has_log(self):
1804
 
        result = self._run_test('test_error')
1805
 
        self.assertEqual(1, len(result.errors))
1806
 
        result_content = result.errors[0][1]
1807
 
        self.assertContainsRe(result_content,
1808
 
            '(?m)^(?:Text attachment: )?log(?:$|: )')
1809
 
        self.assertContainsRe(result_content, 'this test errored')
1810
 
 
1811
 
    def test_skip_has_no_log(self):
1812
 
        result = self._run_test('test_skip')
1813
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
1814
 
        skips = result.skip_reasons['reason']
1815
 
        self.assertEqual(1, len(skips))
1816
 
        test = skips[0]
1817
 
        self.assertFalse('log' in test.getDetails())
1818
 
 
1819
 
    def test_missing_feature_has_no_log(self):
1820
 
        # testtools doesn't know about addNotSupported, so it just gets
1821
 
        # considered as a skip
1822
 
        result = self._run_test('test_missing_feature')
1823
 
        self.assertEqual([missing_feature], result.skip_reasons.keys())
1824
 
        skips = result.skip_reasons[missing_feature]
1825
 
        self.assertEqual(1, len(skips))
1826
 
        test = skips[0]
1827
 
        self.assertFalse('log' in test.getDetails())
1828
 
 
1829
 
    def test_xfail_has_no_log(self):
1830
 
        result = self._run_test('test_xfail')
1831
 
        self.assertEqual(1, len(result.expectedFailures))
1832
 
        result_content = result.expectedFailures[0][1]
1833
 
        self.assertNotContainsRe(result_content,
1834
 
            '(?m)^(?:Text attachment: )?log(?:$|: )')
1835
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
1836
 
 
1837
 
    def test_unexpected_success_has_log(self):
1838
 
        result = self._run_test('test_unexpected_success')
1839
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
1840
 
        # Inconsistency, unexpectedSuccesses is a list of tests,
1841
 
        # expectedFailures is a list of reasons?
1842
 
        test = result.unexpectedSuccesses[0]
1843
 
        details = test.getDetails()
1844
 
        self.assertTrue('log' in details)
1845
 
 
1846
 
 
1847
 
class TestTestCloning(tests.TestCase):
1848
 
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
1849
 
 
1850
 
    def test_cloned_testcase_does_not_share_details(self):
1851
 
        """A TestCase cloned with clone_test does not share mutable attributes
1852
 
        such as details or cleanups.
1853
 
        """
1854
 
        class Test(tests.TestCase):
1855
 
            def test_foo(self):
1856
 
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1857
 
        orig_test = Test('test_foo')
1858
 
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1859
 
        orig_test.run(unittest.TestResult())
1860
 
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1861
 
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
1862
 
 
1863
 
    def test_double_apply_scenario_preserves_first_scenario(self):
1864
 
        """Applying two levels of scenarios to a test preserves the attributes
1865
 
        added by both scenarios.
1866
 
        """
1867
 
        class Test(tests.TestCase):
1868
 
            def test_foo(self):
1869
 
                pass
1870
 
        test = Test('test_foo')
1871
 
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1872
 
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1873
 
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1874
 
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1875
 
        all_tests = list(tests.iter_suite_tests(suite))
1876
 
        self.assertLength(4, all_tests)
1877
 
        all_xys = sorted((t.x, t.y) for t in all_tests)
1878
 
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1879
 
 
1880
 
 
1881
 
# NB: Don't delete this; it's not actually from 0.11!
1882
 
@deprecated_function(deprecated_in((0, 11, 0)))
 
1548
 
 
1549
@symbol_versioning.deprecated_function(zero_eleven)
1883
1550
def sample_deprecated_function():
1884
1551
    """A deprecated function to test applyDeprecated with."""
1885
1552
    return 2
1892
1559
class ApplyDeprecatedHelper(object):
1893
1560
    """A helper class for ApplyDeprecated tests."""
1894
1561
 
1895
 
    @deprecated_method(deprecated_in((0, 11, 0)))
 
1562
    @symbol_versioning.deprecated_method(zero_eleven)
1896
1563
    def sample_deprecated_method(self, param_one):
1897
1564
        """A deprecated method for testing with."""
1898
1565
        return param_one
1900
1567
    def sample_normal_method(self):
1901
1568
        """A undeprecated method."""
1902
1569
 
1903
 
    @deprecated_method(deprecated_in((0, 10, 0)))
 
1570
    @symbol_versioning.deprecated_method(zero_ten)
1904
1571
    def sample_nested_deprecation(self):
1905
1572
        return sample_deprecated_function()
1906
1573
 
1907
1574
 
1908
 
class TestExtraAssertions(tests.TestCase):
 
1575
class TestExtraAssertions(TestCase):
1909
1576
    """Tests for new test assertions in bzrlib test suite"""
1910
1577
 
1911
1578
    def test_assert_isinstance(self):
1912
1579
        self.assertIsInstance(2, int)
1913
1580
        self.assertIsInstance(u'', basestring)
1914
 
        e = self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1915
 
        self.assertEqual(str(e),
1916
 
            "None is an instance of <type 'NoneType'> rather than <type 'int'>")
 
1581
        self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1917
1582
        self.assertRaises(AssertionError, self.assertIsInstance, 23.3, int)
1918
 
        e = self.assertRaises(AssertionError,
1919
 
            self.assertIsInstance, None, int, "it's just not")
1920
 
        self.assertEqual(str(e),
1921
 
            "None is an instance of <type 'NoneType'> rather than <type 'int'>"
1922
 
            ": it's just not")
1923
1583
 
1924
1584
    def test_assertEndsWith(self):
1925
1585
        self.assertEndsWith('foo', 'oo')
1926
1586
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1927
1587
 
1928
 
    def test_assertEqualDiff(self):
1929
 
        e = self.assertRaises(AssertionError,
1930
 
                              self.assertEqualDiff, '', '\n')
1931
 
        self.assertEqual(str(e),
1932
 
                          # Don't blink ! The '+' applies to the second string
1933
 
                          'first string is missing a final newline.\n+ \n')
1934
 
        e = self.assertRaises(AssertionError,
1935
 
                              self.assertEqualDiff, '\n', '')
1936
 
        self.assertEqual(str(e),
1937
 
                          # Don't blink ! The '-' applies to the second string
1938
 
                          'second string is missing a final newline.\n- \n')
1939
 
 
1940
 
 
1941
 
class TestDeprecations(tests.TestCase):
1942
 
 
1943
1588
    def test_applyDeprecated_not_deprecated(self):
1944
1589
        sample_object = ApplyDeprecatedHelper()
1945
1590
        # calling an undeprecated callable raises an assertion
1946
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1947
 
            deprecated_in((0, 11, 0)),
 
1591
        self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
1948
1592
            sample_object.sample_normal_method)
1949
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1950
 
            deprecated_in((0, 11, 0)),
 
1593
        self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
1951
1594
            sample_undeprecated_function, "a param value")
1952
1595
        # calling a deprecated callable (function or method) with the wrong
1953
1596
        # expected deprecation fails.
1954
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1955
 
            deprecated_in((0, 10, 0)),
 
1597
        self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
1956
1598
            sample_object.sample_deprecated_method, "a param value")
1957
 
        self.assertRaises(AssertionError, self.applyDeprecated,
1958
 
            deprecated_in((0, 10, 0)),
 
1599
        self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
1959
1600
            sample_deprecated_function)
1960
1601
        # calling a deprecated callable (function or method) with the right
1961
1602
        # expected deprecation returns the functions result.
1962
 
        self.assertEqual("a param value",
1963
 
            self.applyDeprecated(deprecated_in((0, 11, 0)),
 
1603
        self.assertEqual("a param value", self.applyDeprecated(zero_eleven,
1964
1604
            sample_object.sample_deprecated_method, "a param value"))
1965
 
        self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 11, 0)),
 
1605
        self.assertEqual(2, self.applyDeprecated(zero_eleven,
1966
1606
            sample_deprecated_function))
1967
1607
        # calling a nested deprecation with the wrong deprecation version
1968
 
        # fails even if a deeper nested function was deprecated with the
 
1608
        # fails even if a deeper nested function was deprecated with the 
1969
1609
        # supplied version.
1970
1610
        self.assertRaises(AssertionError, self.applyDeprecated,
1971
 
            deprecated_in((0, 11, 0)), sample_object.sample_nested_deprecation)
 
1611
            zero_eleven, sample_object.sample_nested_deprecation)
1972
1612
        # calling a nested deprecation with the right deprecation value
1973
1613
        # returns the calls result.
1974
 
        self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 10, 0)),
 
1614
        self.assertEqual(2, self.applyDeprecated(zero_ten,
1975
1615
            sample_object.sample_nested_deprecation))
1976
1616
 
1977
1617
    def test_callDeprecated(self):
1978
1618
        def testfunc(be_deprecated, result=None):
1979
1619
            if be_deprecated is True:
1980
 
                symbol_versioning.warn('i am deprecated', DeprecationWarning,
 
1620
                symbol_versioning.warn('i am deprecated', DeprecationWarning, 
1981
1621
                                       stacklevel=1)
1982
1622
            return result
1983
1623
        result = self.callDeprecated(['i am deprecated'], testfunc, True)
1988
1628
        self.callDeprecated([], testfunc, be_deprecated=False)
1989
1629
 
1990
1630
 
1991
 
class TestWarningTests(tests.TestCase):
 
1631
class TestWarningTests(TestCase):
1992
1632
    """Tests for calling methods that raise warnings."""
1993
1633
 
1994
1634
    def test_callCatchWarnings(self):
1996
1636
            warnings.warn("this is your last warning")
1997
1637
            return a + b
1998
1638
        wlist, result = self.callCatchWarnings(meth, 1, 2)
1999
 
        self.assertEqual(3, result)
 
1639
        self.assertEquals(3, result)
2000
1640
        # would like just to compare them, but UserWarning doesn't implement
2001
1641
        # eq well
2002
1642
        w0, = wlist
2003
1643
        self.assertIsInstance(w0, UserWarning)
2004
 
        self.assertEqual("this is your last warning", str(w0))
2005
 
 
2006
 
 
2007
 
class TestConvenienceMakers(tests.TestCaseWithTransport):
 
1644
        self.assertEquals("this is your last warning", str(w0))
 
1645
 
 
1646
 
 
1647
class TestConvenienceMakers(TestCaseWithTransport):
2008
1648
    """Test for the make_* convenience functions."""
2009
1649
 
2010
1650
    def test_make_branch_and_tree_with_format(self):
2011
1651
        # we should be able to supply a format to make_branch_and_tree
2012
1652
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
2013
 
        self.assertIsInstance(bzrlib.controldir.ControlDir.open('a')._format,
 
1653
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
 
1654
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
2014
1655
                              bzrlib.bzrdir.BzrDirMetaFormat1)
 
1656
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
 
1657
                              bzrlib.bzrdir.BzrDirFormat6)
2015
1658
 
2016
1659
    def test_make_branch_and_memory_tree(self):
2017
1660
        # we should be able to get a new branch and a mutable tree from
2019
1662
        tree = self.make_branch_and_memory_tree('a')
2020
1663
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
2021
1664
 
2022
 
    def test_make_tree_for_local_vfs_backed_transport(self):
2023
 
        # make_branch_and_tree has to use local branch and repositories
2024
 
        # when the vfs transport and local disk are colocated, even if
2025
 
        # a different transport is in use for url generation.
2026
 
        self.transport_server = test_server.FakeVFATServer
2027
 
        self.assertFalse(self.get_url('t1').startswith('file://'))
 
1665
 
 
1666
class TestSFTPMakeBranchAndTree(TestCaseWithSFTPServer):
 
1667
 
 
1668
    def test_make_tree_for_sftp_branch(self):
 
1669
        """Transports backed by local directories create local trees."""
 
1670
 
2028
1671
        tree = self.make_branch_and_tree('t1')
2029
1672
        base = tree.bzrdir.root_transport.base
2030
 
        self.assertStartsWith(base, 'file://')
2031
 
        self.assertEqual(tree.bzrdir.root_transport,
 
1673
        self.failIf(base.startswith('sftp'),
 
1674
                'base %r is on sftp but should be local' % base)
 
1675
        self.assertEquals(tree.bzrdir.root_transport,
2032
1676
                tree.branch.bzrdir.root_transport)
2033
 
        self.assertEqual(tree.bzrdir.root_transport,
 
1677
        self.assertEquals(tree.bzrdir.root_transport,
2034
1678
                tree.branch.repository.bzrdir.root_transport)
2035
1679
 
2036
1680
 
2037
 
class SelfTestHelper(object):
2038
 
 
2039
 
    def run_selftest(self, **kwargs):
2040
 
        """Run selftest returning its output."""
2041
 
        output = StringIO()
2042
 
        old_transport = bzrlib.tests.default_transport
2043
 
        old_root = tests.TestCaseWithMemoryTransport.TEST_ROOT
2044
 
        tests.TestCaseWithMemoryTransport.TEST_ROOT = None
2045
 
        try:
2046
 
            self.assertEqual(True, tests.selftest(stream=output, **kwargs))
2047
 
        finally:
2048
 
            bzrlib.tests.default_transport = old_transport
2049
 
            tests.TestCaseWithMemoryTransport.TEST_ROOT = old_root
2050
 
        output.seek(0)
2051
 
        return output
2052
 
 
2053
 
 
2054
 
class TestSelftest(tests.TestCase, SelfTestHelper):
 
1681
class TestSelftest(TestCase):
2055
1682
    """Tests of bzrlib.tests.selftest."""
2056
1683
 
2057
1684
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
2058
1685
        factory_called = []
2059
1686
        def factory():
2060
1687
            factory_called.append(True)
2061
 
            return TestUtil.TestSuite()
 
1688
            return TestSuite()
2062
1689
        out = StringIO()
2063
1690
        err = StringIO()
2064
 
        self.apply_redirected(out, err, None, bzrlib.tests.selftest,
 
1691
        self.apply_redirected(out, err, None, bzrlib.tests.selftest, 
2065
1692
            test_suite_factory=factory)
2066
1693
        self.assertEqual([True], factory_called)
2067
1694
 
2068
 
    def factory(self):
2069
 
        """A test suite factory."""
2070
 
        class Test(tests.TestCase):
2071
 
            def a(self):
2072
 
                pass
2073
 
            def b(self):
2074
 
                pass
2075
 
            def c(self):
2076
 
                pass
2077
 
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
2078
 
 
2079
 
    def test_list_only(self):
2080
 
        output = self.run_selftest(test_suite_factory=self.factory,
2081
 
            list_only=True)
2082
 
        self.assertEqual(3, len(output.readlines()))
2083
 
 
2084
 
    def test_list_only_filtered(self):
2085
 
        output = self.run_selftest(test_suite_factory=self.factory,
2086
 
            list_only=True, pattern="Test.b")
2087
 
        self.assertEndsWith(output.getvalue(), "Test.b\n")
2088
 
        self.assertLength(1, output.readlines())
2089
 
 
2090
 
    def test_list_only_excludes(self):
2091
 
        output = self.run_selftest(test_suite_factory=self.factory,
2092
 
            list_only=True, exclude_pattern="Test.b")
2093
 
        self.assertNotContainsRe("Test.b", output.getvalue())
2094
 
        self.assertLength(2, output.readlines())
2095
 
 
2096
 
    def test_lsprof_tests(self):
2097
 
        self.requireFeature(features.lsprof_feature)
2098
 
        results = []
2099
 
        class Test(object):
2100
 
            def __call__(test, result):
2101
 
                test.run(result)
2102
 
            def run(test, result):
2103
 
                results.append(result)
2104
 
            def countTestCases(self):
2105
 
                return 1
2106
 
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
2107
 
        self.assertLength(1, results)
2108
 
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
2109
 
 
2110
 
    def test_random(self):
2111
 
        # test randomising by listing a number of tests.
2112
 
        output_123 = self.run_selftest(test_suite_factory=self.factory,
2113
 
            list_only=True, random_seed="123")
2114
 
        output_234 = self.run_selftest(test_suite_factory=self.factory,
2115
 
            list_only=True, random_seed="234")
2116
 
        self.assertNotEqual(output_123, output_234)
2117
 
        # "Randominzing test order..\n\n
2118
 
        self.assertLength(5, output_123.readlines())
2119
 
        self.assertLength(5, output_234.readlines())
2120
 
 
2121
 
    def test_random_reuse_is_same_order(self):
2122
 
        # test randomising by listing a number of tests.
2123
 
        expected = self.run_selftest(test_suite_factory=self.factory,
2124
 
            list_only=True, random_seed="123")
2125
 
        repeated = self.run_selftest(test_suite_factory=self.factory,
2126
 
            list_only=True, random_seed="123")
2127
 
        self.assertEqual(expected.getvalue(), repeated.getvalue())
2128
 
 
2129
 
    def test_runner_class(self):
2130
 
        self.requireFeature(features.subunit)
2131
 
        from subunit import ProtocolTestCase
2132
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2133
 
            test_suite_factory=self.factory)
2134
 
        test = ProtocolTestCase(stream)
2135
 
        result = unittest.TestResult()
2136
 
        test.run(result)
2137
 
        self.assertEqual(3, result.testsRun)
2138
 
 
2139
 
    def test_starting_with_single_argument(self):
2140
 
        output = self.run_selftest(test_suite_factory=self.factory,
2141
 
            starting_with=['bzrlib.tests.test_selftest.Test.a'],
2142
 
            list_only=True)
2143
 
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n',
2144
 
            output.getvalue())
2145
 
 
2146
 
    def test_starting_with_multiple_argument(self):
2147
 
        output = self.run_selftest(test_suite_factory=self.factory,
2148
 
            starting_with=['bzrlib.tests.test_selftest.Test.a',
2149
 
                'bzrlib.tests.test_selftest.Test.b'],
2150
 
            list_only=True)
2151
 
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n'
2152
 
            'bzrlib.tests.test_selftest.Test.b\n',
2153
 
            output.getvalue())
2154
 
 
2155
 
    def check_transport_set(self, transport_server):
2156
 
        captured_transport = []
2157
 
        def seen_transport(a_transport):
2158
 
            captured_transport.append(a_transport)
2159
 
        class Capture(tests.TestCase):
2160
 
            def a(self):
2161
 
                seen_transport(bzrlib.tests.default_transport)
2162
 
        def factory():
2163
 
            return TestUtil.TestSuite([Capture("a")])
2164
 
        self.run_selftest(transport=transport_server, test_suite_factory=factory)
2165
 
        self.assertEqual(transport_server, captured_transport[0])
2166
 
 
2167
 
    def test_transport_sftp(self):
2168
 
        self.requireFeature(features.paramiko)
2169
 
        from bzrlib.tests import stub_sftp
2170
 
        self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
2171
 
 
2172
 
    def test_transport_memory(self):
2173
 
        self.check_transport_set(memory.MemoryServer)
2174
 
 
2175
 
 
2176
 
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
2177
 
    # Does IO: reads test.list
2178
 
 
2179
 
    def test_load_list(self):
2180
 
        # Provide a list with one test - this test.
2181
 
        test_id_line = '%s\n' % self.id()
2182
 
        self.build_tree_contents([('test.list', test_id_line)])
2183
 
        # And generate a list of the tests in  the suite.
2184
 
        stream = self.run_selftest(load_list='test.list', list_only=True)
2185
 
        self.assertEqual(test_id_line, stream.getvalue())
2186
 
 
2187
 
    def test_load_unknown(self):
2188
 
        # Provide a list with one test - this test.
2189
 
        # And generate a list of the tests in  the suite.
2190
 
        err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
2191
 
            load_list='missing file name', list_only=True)
2192
 
 
2193
 
 
2194
 
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2195
 
 
2196
 
    _test_needs_features = [features.subunit]
2197
 
 
2198
 
    def run_subunit_stream(self, test_name):
2199
 
        from subunit import ProtocolTestCase
2200
 
        def factory():
2201
 
            return TestUtil.TestSuite([_get_test(test_name)])
2202
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2203
 
            test_suite_factory=factory)
2204
 
        test = ProtocolTestCase(stream)
2205
 
        result = testtools.TestResult()
2206
 
        test.run(result)
2207
 
        content = stream.getvalue()
2208
 
        return content, result
2209
 
 
2210
 
    def test_fail_has_log(self):
2211
 
        content, result = self.run_subunit_stream('test_fail')
2212
 
        self.assertEqual(1, len(result.failures))
2213
 
        self.assertContainsRe(content, '(?m)^log$')
2214
 
        self.assertContainsRe(content, 'this test will fail')
2215
 
 
2216
 
    def test_error_has_log(self):
2217
 
        content, result = self.run_subunit_stream('test_error')
2218
 
        self.assertContainsRe(content, '(?m)^log$')
2219
 
        self.assertContainsRe(content, 'this test errored')
2220
 
 
2221
 
    def test_skip_has_no_log(self):
2222
 
        content, result = self.run_subunit_stream('test_skip')
2223
 
        self.assertNotContainsRe(content, '(?m)^log$')
2224
 
        self.assertNotContainsRe(content, 'this test will be skipped')
2225
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
2226
 
        skips = result.skip_reasons['reason']
2227
 
        self.assertEqual(1, len(skips))
2228
 
        test = skips[0]
2229
 
        # RemotedTestCase doesn't preserve the "details"
2230
 
        ## self.assertFalse('log' in test.getDetails())
2231
 
 
2232
 
    def test_missing_feature_has_no_log(self):
2233
 
        content, result = self.run_subunit_stream('test_missing_feature')
2234
 
        self.assertNotContainsRe(content, '(?m)^log$')
2235
 
        self.assertNotContainsRe(content, 'missing the feature')
2236
 
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
2237
 
        skips = result.skip_reasons['_MissingFeature\n']
2238
 
        self.assertEqual(1, len(skips))
2239
 
        test = skips[0]
2240
 
        # RemotedTestCase doesn't preserve the "details"
2241
 
        ## self.assertFalse('log' in test.getDetails())
2242
 
 
2243
 
    def test_xfail_has_no_log(self):
2244
 
        content, result = self.run_subunit_stream('test_xfail')
2245
 
        self.assertNotContainsRe(content, '(?m)^log$')
2246
 
        self.assertNotContainsRe(content, 'test with expected failure')
2247
 
        self.assertEqual(1, len(result.expectedFailures))
2248
 
        result_content = result.expectedFailures[0][1]
2249
 
        self.assertNotContainsRe(result_content,
2250
 
            '(?m)^(?:Text attachment: )?log(?:$|: )')
2251
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
2252
 
 
2253
 
    def test_unexpected_success_has_log(self):
2254
 
        content, result = self.run_subunit_stream('test_unexpected_success')
2255
 
        self.assertContainsRe(content, '(?m)^log$')
2256
 
        self.assertContainsRe(content, 'test with unexpected success')
2257
 
        # GZ 2011-05-18: Old versions of subunit treat unexpected success as a
2258
 
        #                success, if a min version check is added remove this
2259
 
        from subunit import TestProtocolClient as _Client
2260
 
        if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
2261
 
            self.expectFailure('subunit treats "unexpectedSuccess"'
2262
 
                               ' as a plain success',
2263
 
                self.assertEqual, 1, len(result.unexpectedSuccesses))
2264
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
2265
 
        test = result.unexpectedSuccesses[0]
2266
 
        # RemotedTestCase doesn't preserve the "details"
2267
 
        ## self.assertTrue('log' in test.getDetails())
2268
 
 
2269
 
    def test_success_has_no_log(self):
2270
 
        content, result = self.run_subunit_stream('test_success')
2271
 
        self.assertEqual(1, result.testsRun)
2272
 
        self.assertNotContainsRe(content, '(?m)^log$')
2273
 
        self.assertNotContainsRe(content, 'this test succeeds')
2274
 
 
2275
 
 
2276
 
class TestRunBzr(tests.TestCase):
2277
 
 
2278
 
    out = ''
2279
 
    err = ''
2280
 
 
2281
 
    def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
2282
 
                         working_dir=None):
2283
 
        """Override _run_bzr_core to test how it is invoked by run_bzr.
2284
 
 
2285
 
        Attempts to run bzr from inside this class don't actually run it.
2286
 
 
2287
 
        We test how run_bzr actually invokes bzr in another location.  Here we
2288
 
        only need to test that it passes the right parameters to run_bzr.
2289
 
        """
2290
 
        self.argv = list(argv)
2291
 
        self.retcode = retcode
2292
 
        self.encoding = encoding
2293
 
        self.stdin = stdin
2294
 
        self.working_dir = working_dir
2295
 
        return self.retcode, self.out, self.err
2296
 
 
2297
 
    def test_run_bzr_error(self):
2298
 
        self.out = "It sure does!\n"
2299
 
        out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
2300
 
        self.assertEqual(['rocks'], self.argv)
2301
 
        self.assertEqual(34, self.retcode)
2302
 
        self.assertEqual('It sure does!\n', out)
2303
 
        self.assertEqual(out, self.out)
2304
 
        self.assertEqual('', err)
2305
 
        self.assertEqual(err, self.err)
2306
 
 
2307
 
    def test_run_bzr_error_regexes(self):
2308
 
        self.out = ''
2309
 
        self.err = "bzr: ERROR: foobarbaz is not versioned"
2310
 
        out, err = self.run_bzr_error(
2311
 
            ["bzr: ERROR: foobarbaz is not versioned"],
2312
 
            ['file-id', 'foobarbaz'])
2313
 
 
2314
 
    def test_encoding(self):
2315
 
        """Test that run_bzr passes encoding to _run_bzr_core"""
2316
 
        self.run_bzr('foo bar')
2317
 
        self.assertEqual(None, self.encoding)
2318
 
        self.assertEqual(['foo', 'bar'], self.argv)
2319
 
 
2320
 
        self.run_bzr('foo bar', encoding='baz')
2321
 
        self.assertEqual('baz', self.encoding)
2322
 
        self.assertEqual(['foo', 'bar'], self.argv)
2323
 
 
2324
 
    def test_retcode(self):
2325
 
        """Test that run_bzr passes retcode to _run_bzr_core"""
2326
 
        # Default is retcode == 0
2327
 
        self.run_bzr('foo bar')
2328
 
        self.assertEqual(0, self.retcode)
2329
 
        self.assertEqual(['foo', 'bar'], self.argv)
2330
 
 
2331
 
        self.run_bzr('foo bar', retcode=1)
2332
 
        self.assertEqual(1, self.retcode)
2333
 
        self.assertEqual(['foo', 'bar'], self.argv)
2334
 
 
2335
 
        self.run_bzr('foo bar', retcode=None)
2336
 
        self.assertEqual(None, self.retcode)
2337
 
        self.assertEqual(['foo', 'bar'], self.argv)
2338
 
 
2339
 
        self.run_bzr(['foo', 'bar'], retcode=3)
2340
 
        self.assertEqual(3, self.retcode)
2341
 
        self.assertEqual(['foo', 'bar'], self.argv)
2342
 
 
2343
 
    def test_stdin(self):
2344
 
        # test that the stdin keyword to run_bzr is passed through to
2345
 
        # _run_bzr_core as-is. We do this by overriding
2346
 
        # _run_bzr_core in this class, and then calling run_bzr,
2347
 
        # which is a convenience function for _run_bzr_core, so
2348
 
        # should invoke it.
2349
 
        self.run_bzr('foo bar', stdin='gam')
2350
 
        self.assertEqual('gam', self.stdin)
2351
 
        self.assertEqual(['foo', 'bar'], self.argv)
2352
 
 
2353
 
        self.run_bzr('foo bar', stdin='zippy')
2354
 
        self.assertEqual('zippy', self.stdin)
2355
 
        self.assertEqual(['foo', 'bar'], self.argv)
2356
 
 
2357
 
    def test_working_dir(self):
2358
 
        """Test that run_bzr passes working_dir to _run_bzr_core"""
2359
 
        self.run_bzr('foo bar')
2360
 
        self.assertEqual(None, self.working_dir)
2361
 
        self.assertEqual(['foo', 'bar'], self.argv)
2362
 
 
2363
 
        self.run_bzr('foo bar', working_dir='baz')
2364
 
        self.assertEqual('baz', self.working_dir)
2365
 
        self.assertEqual(['foo', 'bar'], self.argv)
2366
 
 
2367
 
    def test_reject_extra_keyword_arguments(self):
2368
 
        self.assertRaises(TypeError, self.run_bzr, "foo bar",
2369
 
                          error_regex=['error message'])
2370
 
 
2371
 
 
2372
 
class TestRunBzrCaptured(tests.TestCaseWithTransport):
2373
 
    # Does IO when testing the working_dir parameter.
2374
 
 
2375
 
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
2376
 
                         a_callable=None, *args, **kwargs):
2377
 
        self.stdin = stdin
2378
 
        self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
2379
 
        self.factory = bzrlib.ui.ui_factory
2380
 
        self.working_dir = osutils.getcwd()
2381
 
        stdout.write('foo\n')
2382
 
        stderr.write('bar\n')
2383
 
        return 0
2384
 
 
2385
 
    def test_stdin(self):
2386
 
        # test that the stdin keyword to _run_bzr_core is passed through to
2387
 
        # apply_redirected as a StringIO. We do this by overriding
2388
 
        # apply_redirected in this class, and then calling _run_bzr_core,
2389
 
        # which calls apply_redirected.
2390
 
        self.run_bzr(['foo', 'bar'], stdin='gam')
2391
 
        self.assertEqual('gam', self.stdin.read())
2392
 
        self.assertTrue(self.stdin is self.factory_stdin)
2393
 
        self.run_bzr(['foo', 'bar'], stdin='zippy')
2394
 
        self.assertEqual('zippy', self.stdin.read())
2395
 
        self.assertTrue(self.stdin is self.factory_stdin)
2396
 
 
2397
 
    def test_ui_factory(self):
2398
 
        # each invocation of self.run_bzr should get its
2399
 
        # own UI factory, which is an instance of TestUIFactory,
2400
 
        # with stdin, stdout and stderr attached to the stdin,
2401
 
        # stdout and stderr of the invoked run_bzr
2402
 
        current_factory = bzrlib.ui.ui_factory
2403
 
        self.run_bzr(['foo'])
2404
 
        self.assertFalse(current_factory is self.factory)
2405
 
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2406
 
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2407
 
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2408
 
        self.assertEqual('bar\n', self.factory.stderr.getvalue())
2409
 
        self.assertIsInstance(self.factory, tests.TestUIFactory)
2410
 
 
2411
 
    def test_working_dir(self):
2412
 
        self.build_tree(['one/', 'two/'])
2413
 
        cwd = osutils.getcwd()
2414
 
 
2415
 
        # Default is to work in the current directory
2416
 
        self.run_bzr(['foo', 'bar'])
2417
 
        self.assertEqual(cwd, self.working_dir)
2418
 
 
2419
 
        self.run_bzr(['foo', 'bar'], working_dir=None)
2420
 
        self.assertEqual(cwd, self.working_dir)
2421
 
 
2422
 
        # The function should be run in the alternative directory
2423
 
        # but afterwards the current working dir shouldn't be changed
2424
 
        self.run_bzr(['foo', 'bar'], working_dir='one')
2425
 
        self.assertNotEqual(cwd, self.working_dir)
2426
 
        self.assertEndsWith(self.working_dir, 'one')
2427
 
        self.assertEqual(cwd, osutils.getcwd())
2428
 
 
2429
 
        self.run_bzr(['foo', 'bar'], working_dir='two')
2430
 
        self.assertNotEqual(cwd, self.working_dir)
2431
 
        self.assertEndsWith(self.working_dir, 'two')
2432
 
        self.assertEqual(cwd, osutils.getcwd())
2433
 
 
2434
 
 
2435
 
class StubProcess(object):
2436
 
    """A stub process for testing run_bzr_subprocess."""
2437
 
    
2438
 
    def __init__(self, out="", err="", retcode=0):
2439
 
        self.out = out
2440
 
        self.err = err
2441
 
        self.returncode = retcode
2442
 
 
2443
 
    def communicate(self):
2444
 
        return self.out, self.err
2445
 
 
2446
 
 
2447
 
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
2448
 
    """Base class for tests testing how we might run bzr."""
2449
 
 
2450
 
    def setUp(self):
2451
 
        super(TestWithFakedStartBzrSubprocess, self).setUp()
2452
 
        self.subprocess_calls = []
2453
 
 
2454
 
    def start_bzr_subprocess(self, process_args, env_changes=None,
2455
 
                             skip_if_plan_to_signal=False,
2456
 
                             working_dir=None,
2457
 
                             allow_plugins=False):
2458
 
        """capture what run_bzr_subprocess tries to do."""
2459
 
        self.subprocess_calls.append({'process_args':process_args,
2460
 
            'env_changes':env_changes,
2461
 
            'skip_if_plan_to_signal':skip_if_plan_to_signal,
2462
 
            'working_dir':working_dir, 'allow_plugins':allow_plugins})
2463
 
        return self.next_subprocess
2464
 
 
2465
 
 
2466
 
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
2467
 
 
2468
 
    def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
2469
 
        """Run run_bzr_subprocess with args and kwargs using a stubbed process.
2470
 
 
2471
 
        Inside TestRunBzrSubprocessCommands we use a stub start_bzr_subprocess
2472
 
        that will return static results. This assertion method populates those
2473
 
        results and also checks the arguments run_bzr_subprocess generates.
2474
 
        """
2475
 
        self.next_subprocess = process
2476
 
        try:
2477
 
            result = self.run_bzr_subprocess(*args, **kwargs)
2478
 
        except:
2479
 
            self.next_subprocess = None
2480
 
            for key, expected in expected_args.iteritems():
2481
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2482
 
            raise
 
1695
 
 
1696
class TestKnownFailure(TestCase):
 
1697
 
 
1698
    def test_known_failure(self):
 
1699
        """Check that KnownFailure is defined appropriately."""
 
1700
        # a KnownFailure is an assertion error for compatability with unaware
 
1701
        # runners.
 
1702
        self.assertIsInstance(KnownFailure(""), AssertionError)
 
1703
 
 
1704
    def test_expect_failure(self):
 
1705
        try:
 
1706
            self.expectFailure("Doomed to failure", self.assertTrue, False)
 
1707
        except KnownFailure, e:
 
1708
            self.assertEqual('Doomed to failure', e.args[0])
 
1709
        try:
 
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])
2483
1714
        else:
2484
 
            self.next_subprocess = None
2485
 
            for key, expected in expected_args.iteritems():
2486
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2487
 
            return result
2488
 
 
2489
 
    def test_run_bzr_subprocess(self):
2490
 
        """The run_bzr_helper_external command behaves nicely."""
2491
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2492
 
            StubProcess(), '--version')
2493
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2494
 
            StubProcess(), ['--version'])
2495
 
        # retcode=None disables retcode checking
2496
 
        result = self.assertRunBzrSubprocess({},
2497
 
            StubProcess(retcode=3), '--version', retcode=None)
2498
 
        result = self.assertRunBzrSubprocess({},
2499
 
            StubProcess(out="is free software"), '--version')
2500
 
        self.assertContainsRe(result[0], 'is free software')
2501
 
        # Running a subcommand that is missing errors
2502
 
        self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
2503
 
            {'process_args':['--versionn']}, StubProcess(retcode=3),
2504
 
            '--versionn')
2505
 
        # Unless it is told to expect the error from the subprocess
2506
 
        result = self.assertRunBzrSubprocess({},
2507
 
            StubProcess(retcode=3), '--versionn', retcode=3)
2508
 
        # Or to ignore retcode checking
2509
 
        result = self.assertRunBzrSubprocess({},
2510
 
            StubProcess(err="unknown command", retcode=3), '--versionn',
2511
 
            retcode=None)
2512
 
        self.assertContainsRe(result[1], 'unknown command')
2513
 
 
2514
 
    def test_env_change_passes_through(self):
2515
 
        self.assertRunBzrSubprocess(
2516
 
            {'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
2517
 
            StubProcess(), '',
2518
 
            env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
2519
 
 
2520
 
    def test_no_working_dir_passed_as_None(self):
2521
 
        self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
2522
 
 
2523
 
    def test_no_working_dir_passed_through(self):
2524
 
        self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
2525
 
            working_dir='dir')
2526
 
 
2527
 
    def test_run_bzr_subprocess_no_plugins(self):
2528
 
        self.assertRunBzrSubprocess({'allow_plugins': False},
2529
 
            StubProcess(), '')
2530
 
 
2531
 
    def test_allow_plugins(self):
2532
 
        self.assertRunBzrSubprocess({'allow_plugins': True},
2533
 
            StubProcess(), '', allow_plugins=True)
2534
 
 
2535
 
 
2536
 
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2537
 
 
2538
 
    def test_finish_bzr_subprocess_with_error(self):
2539
 
        """finish_bzr_subprocess allows specification of the desired exit code.
2540
 
        """
2541
 
        process = StubProcess(err="unknown command", retcode=3)
2542
 
        result = self.finish_bzr_subprocess(process, retcode=3)
2543
 
        self.assertEqual('', result[0])
2544
 
        self.assertContainsRe(result[1], 'unknown command')
2545
 
 
2546
 
    def test_finish_bzr_subprocess_ignoring_retcode(self):
2547
 
        """finish_bzr_subprocess allows the exit code to be ignored."""
2548
 
        process = StubProcess(err="unknown command", retcode=3)
2549
 
        result = self.finish_bzr_subprocess(process, retcode=None)
2550
 
        self.assertEqual('', result[0])
2551
 
        self.assertContainsRe(result[1], 'unknown command')
2552
 
 
2553
 
    def test_finish_subprocess_with_unexpected_retcode(self):
2554
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2555
 
        not the expected one.
2556
 
        """
2557
 
        process = StubProcess(err="unknown command", retcode=3)
2558
 
        self.assertRaises(self.failureException, self.finish_bzr_subprocess,
2559
 
                          process)
2560
 
 
2561
 
 
2562
 
class _DontSpawnProcess(Exception):
2563
 
    """A simple exception which just allows us to skip unnecessary steps"""
2564
 
 
2565
 
 
2566
 
class TestStartBzrSubProcess(tests.TestCase):
2567
 
    """Stub test start_bzr_subprocess."""
2568
 
 
2569
 
    def _subprocess_log_cleanup(self):
2570
 
        """Inhibits the base version as we don't produce a log file."""
2571
 
 
2572
 
    def _popen(self, *args, **kwargs):
2573
 
        """Override the base version to record the command that is run.
2574
 
 
2575
 
        From there we can ensure it is correct without spawning a real process.
2576
 
        """
2577
 
        self.check_popen_state()
2578
 
        self._popen_args = args
2579
 
        self._popen_kwargs = kwargs
2580
 
        raise _DontSpawnProcess()
2581
 
 
2582
 
    def check_popen_state(self):
2583
 
        """Replace to make assertions when popen is called."""
2584
 
 
2585
 
    def test_run_bzr_subprocess_no_plugins(self):
2586
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2587
 
        command = self._popen_args[0]
2588
 
        self.assertEqual(sys.executable, command[0])
2589
 
        self.assertEqual(self.get_bzr_path(), command[1])
2590
 
        self.assertEqual(['--no-plugins'], command[2:])
2591
 
 
2592
 
    def test_allow_plugins(self):
2593
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2594
 
                          allow_plugins=True)
2595
 
        command = self._popen_args[0]
2596
 
        self.assertEqual([], command[2:])
2597
 
 
2598
 
    def test_set_env(self):
2599
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2600
 
        # set in the child
2601
 
        def check_environment():
2602
 
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2603
 
        self.check_popen_state = check_environment
2604
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2605
 
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
2606
 
        # not set in theparent
2607
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2608
 
 
2609
 
    def test_run_bzr_subprocess_env_del(self):
2610
 
        """run_bzr_subprocess can remove environment variables too."""
2611
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2612
 
        def check_environment():
2613
 
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2614
 
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2615
 
        self.check_popen_state = check_environment
2616
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2617
 
                          env_changes={'EXISTANT_ENV_VAR':None})
2618
 
        # Still set in parent
2619
 
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2620
 
        del os.environ['EXISTANT_ENV_VAR']
2621
 
 
2622
 
    def test_env_del_missing(self):
2623
 
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2624
 
        def check_environment():
2625
 
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2626
 
        self.check_popen_state = check_environment
2627
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2628
 
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
2629
 
 
2630
 
    def test_working_dir(self):
2631
 
        """Test that we can specify the working dir for the child"""
2632
 
        orig_getcwd = osutils.getcwd
2633
 
        orig_chdir = os.chdir
2634
 
        chdirs = []
2635
 
        def chdir(path):
2636
 
            chdirs.append(path)
2637
 
        self.overrideAttr(os, 'chdir', chdir)
2638
 
        def getcwd():
2639
 
            return 'current'
2640
 
        self.overrideAttr(osutils, 'getcwd', getcwd)
2641
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2642
 
                          working_dir='foo')
2643
 
        self.assertEqual(['foo', 'current'], chdirs)
2644
 
 
2645
 
    def test_get_bzr_path_with_cwd_bzrlib(self):
2646
 
        self.get_source_path = lambda: ""
2647
 
        self.overrideAttr(os.path, "isfile", lambda path: True)
2648
 
        self.assertEqual(self.get_bzr_path(), "bzr")
2649
 
 
2650
 
 
2651
 
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2652
 
    """Tests that really need to do things with an external bzr."""
2653
 
 
2654
 
    def test_start_and_stop_bzr_subprocess_send_signal(self):
2655
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2656
 
        not the expected one.
2657
 
        """
2658
 
        self.disable_missing_extensions_warning()
2659
 
        process = self.start_bzr_subprocess(['wait-until-signalled'],
2660
 
                                            skip_if_plan_to_signal=True)
2661
 
        self.assertEqual('running\n', process.stdout.readline())
2662
 
        result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
2663
 
                                            retcode=3)
2664
 
        self.assertEqual('', result[0])
2665
 
        self.assertEqual('bzr: interrupted\n', result[1])
2666
 
 
2667
 
 
2668
 
class TestSelftestFiltering(tests.TestCase):
 
1715
            self.fail('Assertion not raised')
 
1716
 
 
1717
 
 
1718
class TestFeature(TestCase):
 
1719
 
 
1720
    def test_caching(self):
 
1721
        """Feature._probe is called by the feature at most once."""
 
1722
        class InstrumentedFeature(Feature):
 
1723
            def __init__(self):
 
1724
                Feature.__init__(self)
 
1725
                self.calls = []
 
1726
            def _probe(self):
 
1727
                self.calls.append('_probe')
 
1728
                return False
 
1729
        feature = InstrumentedFeature()
 
1730
        feature.available()
 
1731
        self.assertEqual(['_probe'], feature.calls)
 
1732
        feature.available()
 
1733
        self.assertEqual(['_probe'], feature.calls)
 
1734
 
 
1735
    def test_named_str(self):
 
1736
        """Feature.__str__ should thunk to feature_name()."""
 
1737
        class NamedFeature(Feature):
 
1738
            def feature_name(self):
 
1739
                return 'symlinks'
 
1740
        feature = NamedFeature()
 
1741
        self.assertEqual('symlinks', str(feature))
 
1742
 
 
1743
    def test_default_str(self):
 
1744
        """Feature.__str__ should default to __class__.__name__."""
 
1745
        class NamedFeature(Feature):
 
1746
            pass
 
1747
        feature = NamedFeature()
 
1748
        self.assertEqual('NamedFeature', str(feature))
 
1749
 
 
1750
 
 
1751
class TestUnavailableFeature(TestCase):
 
1752
 
 
1753
    def test_access_feature(self):
 
1754
        feature = Feature()
 
1755
        exception = UnavailableFeature(feature)
 
1756
        self.assertIs(feature, exception.args[0])
 
1757
 
 
1758
 
 
1759
class TestSelftestFiltering(TestCase):
2669
1760
 
2670
1761
    def setUp(self):
2671
 
        super(TestSelftestFiltering, self).setUp()
2672
1762
        self.suite = TestUtil.TestSuite()
2673
1763
        self.loader = TestUtil.TestLoader()
2674
 
        self.suite.addTest(self.loader.loadTestsFromModule(
2675
 
            sys.modules['bzrlib.tests.test_selftest']))
 
1764
        self.suite.addTest(self.loader.loadTestsFromModuleNames([
 
1765
            'bzrlib.tests.test_selftest']))
2676
1766
        self.all_names = _test_ids(self.suite)
2677
1767
 
2678
1768
    def test_condition_id_re(self):
2679
1769
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2680
1770
            'test_condition_id_re')
2681
 
        filtered_suite = tests.filter_suite_by_condition(
2682
 
            self.suite, tests.condition_id_re('test_condition_id_re'))
 
1771
        filtered_suite = filter_suite_by_condition(self.suite,
 
1772
            condition_id_re('test_condition_id_re'))
2683
1773
        self.assertEqual([test_name], _test_ids(filtered_suite))
2684
1774
 
2685
1775
    def test_condition_id_in_list(self):
2686
1776
        test_names = ['bzrlib.tests.test_selftest.TestSelftestFiltering.'
2687
1777
                      'test_condition_id_in_list']
2688
1778
        id_list = tests.TestIdList(test_names)
2689
 
        filtered_suite = tests.filter_suite_by_condition(
 
1779
        filtered_suite = filter_suite_by_condition(
2690
1780
            self.suite, tests.condition_id_in_list(id_list))
2691
1781
        my_pattern = 'TestSelftestFiltering.*test_condition_id_in_list'
2692
 
        re_filtered = tests.filter_suite_by_re(self.suite, my_pattern)
 
1782
        re_filtered = filter_suite_by_re(self.suite, my_pattern)
2693
1783
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
2694
1784
 
2695
1785
    def test_condition_id_startswith(self):
2699
1789
        test_names = [ klass + 'test_condition_id_in_list',
2700
1790
                      klass + 'test_condition_id_startswith',
2701
1791
                     ]
2702
 
        filtered_suite = tests.filter_suite_by_condition(
 
1792
        filtered_suite = filter_suite_by_condition(
2703
1793
            self.suite, tests.condition_id_startswith([start1, start2]))
2704
1794
        self.assertEqual(test_names, _test_ids(filtered_suite))
2705
1795
 
2706
1796
    def test_condition_isinstance(self):
2707
 
        filtered_suite = tests.filter_suite_by_condition(
2708
 
            self.suite, tests.condition_isinstance(self.__class__))
 
1797
        filtered_suite = filter_suite_by_condition(self.suite,
 
1798
            condition_isinstance(self.__class__))
2709
1799
        class_pattern = 'bzrlib.tests.test_selftest.TestSelftestFiltering.'
2710
 
        re_filtered = tests.filter_suite_by_re(self.suite, class_pattern)
 
1800
        re_filtered = filter_suite_by_re(self.suite, class_pattern)
2711
1801
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
2712
1802
 
2713
1803
    def test_exclude_tests_by_condition(self):
2714
1804
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2715
1805
            'test_exclude_tests_by_condition')
2716
 
        filtered_suite = tests.exclude_tests_by_condition(self.suite,
 
1806
        filtered_suite = exclude_tests_by_condition(self.suite,
2717
1807
            lambda x:x.id() == excluded_name)
2718
1808
        self.assertEqual(len(self.all_names) - 1,
2719
1809
            filtered_suite.countTestCases())
2724
1814
 
2725
1815
    def test_exclude_tests_by_re(self):
2726
1816
        self.all_names = _test_ids(self.suite)
2727
 
        filtered_suite = tests.exclude_tests_by_re(self.suite,
2728
 
                                                   'exclude_tests_by_re')
 
1817
        filtered_suite = exclude_tests_by_re(self.suite, 'exclude_tests_by_re')
2729
1818
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2730
1819
            'test_exclude_tests_by_re')
2731
1820
        self.assertEqual(len(self.all_names) - 1,
2738
1827
    def test_filter_suite_by_condition(self):
2739
1828
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2740
1829
            'test_filter_suite_by_condition')
2741
 
        filtered_suite = tests.filter_suite_by_condition(self.suite,
 
1830
        filtered_suite = filter_suite_by_condition(self.suite,
2742
1831
            lambda x:x.id() == test_name)
2743
1832
        self.assertEqual([test_name], _test_ids(filtered_suite))
2744
1833
 
2745
1834
    def test_filter_suite_by_re(self):
2746
 
        filtered_suite = tests.filter_suite_by_re(self.suite,
2747
 
                                                  'test_filter_suite_by_r')
 
1835
        filtered_suite = filter_suite_by_re(self.suite, 'test_filter_suite_by_r')
2748
1836
        filtered_names = _test_ids(filtered_suite)
2749
1837
        self.assertEqual(filtered_names, ['bzrlib.tests.test_selftest.'
2750
1838
            'TestSelftestFiltering.test_filter_suite_by_re'])
2778
1866
 
2779
1867
    def test_preserve_input(self):
2780
1868
        # NB: Surely this is something in the stdlib to do this?
2781
 
        self.assertTrue(self.suite is tests.preserve_input(self.suite))
2782
 
        self.assertTrue("@#$" is tests.preserve_input("@#$"))
 
1869
        self.assertTrue(self.suite is preserve_input(self.suite))
 
1870
        self.assertTrue("@#$" is preserve_input("@#$"))
2783
1871
 
2784
1872
    def test_randomize_suite(self):
2785
 
        randomized_suite = tests.randomize_suite(self.suite)
 
1873
        randomized_suite = randomize_suite(self.suite)
2786
1874
        # randomizing should not add or remove test names.
2787
1875
        self.assertEqual(set(_test_ids(self.suite)),
2788
1876
                         set(_test_ids(randomized_suite)))
2798
1886
 
2799
1887
    def test_split_suit_by_condition(self):
2800
1888
        self.all_names = _test_ids(self.suite)
2801
 
        condition = tests.condition_id_re('test_filter_suite_by_r')
2802
 
        split_suite = tests.split_suite_by_condition(self.suite, condition)
 
1889
        condition = condition_id_re('test_filter_suite_by_r')
 
1890
        split_suite = split_suite_by_condition(self.suite, condition)
2803
1891
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2804
1892
            'test_filter_suite_by_re')
2805
1893
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
2810
1898
 
2811
1899
    def test_split_suit_by_re(self):
2812
1900
        self.all_names = _test_ids(self.suite)
2813
 
        split_suite = tests.split_suite_by_re(self.suite,
2814
 
                                              'test_filter_suite_by_r')
 
1901
        split_suite = split_suite_by_re(self.suite, 'test_filter_suite_by_r')
2815
1902
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2816
1903
            'test_filter_suite_by_re')
2817
1904
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
2821
1908
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2822
1909
 
2823
1910
 
2824
 
class TestCheckTreeShape(tests.TestCaseWithTransport):
 
1911
class TestCheckInventoryShape(TestCaseWithTransport):
2825
1912
 
2826
 
    def test_check_tree_shape(self):
 
1913
    def test_check_inventory_shape(self):
2827
1914
        files = ['a', 'b/', 'b/c']
2828
1915
        tree = self.make_branch_and_tree('.')
2829
1916
        self.build_tree(files)
2830
1917
        tree.add(files)
2831
1918
        tree.lock_read()
2832
1919
        try:
2833
 
            self.check_tree_shape(tree, files)
 
1920
            self.check_inventory_shape(tree.inventory, files)
2834
1921
        finally:
2835
1922
            tree.unlock()
2836
1923
 
2837
1924
 
2838
 
class TestBlackboxSupport(tests.TestCase):
 
1925
class TestBlackboxSupport(TestCase):
2839
1926
    """Tests for testsuite blackbox features."""
2840
1927
 
2841
1928
    def test_run_bzr_failure_not_caught(self):
2847
1934
            self.run_bzr, ['assert-fail'])
2848
1935
        # make sure we got the real thing, not an error from somewhere else in
2849
1936
        # the test framework
2850
 
        self.assertEqual('always fails', str(e))
 
1937
        self.assertEquals('always fails', str(e))
2851
1938
        # check that there's no traceback in the test log
2852
 
        self.assertNotContainsRe(self.get_log(), r'Traceback')
 
1939
        self.assertNotContainsRe(self._get_log(keep_log_file=True),
 
1940
            r'Traceback')
2853
1941
 
2854
1942
    def test_run_bzr_user_error_caught(self):
2855
1943
        # Running bzr in blackbox mode, normal/expected/user errors should be
2856
1944
        # caught in the regular way and turned into an error message plus exit
2857
1945
        # code.
2858
 
        transport_server = memory.MemoryServer()
2859
 
        transport_server.start_server()
2860
 
        self.addCleanup(transport_server.stop_server)
2861
 
        url = transport_server.get_url()
2862
 
        self.permit_url(url)
2863
 
        out, err = self.run_bzr(["log", "%s/nonexistantpath" % url], retcode=3)
 
1946
        out, err = self.run_bzr(["log", "/nonexistantpath"], retcode=3)
2864
1947
        self.assertEqual(out, '')
2865
1948
        self.assertContainsRe(err,
2866
1949
            'bzr: ERROR: Not a branch: ".*nonexistantpath/".\n')
2867
1950
 
2868
1951
 
2869
 
class TestTestLoader(tests.TestCase):
 
1952
class TestTestLoader(TestCase):
2870
1953
    """Tests for the test loader."""
2871
1954
 
2872
1955
    def _get_loader_and_module(self):
2873
1956
        """Gets a TestLoader and a module with one test in it."""
2874
1957
        loader = TestUtil.TestLoader()
2875
1958
        module = {}
2876
 
        class Stub(tests.TestCase):
 
1959
        class Stub(TestCase):
2877
1960
            def test_foo(self):
2878
1961
                pass
2879
1962
        class MyModule(object):
2892
1975
        # load_tests do not need that :)
2893
1976
        def load_tests(self, standard_tests, module, loader):
2894
1977
            result = loader.suiteClass()
2895
 
            for test in tests.iter_suite_tests(standard_tests):
 
1978
            for test in iter_suite_tests(standard_tests):
2896
1979
                result.addTests([test, test])
2897
1980
            return result
2898
1981
        # add a load_tests() method which multiplies the tests from the module.
2902
1985
    def test_load_tests_from_module_name_smoke_test(self):
2903
1986
        loader = TestUtil.TestLoader()
2904
1987
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2905
 
        self.assertEqual(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
 
1988
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
2906
1989
                          _test_ids(suite))
2907
1990
 
2908
1991
    def test_load_tests_from_module_name_with_bogus_module_name(self):
2917
2000
 
2918
2001
    def _create_suite(self, test_id_list):
2919
2002
 
2920
 
        class Stub(tests.TestCase):
 
2003
        class Stub(TestCase):
2921
2004
            def test_foo(self):
2922
2005
                pass
2923
2006
 
2933
2016
 
2934
2017
    def _test_ids(self, test_suite):
2935
2018
        """Get the ids for the tests in a test suite."""
2936
 
        return [t.id() for t in tests.iter_suite_tests(test_suite)]
 
2019
        return [t.id() for t in iter_suite_tests(test_suite)]
2937
2020
 
2938
2021
    def test_empty_list(self):
2939
2022
        id_list = self._create_id_list([])
2940
 
        self.assertEqual({}, id_list.tests)
2941
 
        self.assertEqual({}, id_list.modules)
 
2023
        self.assertEquals({}, id_list.tests)
 
2024
        self.assertEquals({}, id_list.modules)
2942
2025
 
2943
2026
    def test_valid_list(self):
2944
2027
        id_list = self._create_id_list(
2965
2048
        self.assertTrue(id_list.refers_to('mod.class'))
2966
2049
        self.assertTrue(id_list.refers_to('mod.class.meth'))
2967
2050
 
 
2051
    def test_test_suite(self):
 
2052
        # This test is slow, so we do a single test with one test in each
 
2053
        # category
 
2054
        test_list = [
 
2055
            # testmod_names
 
2056
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
 
2057
            'bzrlib.tests.test_selftest.TestTestIdList.test_test_suite',
 
2058
            # transport implementations
 
2059
            'bzrlib.tests.test_transport_implementations.TransportTests'
 
2060
            '.test_abspath(LocalURLServer)',
 
2061
            # modules_to_doctest
 
2062
            'bzrlib.timestamp.format_highres_date',
 
2063
            # plugins can't be tested that way since selftest may be run with
 
2064
            # --no-plugins
 
2065
            ]
 
2066
        suite = tests.test_suite(test_list)
 
2067
        self.assertEquals(test_list, _test_ids(suite))
 
2068
 
2968
2069
    def test_test_suite_matches_id_list_with_unknown(self):
2969
2070
        loader = TestUtil.TestLoader()
2970
2071
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2971
2072
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',
2972
2073
                     'bogus']
2973
2074
        not_found, duplicates = tests.suite_matches_id_list(suite, test_list)
2974
 
        self.assertEqual(['bogus'], not_found)
2975
 
        self.assertEqual([], duplicates)
 
2075
        self.assertEquals(['bogus'], not_found)
 
2076
        self.assertEquals([], duplicates)
2976
2077
 
2977
2078
    def test_suite_matches_id_list_with_duplicates(self):
2978
2079
        loader = TestUtil.TestLoader()
2979
2080
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2980
2081
        dupes = loader.suiteClass()
2981
 
        for test in tests.iter_suite_tests(suite):
 
2082
        for test in iter_suite_tests(suite):
2982
2083
            dupes.addTest(test)
2983
2084
            dupes.addTest(test) # Add it again
2984
2085
 
2985
2086
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',]
2986
2087
        not_found, duplicates = tests.suite_matches_id_list(
2987
2088
            dupes, test_list)
2988
 
        self.assertEqual([], not_found)
2989
 
        self.assertEqual(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
 
2089
        self.assertEquals([], not_found)
 
2090
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
2990
2091
                          duplicates)
2991
2092
 
2992
2093
 
2993
 
class TestTestSuite(tests.TestCase):
2994
 
 
2995
 
    def test__test_suite_testmod_names(self):
2996
 
        # Test that a plausible list of test module names are returned
2997
 
        # by _test_suite_testmod_names.
2998
 
        test_list = tests._test_suite_testmod_names()
2999
 
        self.assertSubset([
3000
 
            'bzrlib.tests.blackbox',
3001
 
            'bzrlib.tests.per_transport',
3002
 
            'bzrlib.tests.test_selftest',
3003
 
            ],
3004
 
            test_list)
3005
 
 
3006
 
    def test__test_suite_modules_to_doctest(self):
3007
 
        # Test that a plausible list of modules to doctest is returned
3008
 
        # by _test_suite_modules_to_doctest.
3009
 
        test_list = tests._test_suite_modules_to_doctest()
3010
 
        if __doc__ is None:
3011
 
            # When docstrings are stripped, there are no modules to doctest
3012
 
            self.assertEqual([], test_list)
3013
 
            return
3014
 
        self.assertSubset([
3015
 
            'bzrlib.timestamp',
3016
 
            ],
3017
 
            test_list)
3018
 
 
3019
 
    def test_test_suite(self):
3020
 
        # test_suite() loads the entire test suite to operate. To avoid this
3021
 
        # overhead, and yet still be confident that things are happening,
3022
 
        # we temporarily replace two functions used by test_suite with 
3023
 
        # test doubles that supply a few sample tests to load, and check they
3024
 
        # are loaded.
3025
 
        calls = []
3026
 
        def testmod_names():
3027
 
            calls.append("testmod_names")
3028
 
            return [
3029
 
                'bzrlib.tests.blackbox.test_branch',
3030
 
                'bzrlib.tests.per_transport',
3031
 
                'bzrlib.tests.test_selftest',
3032
 
                ]
3033
 
        self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
3034
 
        def doctests():
3035
 
            calls.append("modules_to_doctest")
3036
 
            if __doc__ is None:
3037
 
                return []
3038
 
            return ['bzrlib.timestamp']
3039
 
        self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
3040
 
        expected_test_list = [
3041
 
            # testmod_names
3042
 
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
3043
 
            ('bzrlib.tests.per_transport.TransportTests'
3044
 
             '.test_abspath(LocalTransport,LocalURLServer)'),
3045
 
            'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
3046
 
            # plugins can't be tested that way since selftest may be run with
3047
 
            # --no-plugins
3048
 
            ]
3049
 
        if __doc__ is not None:
3050
 
            expected_test_list.extend([
3051
 
                # modules_to_doctest
3052
 
                'bzrlib.timestamp.format_highres_date',
3053
 
                ])
3054
 
        suite = tests.test_suite()
3055
 
        self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
3056
 
            set(calls))
3057
 
        self.assertSubset(expected_test_list, _test_ids(suite))
3058
 
 
3059
 
    def test_test_suite_list_and_start(self):
3060
 
        # We cannot test this at the same time as the main load, because we want
3061
 
        # to know that starting_with == None works. So a second load is
3062
 
        # incurred - note that the starting_with parameter causes a partial load
3063
 
        # rather than a full load so this test should be pretty quick.
3064
 
        test_list = ['bzrlib.tests.test_selftest.TestTestSuite.test_test_suite']
3065
 
        suite = tests.test_suite(test_list,
3066
 
                                 ['bzrlib.tests.test_selftest.TestTestSuite'])
3067
 
        # test_test_suite_list_and_start is not included 
3068
 
        self.assertEqual(test_list, _test_ids(suite))
3069
 
 
3070
 
 
3071
2094
class TestLoadTestIdList(tests.TestCaseInTempDir):
3072
2095
 
3073
2096
    def _create_test_list_file(self, file_name, content):
3084
2107
        self._create_test_list_file(test_list_fname,
3085
2108
                                    'mod1.cl1.meth1\nmod2.cl2.meth2\n')
3086
2109
        tlist = tests.load_test_id_list(test_list_fname)
3087
 
        self.assertEqual(2, len(tlist))
3088
 
        self.assertEqual('mod1.cl1.meth1', tlist[0])
3089
 
        self.assertEqual('mod2.cl2.meth2', tlist[1])
 
2110
        self.assertEquals(2, len(tlist))
 
2111
        self.assertEquals('mod1.cl1.meth1', tlist[0])
 
2112
        self.assertEquals('mod2.cl2.meth2', tlist[1])
3090
2113
 
3091
2114
    def test_load_dirty_file(self):
3092
2115
        test_list_fname = 'test.list'
3094
2117
                                    '  mod1.cl1.meth1\n\nmod2.cl2.meth2  \n'
3095
2118
                                    'bar baz\n')
3096
2119
        tlist = tests.load_test_id_list(test_list_fname)
3097
 
        self.assertEqual(4, len(tlist))
3098
 
        self.assertEqual('mod1.cl1.meth1', tlist[0])
3099
 
        self.assertEqual('', tlist[1])
3100
 
        self.assertEqual('mod2.cl2.meth2', tlist[2])
3101
 
        self.assertEqual('bar baz', tlist[3])
 
2120
        self.assertEquals(4, len(tlist))
 
2121
        self.assertEquals('mod1.cl1.meth1', tlist[0])
 
2122
        self.assertEquals('', tlist[1])
 
2123
        self.assertEquals('mod2.cl2.meth2', tlist[2])
 
2124
        self.assertEquals('bar baz', tlist[3])
3102
2125
 
3103
2126
 
3104
2127
class TestFilteredByModuleTestLoader(tests.TestCase):
3111
2134
    def test_load_tests(self):
3112
2135
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3113
2136
        loader = self._create_loader(test_list)
 
2137
 
3114
2138
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3115
 
        self.assertEqual(test_list, _test_ids(suite))
 
2139
        self.assertEquals(test_list, _test_ids(suite))
3116
2140
 
3117
2141
    def test_exclude_tests(self):
3118
2142
        test_list = ['bogus']
3119
2143
        loader = self._create_loader(test_list)
 
2144
 
3120
2145
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3121
 
        self.assertEqual([], _test_ids(suite))
 
2146
        self.assertEquals([], _test_ids(suite))
3122
2147
 
3123
2148
 
3124
2149
class TestFilteredByNameStartTestLoader(tests.TestCase):
3134
2159
        loader = self._create_loader('bzrlib.tests.test_samp')
3135
2160
 
3136
2161
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3137
 
        self.assertEqual(test_list, _test_ids(suite))
 
2162
        self.assertEquals(test_list, _test_ids(suite))
3138
2163
 
3139
2164
    def test_load_tests_inside_module(self):
3140
2165
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3141
2166
        loader = self._create_loader('bzrlib.tests.test_sampler.Demo')
3142
2167
 
3143
2168
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3144
 
        self.assertEqual(test_list, _test_ids(suite))
 
2169
        self.assertEquals(test_list, _test_ids(suite))
3145
2170
 
3146
2171
    def test_exclude_tests(self):
3147
2172
        test_list = ['bogus']
3148
2173
        loader = self._create_loader('bogus')
3149
2174
 
3150
2175
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3151
 
        self.assertEqual([], _test_ids(suite))
 
2176
        self.assertEquals([], _test_ids(suite))
3152
2177
 
3153
2178
 
3154
2179
class TestTestPrefixRegistry(tests.TestCase):
3160
2185
    def test_register_new_prefix(self):
3161
2186
        tpr = self._get_registry()
3162
2187
        tpr.register('foo', 'fff.ooo.ooo')
3163
 
        self.assertEqual('fff.ooo.ooo', tpr.get('foo'))
 
2188
        self.assertEquals('fff.ooo.ooo', tpr.get('foo'))
3164
2189
 
3165
2190
    def test_register_existing_prefix(self):
3166
2191
        tpr = self._get_registry()
3167
2192
        tpr.register('bar', 'bbb.aaa.rrr')
3168
2193
        tpr.register('bar', 'bBB.aAA.rRR')
3169
 
        self.assertEqual('bbb.aaa.rrr', tpr.get('bar'))
3170
 
        self.assertThat(self.get_log(),
3171
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
3172
 
                           doctest.ELLIPSIS))
 
2194
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
 
2195
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
2196
                              r'.*bar.*bbb.aaa.rrr.*bBB.aAA.rRR')
3173
2197
 
3174
2198
    def test_get_unknown_prefix(self):
3175
2199
        tpr = self._get_registry()
3178
2202
    def test_resolve_prefix(self):
3179
2203
        tpr = self._get_registry()
3180
2204
        tpr.register('bar', 'bb.aa.rr')
3181
 
        self.assertEqual('bb.aa.rr', tpr.resolve_alias('bar'))
 
2205
        self.assertEquals('bb.aa.rr', tpr.resolve_alias('bar'))
3182
2206
 
3183
2207
    def test_resolve_unknown_alias(self):
3184
2208
        tpr = self._get_registry()
3187
2211
 
3188
2212
    def test_predefined_prefixes(self):
3189
2213
        tpr = tests.test_prefix_alias_registry
3190
 
        self.assertEqual('bzrlib', tpr.resolve_alias('bzrlib'))
3191
 
        self.assertEqual('bzrlib.doc', tpr.resolve_alias('bd'))
3192
 
        self.assertEqual('bzrlib.utils', tpr.resolve_alias('bu'))
3193
 
        self.assertEqual('bzrlib.tests', tpr.resolve_alias('bt'))
3194
 
        self.assertEqual('bzrlib.tests.blackbox', tpr.resolve_alias('bb'))
3195
 
        self.assertEqual('bzrlib.plugins', tpr.resolve_alias('bp'))
3196
 
 
3197
 
 
3198
 
class TestThreadLeakDetection(tests.TestCase):
3199
 
    """Ensure when tests leak threads we detect and report it"""
3200
 
 
3201
 
    class LeakRecordingResult(tests.ExtendedTestResult):
3202
 
        def __init__(self):
3203
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3204
 
            self.leaks = []
3205
 
        def _report_thread_leak(self, test, leaks, alive):
3206
 
            self.leaks.append((test, leaks))
3207
 
 
3208
 
    def test_testcase_without_addCleanups(self):
3209
 
        """Check old TestCase instances don't break with leak detection"""
3210
 
        class Test(unittest.TestCase):
3211
 
            def runTest(self):
3212
 
                pass
3213
 
        result = self.LeakRecordingResult()
3214
 
        test = Test()
3215
 
        result.startTestRun()
3216
 
        test.run(result)
3217
 
        result.stopTestRun()
3218
 
        self.assertEqual(result._tests_leaking_threads_count, 0)
3219
 
        self.assertEqual(result.leaks, [])
3220
 
        
3221
 
    def test_thread_leak(self):
3222
 
        """Ensure a thread that outlives the running of a test is reported
3223
 
 
3224
 
        Uses a thread that blocks on an event, and is started by the inner
3225
 
        test case. As the thread outlives the inner case's run, it should be
3226
 
        detected as a leak, but the event is then set so that the thread can
3227
 
        be safely joined in cleanup so it's not leaked for real.
3228
 
        """
3229
 
        event = threading.Event()
3230
 
        thread = threading.Thread(name="Leaker", target=event.wait)
3231
 
        class Test(tests.TestCase):
3232
 
            def test_leak(self):
3233
 
                thread.start()
3234
 
        result = self.LeakRecordingResult()
3235
 
        test = Test("test_leak")
3236
 
        self.addCleanup(thread.join)
3237
 
        self.addCleanup(event.set)
3238
 
        result.startTestRun()
3239
 
        test.run(result)
3240
 
        result.stopTestRun()
3241
 
        self.assertEqual(result._tests_leaking_threads_count, 1)
3242
 
        self.assertEqual(result._first_thread_leaker_id, test.id())
3243
 
        self.assertEqual(result.leaks, [(test, set([thread]))])
3244
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3245
 
 
3246
 
    def test_multiple_leaks(self):
3247
 
        """Check multiple leaks are blamed on the test cases at fault
3248
 
 
3249
 
        Same concept as the previous test, but has one inner test method that
3250
 
        leaks two threads, and one that doesn't leak at all.
3251
 
        """
3252
 
        event = threading.Event()
3253
 
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
3254
 
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
3255
 
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
3256
 
        class Test(tests.TestCase):
3257
 
            def test_first_leak(self):
3258
 
                thread_b.start()
3259
 
            def test_second_no_leak(self):
3260
 
                pass
3261
 
            def test_third_leak(self):
3262
 
                thread_c.start()
3263
 
                thread_a.start()
3264
 
        result = self.LeakRecordingResult()
3265
 
        first_test = Test("test_first_leak")
3266
 
        third_test = Test("test_third_leak")
3267
 
        self.addCleanup(thread_a.join)
3268
 
        self.addCleanup(thread_b.join)
3269
 
        self.addCleanup(thread_c.join)
3270
 
        self.addCleanup(event.set)
3271
 
        result.startTestRun()
3272
 
        unittest.TestSuite(
3273
 
            [first_test, Test("test_second_no_leak"), third_test]
3274
 
            ).run(result)
3275
 
        result.stopTestRun()
3276
 
        self.assertEqual(result._tests_leaking_threads_count, 2)
3277
 
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
3278
 
        self.assertEqual(result.leaks, [
3279
 
            (first_test, set([thread_b])),
3280
 
            (third_test, set([thread_a, thread_c]))])
3281
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3282
 
 
3283
 
 
3284
 
class TestPostMortemDebugging(tests.TestCase):
3285
 
    """Check post mortem debugging works when tests fail or error"""
3286
 
 
3287
 
    class TracebackRecordingResult(tests.ExtendedTestResult):
3288
 
        def __init__(self):
3289
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3290
 
            self.postcode = None
3291
 
        def _post_mortem(self, tb=None):
3292
 
            """Record the code object at the end of the current traceback"""
3293
 
            tb = tb or sys.exc_info()[2]
3294
 
            if tb is not None:
3295
 
                next = tb.tb_next
3296
 
                while next is not None:
3297
 
                    tb = next
3298
 
                    next = next.tb_next
3299
 
                self.postcode = tb.tb_frame.f_code
3300
 
        def report_error(self, test, err):
3301
 
            pass
3302
 
        def report_failure(self, test, err):
3303
 
            pass
3304
 
 
3305
 
    def test_location_unittest_error(self):
3306
 
        """Needs right post mortem traceback with erroring unittest case"""
3307
 
        class Test(unittest.TestCase):
3308
 
            def runTest(self):
3309
 
                raise RuntimeError
3310
 
        result = self.TracebackRecordingResult()
3311
 
        Test().run(result)
3312
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3313
 
 
3314
 
    def test_location_unittest_failure(self):
3315
 
        """Needs right post mortem traceback with failing unittest case"""
3316
 
        class Test(unittest.TestCase):
3317
 
            def runTest(self):
3318
 
                raise self.failureException
3319
 
        result = self.TracebackRecordingResult()
3320
 
        Test().run(result)
3321
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3322
 
 
3323
 
    def test_location_bt_error(self):
3324
 
        """Needs right post mortem traceback with erroring bzrlib.tests case"""
3325
 
        class Test(tests.TestCase):
3326
 
            def test_error(self):
3327
 
                raise RuntimeError
3328
 
        result = self.TracebackRecordingResult()
3329
 
        Test("test_error").run(result)
3330
 
        self.assertEqual(result.postcode, Test.test_error.func_code)
3331
 
 
3332
 
    def test_location_bt_failure(self):
3333
 
        """Needs right post mortem traceback with failing bzrlib.tests case"""
3334
 
        class Test(tests.TestCase):
3335
 
            def test_failure(self):
3336
 
                raise self.failureException
3337
 
        result = self.TracebackRecordingResult()
3338
 
        Test("test_failure").run(result)
3339
 
        self.assertEqual(result.postcode, Test.test_failure.func_code)
3340
 
 
3341
 
    def test_env_var_triggers_post_mortem(self):
3342
 
        """Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
3343
 
        import pdb
3344
 
        result = tests.ExtendedTestResult(StringIO(), 0, 1)
3345
 
        post_mortem_calls = []
3346
 
        self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3347
 
        self.overrideEnv('BZR_TEST_PDB', None)
3348
 
        result._post_mortem(1)
3349
 
        self.overrideEnv('BZR_TEST_PDB', 'on')
3350
 
        result._post_mortem(2)
3351
 
        self.assertEqual([2], post_mortem_calls)
3352
 
 
3353
 
 
3354
 
class TestRunSuite(tests.TestCase):
3355
 
 
3356
 
    def test_runner_class(self):
3357
 
        """run_suite accepts and uses a runner_class keyword argument."""
3358
 
        class Stub(tests.TestCase):
3359
 
            def test_foo(self):
3360
 
                pass
3361
 
        suite = Stub("test_foo")
3362
 
        calls = []
3363
 
        class MyRunner(tests.TextTestRunner):
3364
 
            def run(self, test):
3365
 
                calls.append(test)
3366
 
                return tests.ExtendedTestResult(self.stream, self.descriptions,
3367
 
                                                self.verbosity)
3368
 
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
3369
 
        self.assertLength(1, calls)
3370
 
 
3371
 
 
3372
 
class _Selftest(object):
3373
 
    """Mixin for tests needing full selftest output"""
3374
 
 
3375
 
    def _inject_stream_into_subunit(self, stream):
3376
 
        """To be overridden by subclasses that run tests out of process"""
3377
 
 
3378
 
    def _run_selftest(self, **kwargs):
3379
 
        sio = StringIO()
3380
 
        self._inject_stream_into_subunit(sio)
3381
 
        tests.selftest(stream=sio, stop_on_failure=False, **kwargs)
3382
 
        return sio.getvalue()
3383
 
 
3384
 
 
3385
 
class _ForkedSelftest(_Selftest):
3386
 
    """Mixin for tests needing full selftest output with forked children"""
3387
 
 
3388
 
    _test_needs_features = [features.subunit]
3389
 
 
3390
 
    def _inject_stream_into_subunit(self, stream):
3391
 
        """Monkey-patch subunit so the extra output goes to stream not stdout
3392
 
 
3393
 
        Some APIs need rewriting so this kind of bogus hackery can be replaced
3394
 
        by passing the stream param from run_tests down into ProtocolTestCase.
3395
 
        """
3396
 
        from subunit import ProtocolTestCase
3397
 
        _original_init = ProtocolTestCase.__init__
3398
 
        def _init_with_passthrough(self, *args, **kwargs):
3399
 
            _original_init(self, *args, **kwargs)
3400
 
            self._passthrough = stream
3401
 
        self.overrideAttr(ProtocolTestCase, "__init__", _init_with_passthrough)
3402
 
 
3403
 
    def _run_selftest(self, **kwargs):
3404
 
        # GZ 2011-05-26: Add a PosixSystem feature so this check can go away
3405
 
        if getattr(os, "fork", None) is None:
3406
 
            raise tests.TestNotApplicable("Platform doesn't support forking")
3407
 
        # Make sure the fork code is actually invoked by claiming two cores
3408
 
        self.overrideAttr(osutils, "local_concurrency", lambda: 2)
3409
 
        kwargs.setdefault("suite_decorators", []).append(tests.fork_decorator)
3410
 
        return super(_ForkedSelftest, self)._run_selftest(**kwargs)
3411
 
 
3412
 
 
3413
 
class TestParallelFork(_ForkedSelftest, tests.TestCase):
3414
 
    """Check operation of --parallel=fork selftest option"""
3415
 
 
3416
 
    def test_error_in_child_during_fork(self):
3417
 
        """Error in a forked child during test setup should get reported"""
3418
 
        class Test(tests.TestCase):
3419
 
            def testMethod(self):
3420
 
                pass
3421
 
        # We don't care what, just break something that a child will run
3422
 
        self.overrideAttr(tests, "workaround_zealous_crypto_random", None)
3423
 
        out = self._run_selftest(test_suite_factory=Test)
3424
 
        # Lines from the tracebacks of the two child processes may be mixed
3425
 
        # together due to the way subunit parses and forwards the streams,
3426
 
        # so permit extra lines between each part of the error output.
3427
 
        self.assertContainsRe(out,
3428
 
            "Traceback.*:\n"
3429
 
            "(?:.*\n)*"
3430
 
            ".+ in fork_for_tests\n"
3431
 
            "(?:.*\n)*"
3432
 
            "\s*workaround_zealous_crypto_random\(\)\n"
3433
 
            "(?:.*\n)*"
3434
 
            "TypeError:")
3435
 
 
3436
 
 
3437
 
class TestUncollectedWarnings(_Selftest, tests.TestCase):
3438
 
    """Check a test case still alive after being run emits a warning"""
3439
 
 
3440
 
    class Test(tests.TestCase):
3441
 
        def test_pass(self):
3442
 
            pass
3443
 
        def test_self_ref(self):
3444
 
            self.also_self = self.test_self_ref
3445
 
        def test_skip(self):
3446
 
            self.skip("Don't need")
3447
 
 
3448
 
    def _get_suite(self):
3449
 
        return TestUtil.TestSuite([
3450
 
            self.Test("test_pass"),
3451
 
            self.Test("test_self_ref"),
3452
 
            self.Test("test_skip"),
3453
 
            ])
3454
 
 
3455
 
    def _run_selftest_with_suite(self, **kwargs):
3456
 
        old_flags = tests.selftest_debug_flags
3457
 
        tests.selftest_debug_flags = old_flags.union(["uncollected_cases"])
3458
 
        gc_on = gc.isenabled()
3459
 
        if gc_on:
3460
 
            gc.disable()
3461
 
        try:
3462
 
            output = self._run_selftest(test_suite_factory=self._get_suite,
3463
 
                **kwargs)
3464
 
        finally:
3465
 
            if gc_on:
3466
 
                gc.enable()
3467
 
            tests.selftest_debug_flags = old_flags
3468
 
        self.assertNotContainsRe(output, "Uncollected test case.*test_pass")
3469
 
        self.assertContainsRe(output, "Uncollected test case.*test_self_ref")
3470
 
        return output
3471
 
 
3472
 
    def test_testsuite(self):
3473
 
        self._run_selftest_with_suite()
3474
 
 
3475
 
    def test_pattern(self):
3476
 
        out = self._run_selftest_with_suite(pattern="test_(?:pass|self_ref)$")
3477
 
        self.assertNotContainsRe(out, "test_skip")
3478
 
 
3479
 
    def test_exclude_pattern(self):
3480
 
        out = self._run_selftest_with_suite(exclude_pattern="test_skip$")
3481
 
        self.assertNotContainsRe(out, "test_skip")
3482
 
 
3483
 
    def test_random_seed(self):
3484
 
        self._run_selftest_with_suite(random_seed="now")
3485
 
 
3486
 
    def test_matching_tests_first(self):
3487
 
        self._run_selftest_with_suite(matching_tests_first=True,
3488
 
            pattern="test_self_ref$")
3489
 
 
3490
 
    def test_starting_with_and_exclude(self):
3491
 
        out = self._run_selftest_with_suite(starting_with=["bt."],
3492
 
            exclude_pattern="test_skip$")
3493
 
        self.assertNotContainsRe(out, "test_skip")
3494
 
 
3495
 
    def test_additonal_decorator(self):
3496
 
        out = self._run_selftest_with_suite(
3497
 
            suite_decorators=[tests.TestDecorator])
3498
 
 
3499
 
 
3500
 
class TestUncollectedWarningsSubunit(TestUncollectedWarnings):
3501
 
    """Check warnings from tests staying alive are emitted with subunit"""
3502
 
 
3503
 
    _test_needs_features = [features.subunit]
3504
 
 
3505
 
    def _run_selftest_with_suite(self, **kwargs):
3506
 
        return TestUncollectedWarnings._run_selftest_with_suite(self,
3507
 
            runner_class=tests.SubUnitBzrRunner, **kwargs)
3508
 
 
3509
 
 
3510
 
class TestUncollectedWarningsForked(_ForkedSelftest, TestUncollectedWarnings):
3511
 
    """Check warnings from tests staying alive are emitted when forking"""
3512
 
 
3513
 
 
3514
 
class TestEnvironHandling(tests.TestCase):
3515
 
 
3516
 
    def test_overrideEnv_None_called_twice_doesnt_leak(self):
3517
 
        self.assertFalse('MYVAR' in os.environ)
3518
 
        self.overrideEnv('MYVAR', '42')
3519
 
        # We use an embedded test to make sure we fix the _captureVar bug
3520
 
        class Test(tests.TestCase):
3521
 
            def test_me(self):
3522
 
                # The first call save the 42 value
3523
 
                self.overrideEnv('MYVAR', None)
3524
 
                self.assertEqual(None, os.environ.get('MYVAR'))
3525
 
                # Make sure we can call it twice
3526
 
                self.overrideEnv('MYVAR', None)
3527
 
                self.assertEqual(None, os.environ.get('MYVAR'))
3528
 
        output = StringIO()
3529
 
        result = tests.TextTestResult(output, 0, 1)
3530
 
        Test('test_me').run(result)
3531
 
        if not result.wasStrictlySuccessful():
3532
 
            self.fail(output.getvalue())
3533
 
        # We get our value back
3534
 
        self.assertEqual('42', os.environ.get('MYVAR'))
3535
 
 
3536
 
 
3537
 
class TestIsolatedEnv(tests.TestCase):
3538
 
    """Test isolating tests from os.environ.
3539
 
 
3540
 
    Since we use tests that are already isolated from os.environ a bit of care
3541
 
    should be taken when designing the tests to avoid bootstrap side-effects.
3542
 
    The tests start an already clean os.environ which allow doing valid
3543
 
    assertions about which variables are present or not and design tests around
3544
 
    these assertions.
3545
 
    """
3546
 
 
3547
 
    class ScratchMonkey(tests.TestCase):
3548
 
 
3549
 
        def test_me(self):
3550
 
            pass
3551
 
 
3552
 
    def test_basics(self):
3553
 
        # Make sure we know the definition of BZR_HOME: not part of os.environ
3554
 
        # for tests.TestCase.
3555
 
        self.assertTrue('BZR_HOME' in tests.isolated_environ)
3556
 
        self.assertEqual(None, tests.isolated_environ['BZR_HOME'])
3557
 
        # Being part of isolated_environ, BZR_HOME should not appear here
3558
 
        self.assertFalse('BZR_HOME' in os.environ)
3559
 
        # Make sure we know the definition of LINES: part of os.environ for
3560
 
        # tests.TestCase
3561
 
        self.assertTrue('LINES' in tests.isolated_environ)
3562
 
        self.assertEqual('25', tests.isolated_environ['LINES'])
3563
 
        self.assertEqual('25', os.environ['LINES'])
3564
 
 
3565
 
    def test_injecting_unknown_variable(self):
3566
 
        # BZR_HOME is known to be absent from os.environ
3567
 
        test = self.ScratchMonkey('test_me')
3568
 
        tests.override_os_environ(test, {'BZR_HOME': 'foo'})
3569
 
        self.assertEqual('foo', os.environ['BZR_HOME'])
3570
 
        tests.restore_os_environ(test)
3571
 
        self.assertFalse('BZR_HOME' in os.environ)
3572
 
 
3573
 
    def test_injecting_known_variable(self):
3574
 
        test = self.ScratchMonkey('test_me')
3575
 
        # LINES is known to be present in os.environ
3576
 
        tests.override_os_environ(test, {'LINES': '42'})
3577
 
        self.assertEqual('42', os.environ['LINES'])
3578
 
        tests.restore_os_environ(test)
3579
 
        self.assertEqual('25', os.environ['LINES'])
3580
 
 
3581
 
    def test_deleting_variable(self):
3582
 
        test = self.ScratchMonkey('test_me')
3583
 
        # LINES is known to be present in os.environ
3584
 
        tests.override_os_environ(test, {'LINES': None})
3585
 
        self.assertTrue('LINES' not in os.environ)
3586
 
        tests.restore_os_environ(test)
3587
 
        self.assertEqual('25', os.environ['LINES'])
3588
 
 
3589
 
 
3590
 
class TestDocTestSuiteIsolation(tests.TestCase):
3591
 
    """Test that `tests.DocTestSuite` isolates doc tests from os.environ.
3592
 
 
3593
 
    Since tests.TestCase alreay provides an isolation from os.environ, we use
3594
 
    the clean environment as a base for testing. To precisely capture the
3595
 
    isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
3596
 
    compare against.
3597
 
 
3598
 
    We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
3599
 
    not `os.environ` so each test overrides it to suit its needs.
3600
 
 
3601
 
    """
3602
 
 
3603
 
    def get_doctest_suite_for_string(self, klass, string):
3604
 
        class Finder(doctest.DocTestFinder):
3605
 
 
3606
 
            def find(*args, **kwargs):
3607
 
                test = doctest.DocTestParser().get_doctest(
3608
 
                    string, {}, 'foo', 'foo.py', 0)
3609
 
                return [test]
3610
 
 
3611
 
        suite = klass(test_finder=Finder())
3612
 
        return suite
3613
 
 
3614
 
    def run_doctest_suite_for_string(self, klass, string):
3615
 
        suite = self.get_doctest_suite_for_string(klass, string)
3616
 
        output = StringIO()
3617
 
        result = tests.TextTestResult(output, 0, 1)
3618
 
        suite.run(result)
3619
 
        return result, output
3620
 
 
3621
 
    def assertDocTestStringSucceds(self, klass, string):
3622
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3623
 
        if not result.wasStrictlySuccessful():
3624
 
            self.fail(output.getvalue())
3625
 
 
3626
 
    def assertDocTestStringFails(self, klass, string):
3627
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3628
 
        if result.wasStrictlySuccessful():
3629
 
            self.fail(output.getvalue())
3630
 
 
3631
 
    def test_injected_variable(self):
3632
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
3633
 
        test = """
3634
 
            >>> import os
3635
 
            >>> os.environ['LINES']
3636
 
            '42'
3637
 
            """
3638
 
        # doctest.DocTestSuite fails as it sees '25'
3639
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3640
 
        # tests.DocTestSuite sees '42'
3641
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3642
 
 
3643
 
    def test_deleted_variable(self):
3644
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
3645
 
        test = """
3646
 
            >>> import os
3647
 
            >>> os.environ.get('LINES')
3648
 
            """
3649
 
        # doctest.DocTestSuite fails as it sees '25'
3650
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3651
 
        # tests.DocTestSuite sees None
3652
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3653
 
 
3654
 
 
3655
 
class TestSelftestExcludePatterns(tests.TestCase):
3656
 
 
3657
 
    def setUp(self):
3658
 
        super(TestSelftestExcludePatterns, self).setUp()
3659
 
        self.overrideAttr(tests, 'test_suite', self.suite_factory)
3660
 
 
3661
 
    def suite_factory(self, keep_only=None, starting_with=None):
3662
 
        """A test suite factory with only a few tests."""
3663
 
        class Test(tests.TestCase):
3664
 
            def id(self):
3665
 
                # We don't need the full class path
3666
 
                return self._testMethodName
3667
 
            def a(self):
3668
 
                pass
3669
 
            def b(self):
3670
 
                pass
3671
 
            def c(self):
3672
 
                pass
3673
 
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
3674
 
 
3675
 
    def assertTestList(self, expected, *selftest_args):
3676
 
        # We rely on setUp installing the right test suite factory so we can
3677
 
        # test at the command level without loading the whole test suite
3678
 
        out, err = self.run_bzr(('selftest', '--list') + selftest_args)
3679
 
        actual = out.splitlines()
3680
 
        self.assertEqual(expected, actual)
3681
 
 
3682
 
    def test_full_list(self):
3683
 
        self.assertTestList(['a', 'b', 'c'])
3684
 
 
3685
 
    def test_single_exclude(self):
3686
 
        self.assertTestList(['b', 'c'], '-x', 'a')
3687
 
 
3688
 
    def test_mutiple_excludes(self):
3689
 
        self.assertTestList(['c'], '-x', 'a', '-x', 'b')
3690
 
 
3691
 
 
3692
 
class TestCounterHooks(tests.TestCase, SelfTestHelper):
3693
 
 
3694
 
    _test_needs_features = [features.subunit]
3695
 
 
3696
 
    def setUp(self):
3697
 
        super(TestCounterHooks, self).setUp()
3698
 
        class Test(tests.TestCase):
3699
 
 
3700
 
            def setUp(self):
3701
 
                super(Test, self).setUp()
3702
 
                self.hooks = hooks.Hooks()
3703
 
                self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
3704
 
                self.install_counter_hook(self.hooks, 'myhook')
3705
 
 
3706
 
            def no_hook(self):
3707
 
                pass
3708
 
 
3709
 
            def run_hook_once(self):
3710
 
                for hook in self.hooks['myhook']:
3711
 
                    hook(self)
3712
 
 
3713
 
        self.test_class = Test
3714
 
 
3715
 
    def assertHookCalls(self, expected_calls, test_name):
3716
 
        test = self.test_class(test_name)
3717
 
        result = unittest.TestResult()
3718
 
        test.run(result)
3719
 
        self.assertTrue(hasattr(test, '_counters'))
3720
 
        self.assertTrue(test._counters.has_key('myhook'))
3721
 
        self.assertEqual(expected_calls, test._counters['myhook'])
3722
 
 
3723
 
    def test_no_hook(self):
3724
 
        self.assertHookCalls(0, 'no_hook')
3725
 
 
3726
 
    def test_run_hook_once(self):
3727
 
        tt = features.testtools
3728
 
        if tt.module.__version__ < (0, 9, 8):
3729
 
            raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
3730
 
        self.assertHookCalls(1, 'run_hook_once')
 
2214
        self.assertEquals('bzrlib', tpr.resolve_alias('bzrlib'))
 
2215
        self.assertEquals('bzrlib.doc', tpr.resolve_alias('bd'))
 
2216
        self.assertEquals('bzrlib.utils', tpr.resolve_alias('bu'))
 
2217
        self.assertEquals('bzrlib.tests', tpr.resolve_alias('bt'))
 
2218
        self.assertEquals('bzrlib.tests.blackbox', tpr.resolve_alias('bb'))
 
2219
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))