~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Tarmac
  • Author(s): Vincent Ladeuil
  • Date: 2017-01-30 14:42:05 UTC
  • mfrom: (6620.1.1 trunk)
  • Revision ID: tarmac-20170130144205-r8fh2xpmiuxyozpv
Merge  2.7 into trunk including fix for bug #1657238 [r=vila]

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007, 2008 Canonical Ltd
 
1
# Copyright (C) 2005-2013, 2016 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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
16
 
17
17
"""Tests for the test framework."""
18
18
 
19
 
import cStringIO
 
19
from cStringIO import StringIO
 
20
import gc
 
21
import doctest
20
22
import os
21
 
from StringIO import StringIO
 
23
import signal
22
24
import sys
 
25
import threading
23
26
import time
24
27
import unittest
25
28
import warnings
26
29
 
 
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
 
27
42
import bzrlib
28
43
from bzrlib import (
29
44
    branchbuilder,
30
45
    bzrdir,
 
46
    controldir,
31
47
    errors,
 
48
    hooks,
 
49
    lockdir,
32
50
    memorytree,
33
51
    osutils,
34
52
    remote,
35
53
    repository,
36
54
    symbol_versioning,
37
55
    tests,
 
56
    transport,
38
57
    workingtree,
 
58
    workingtree_3,
 
59
    workingtree_4,
39
60
    )
40
 
from bzrlib.progress import _BaseProgressBar
41
61
from bzrlib.repofmt import (
42
 
    pack_repo,
43
 
    weaverepo,
 
62
    groupcompress_repo,
44
63
    )
45
64
from bzrlib.symbol_versioning import (
46
 
    one_zero,
47
 
    zero_eleven,
48
 
    zero_ten,
 
65
    deprecated_function,
 
66
    deprecated_in,
 
67
    deprecated_method,
49
68
    )
50
69
from bzrlib.tests import (
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
 
70
    features,
 
71
    test_lsprof,
 
72
    test_server,
 
73
    TestUtil,
 
74
    )
 
75
from bzrlib.trace import note, mutter
 
76
from bzrlib.transport import memory
84
77
 
85
78
 
86
79
def _test_ids(test_suite):
87
80
    """Get the ids for the tests in a test suite."""
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):
 
81
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
 
82
 
 
83
 
 
84
class MetaTestLog(tests.TestCase):
103
85
 
104
86
    def test_logging(self):
105
87
        """Test logs are captured when a test fails."""
106
88
        self.log('a test message')
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):
 
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):
122
99
 
123
100
    def test_unicode_paths(self):
124
 
        self.requireFeature(tests.UnicodeFilename)
 
101
        self.requireFeature(features.UnicodeFilenameFeature)
125
102
 
126
103
        filename = u'hell\u00d8'
127
104
        self.build_tree_contents([(filename, 'contents of hello')])
128
 
        self.failUnlessExists(filename)
129
 
 
130
 
 
131
 
class TestTransportProviderAdapter(TestCase):
 
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):
132
122
    """A group of tests that test the transport implementation adaption core.
133
123
 
134
 
    This is a meta test that the tests are applied to all available 
 
124
    This is a meta test that the tests are applied to all available
135
125
    transports.
136
126
 
137
 
    This will be generalised in the future which is why it is in this 
 
127
    This will be generalised in the future which is why it is in this
138
128
    test file even though it is specific to transport tests at the moment.
139
129
    """
140
130
 
141
131
    def test_get_transport_permutations(self):
142
132
        # this checks that get_test_permutations defined by the module is
143
 
        # called by the adapter get_transport_test_permutations method.
 
133
        # called by the get_transport_test_permutations function.
144
134
        class MockModule(object):
145
135
            def get_test_permutations(self):
146
136
                return sample_permutation
147
137
        sample_permutation = [(1,2), (3,4)]
148
 
        from bzrlib.tests.test_transport_implementations \
149
 
            import TransportTestProviderAdapter
150
 
        adapter = TransportTestProviderAdapter()
 
138
        from bzrlib.tests.per_transport import get_transport_test_permutations
151
139
        self.assertEqual(sample_permutation,
152
 
                         adapter.get_transport_test_permutations(MockModule()))
 
140
                         get_transport_test_permutations(MockModule()))
153
141
 
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
 
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
161
148
        from bzrlib.transport import _get_transport_modules
162
149
        modules = _get_transport_modules()
163
150
        permutation_count = 0
164
151
        for module in modules:
165
152
            try:
166
 
                permutation_count += len(reduce(getattr, 
 
153
                permutation_count += len(reduce(getattr,
167
154
                    (module + ".get_test_permutations").split('.')[1:],
168
155
                     __import__(module))())
169
156
            except errors.DependencyNotPresent:
170
157
                pass
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)))))
 
158
        scenarios = transport_test_permutations()
 
159
        self.assertEqual(permutation_count, len(scenarios))
176
160
 
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
 
        #
 
161
    def test_scenarios_include_transport_class(self):
181
162
        # This test used to know about all the possible transports and the
182
163
        # order they were returned but that seems overly brittle (mbp
183
164
        # 20060307)
184
 
        from bzrlib.tests.test_transport_implementations \
185
 
            import TransportTestProviderAdapter
186
 
        scenarios = TransportTestProviderAdapter().scenarios
 
165
        from bzrlib.tests.per_transport import transport_test_permutations
 
166
        scenarios = transport_test_permutations()
187
167
        # there are at least that many builtin transports
188
168
        self.assertTrue(len(scenarios) > 6)
189
169
        one_scenario = scenarios[0]
194
174
                                   bzrlib.transport.Server))
195
175
 
196
176
 
197
 
class TestBranchProviderAdapter(TestCase):
198
 
    """A group of tests that test the branch implementation test adapter."""
 
177
class TestBranchScenarios(tests.TestCase):
199
178
 
200
 
    def test_constructor(self):
 
179
    def test_scenarios(self):
201
180
        # check that constructor parameters are passed through to the adapted
202
181
        # test.
203
 
        from bzrlib.tests.branch_implementations import BranchTestProviderAdapter
 
182
        from bzrlib.tests.per_branch import make_scenarios
204
183
        server1 = "a"
205
184
        server2 = "b"
206
185
        formats = [("c", "C"), ("d", "D")]
207
 
        adapter = BranchTestProviderAdapter(server1, server2, formats)
208
 
        self.assertEqual(2, len(adapter.scenarios))
 
186
        scenarios = make_scenarios(server1, server2, formats)
 
187
        self.assertEqual(2, len(scenarios))
209
188
        self.assertEqual([
210
189
            ('str',
211
190
             {'branch_format': 'c',
217
196
              'bzrdir_format': 'D',
218
197
              'transport_readonly_server': 'b',
219
198
              'transport_server': 'a'})],
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):
 
199
            scenarios)
 
200
 
 
201
 
 
202
class TestBzrDirScenarios(tests.TestCase):
 
203
 
 
204
    def test_scenarios(self):
227
205
        # check that constructor parameters are passed through to the adapted
228
206
        # test.
229
 
        from bzrlib.tests.bzrdir_implementations import BzrDirTestProviderAdapter
 
207
        from bzrlib.tests.per_controldir import make_scenarios
230
208
        vfs_factory = "v"
231
209
        server1 = "a"
232
210
        server2 = "b"
233
211
        formats = ["c", "d"]
234
 
        adapter = BzrDirTestProviderAdapter(vfs_factory,
235
 
            server1, server2, formats)
 
212
        scenarios = make_scenarios(vfs_factory, server1, server2, formats)
236
213
        self.assertEqual([
237
214
            ('str',
238
215
             {'bzrdir_format': 'c',
244
221
              'transport_readonly_server': 'b',
245
222
              'transport_server': 'a',
246
223
              'vfs_transport_factory': 'v'})],
247
 
            adapter.scenarios)
248
 
 
249
 
 
250
 
class TestRepositoryParameterisation(TestCase):
251
 
    """A group of tests that test the repository implementation test adapter."""
 
224
            scenarios)
 
225
 
 
226
 
 
227
class TestRepositoryScenarios(tests.TestCase):
252
228
 
253
229
    def test_formats_to_scenarios(self):
254
 
        """The adapter can generate all the scenarios needed."""
255
 
        from bzrlib.tests.repository_implementations import formats_to_scenarios
 
230
        from bzrlib.tests.per_repository import formats_to_scenarios
256
231
        formats = [("(c)", remote.RemoteRepositoryFormat()),
257
232
                   ("(d)", repository.format_registry.get(
258
 
                        'Bazaar pack repository format 1 (needs bzr 0.92)\n'))]
 
233
                    'Bazaar repository format 2a (needs bzr 1.16 or later)\n'))]
259
234
        no_vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
260
235
            None)
261
236
        vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
262
237
            vfs_transport_factory="vfs")
263
238
        # no_vfs generate scenarios without vfs_transport_factory
264
 
        self.assertEqual([
 
239
        expected = [
265
240
            ('RemoteRepositoryFormat(c)',
266
241
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
267
242
              'repository_format': remote.RemoteRepositoryFormat(),
268
243
              'transport_readonly_server': 'readonly',
269
244
              'transport_server': 'server'}),
270
 
            ('RepositoryFormatKnitPack1(d)',
 
245
            ('RepositoryFormat2a(d)',
271
246
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
272
 
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
 
247
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
273
248
              'transport_readonly_server': 'readonly',
274
 
              'transport_server': 'server'})],
275
 
            no_vfs_scenarios)
 
249
              'transport_server': 'server'})]
 
250
        self.assertEqual(expected, no_vfs_scenarios)
276
251
        self.assertEqual([
277
252
            ('RemoteRepositoryFormat(c)',
278
253
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
280
255
              'transport_readonly_server': 'readonly',
281
256
              'transport_server': 'server',
282
257
              'vfs_transport_factory': 'vfs'}),
283
 
            ('RepositoryFormatKnitPack1(d)',
 
258
            ('RepositoryFormat2a(d)',
284
259
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
285
 
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
 
260
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
286
261
              'transport_readonly_server': 'readonly',
287
262
              'transport_server': 'server',
288
263
              'vfs_transport_factory': 'vfs'})],
289
264
            vfs_scenarios)
290
265
 
291
266
 
292
 
class TestTestScenarioApplier(TestCase):
 
267
class TestTestScenarioApplication(tests.TestCase):
293
268
    """Tests for the test adaption facilities."""
294
269
 
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()
 
270
    def test_apply_scenario(self):
 
271
        from bzrlib.tests import apply_scenario
 
272
        input_test = TestTestScenarioApplication("test_apply_scenario")
313
273
        # setup two adapted tests
314
 
        adapted_test1 = adapter.adapt_test_to_scenario(input_test,
 
274
        adapted_test1 = apply_scenario(input_test,
315
275
            ("new id",
316
276
            {"bzrdir_format":"bzr_format",
317
277
             "repository_format":"repo_fmt",
318
278
             "transport_server":"transport_server",
319
279
             "transport_readonly_server":"readonly-server"}))
320
 
        adapted_test2 = adapter.adapt_test_to_scenario(input_test,
 
280
        adapted_test2 = apply_scenario(input_test,
321
281
            ("new id 2", {"bzrdir_format":None}))
322
282
        # input_test should have been altered.
323
283
        self.assertRaises(AttributeError, getattr, input_test, "bzrdir_format")
324
 
        # the new tests are mutually incompatible, ensuring it has 
 
284
        # the new tests are mutually incompatible, ensuring it has
325
285
        # made new ones, and unspecified elements in the scenario
326
286
        # should not have been altered.
327
287
        self.assertEqual("bzr_format", adapted_test1.bzrdir_format)
330
290
        self.assertEqual("readonly-server",
331
291
            adapted_test1.transport_readonly_server)
332
292
        self.assertEqual(
333
 
            "bzrlib.tests.test_selftest.TestTestScenarioApplier."
334
 
            "test_adapt_test_to_scenario(new id)",
 
293
            "bzrlib.tests.test_selftest.TestTestScenarioApplication."
 
294
            "test_apply_scenario(new id)",
335
295
            adapted_test1.id())
336
296
        self.assertEqual(None, adapted_test2.bzrdir_format)
337
297
        self.assertEqual(
338
 
            "bzrlib.tests.test_selftest.TestTestScenarioApplier."
339
 
            "test_adapt_test_to_scenario(new id 2)",
 
298
            "bzrlib.tests.test_selftest.TestTestScenarioApplication."
 
299
            "test_apply_scenario(new id 2)",
340
300
            adapted_test2.id())
341
301
 
342
302
 
343
 
class TestInterRepositoryProviderAdapter(TestCase):
344
 
    """A group of tests that test the InterRepository test adapter."""
 
303
class TestInterRepositoryScenarios(tests.TestCase):
345
304
 
346
 
    def test_adapted_tests(self):
 
305
    def test_scenarios(self):
347
306
        # check that constructor parameters are passed through to the adapted
348
307
        # test.
349
 
        from bzrlib.tests.interrepository_implementations import \
350
 
            InterRepositoryTestProviderAdapter
 
308
        from bzrlib.tests.per_interrepository import make_scenarios
351
309
        server1 = "a"
352
310
        server2 = "b"
353
 
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
354
 
        adapter = InterRepositoryTestProviderAdapter(server1, server2, formats)
 
311
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
 
312
        scenarios = make_scenarios(server1, server2, formats)
355
313
        self.assertEqual([
356
 
            ('str,str,str',
357
 
             {'interrepo_class': str,
358
 
              'repository_format': 'C1',
 
314
            ('C0,str,str',
 
315
             {'repository_format': 'C1',
359
316
              'repository_format_to': 'C2',
360
317
              'transport_readonly_server': 'b',
361
 
              'transport_server': 'a'}),
362
 
            ('int,str,str',
363
 
             {'interrepo_class': int,
364
 
              'repository_format': 'D1',
 
318
              'transport_server': 'a',
 
319
              'extra_setup': 'C3'}),
 
320
            ('D0,str,str',
 
321
             {'repository_format': 'D1',
365
322
              'repository_format_to': 'D2',
366
323
              'transport_readonly_server': 'b',
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."""
 
324
              'transport_server': 'a',
 
325
              'extra_setup': 'D3'})],
 
326
            scenarios)
 
327
 
 
328
 
 
329
class TestWorkingTreeScenarios(tests.TestCase):
373
330
 
374
331
    def test_scenarios(self):
375
332
        # check that constructor parameters are passed through to the adapted
376
333
        # test.
377
 
        from bzrlib.tests.workingtree_implementations \
378
 
            import WorkingTreeTestProviderAdapter
 
334
        from bzrlib.tests.per_workingtree import make_scenarios
379
335
        server1 = "a"
380
336
        server2 = "b"
381
 
        formats = [workingtree.WorkingTreeFormat2(),
382
 
                   workingtree.WorkingTreeFormat3(),]
383
 
        adapter = WorkingTreeTestProviderAdapter(server1, server2, formats)
 
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')
384
343
        self.assertEqual([
385
 
            ('WorkingTreeFormat2',
 
344
            ('WorkingTreeFormat4',
386
345
             {'bzrdir_format': formats[0]._matchingbzrdir,
387
346
              'transport_readonly_server': 'b',
388
347
              'transport_server': 'a',
391
350
             {'bzrdir_format': formats[1]._matchingbzrdir,
392
351
              'transport_readonly_server': 'b',
393
352
              'transport_server': 'a',
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
 
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
407
379
        # revision_tree_from_workingtree.
408
380
 
409
 
        from bzrlib.tests.tree_implementations import (
410
 
            TreeTestProviderAdapter,
 
381
        from bzrlib.tests.per_tree import (
 
382
            _dirstate_tree_from_workingtree,
 
383
            make_scenarios,
 
384
            preview_tree_pre,
 
385
            preview_tree_post,
411
386
            return_parameter,
412
387
            revision_tree_from_workingtree
413
388
            )
414
 
        input_test = TestTreeProviderAdapter(
415
 
            "test_adapted_tests")
416
389
        server1 = "a"
417
390
        server2 = "b"
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):
 
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):
449
464
    """A group of tests that test the InterTreeTestAdapter."""
450
465
 
451
 
    def test_adapted_tests(self):
 
466
    def test_scenarios(self):
452
467
        # check that constructor parameters are passed through to the adapted
453
468
        # test.
454
469
        # for InterTree tests we want the machinery to bring up two trees in
458
473
        # unlike the TestProviderAdapter we dont want to automatically add a
459
474
        # parameterized one for WorkingTree - the optimisers will tell us what
460
475
        # ones to add.
461
 
        from bzrlib.tests.tree_implementations import (
 
476
        from bzrlib.tests.per_tree import (
462
477
            return_parameter,
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")
 
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")
471
486
        server1 = "a"
472
487
        server2 = "b"
473
 
        format1 = WorkingTreeFormat2()
 
488
        format1 = WorkingTreeFormat4()
474
489
        format2 = WorkingTreeFormat3()
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):
 
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):
498
520
 
499
521
    def test_home_is_not_working(self):
500
522
        self.assertNotEqual(self.test_dir, self.test_home_dir)
502
524
        self.assertIsSameRealPath(self.test_dir, cwd)
503
525
        self.assertIsSameRealPath(self.test_home_dir, os.environ['HOME'])
504
526
 
505
 
 
506
 
class TestTestCaseWithMemoryTransport(TestCaseWithMemoryTransport):
 
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):
507
561
 
508
562
    def test_home_is_non_existant_dir_under_root(self):
509
563
        """The test_home_dir for TestCaseWithMemoryTransport is missing.
510
564
 
511
565
        This is because TestCaseWithMemoryTransport is for tests that do not
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 
 
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
514
568
        few tests should need to do that), and having a missing dir as home is
515
569
        an effective way to ensure that this is the case.
516
570
        """
518
572
            self.TEST_ROOT + "/MemoryTransportMissingHomeDir",
519
573
            self.test_home_dir)
520
574
        self.assertIsSameRealPath(self.test_home_dir, os.environ['HOME'])
521
 
        
 
575
 
522
576
    def test_cwd_is_TEST_ROOT(self):
523
577
        self.assertIsSameRealPath(self.test_dir, self.TEST_ROOT)
524
578
        cwd = osutils.getcwd()
525
579
        self.assertIsSameRealPath(self.test_dir, cwd)
526
580
 
 
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
 
527
589
    def test_make_branch_and_memory_tree(self):
528
590
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
529
591
 
533
595
        tree = self.make_branch_and_memory_tree('dir')
534
596
        # Guard against regression into MemoryTransport leaking
535
597
        # files to disk instead of keeping them in memory.
536
 
        self.failIf(osutils.lexists('dir'))
 
598
        self.assertFalse(osutils.lexists('dir'))
537
599
        self.assertIsInstance(tree, memorytree.MemoryTree)
538
600
 
539
601
    def test_make_branch_and_memory_tree_with_format(self):
540
602
        """make_branch_and_memory_tree should accept a format option."""
541
603
        format = bzrdir.BzrDirMetaFormat1()
542
 
        format.repository_format = weaverepo.RepositoryFormat7()
 
604
        format.repository_format = repository.format_registry.get_default()
543
605
        tree = self.make_branch_and_memory_tree('dir', format=format)
544
606
        # Guard against regression into MemoryTransport leaking
545
607
        # files to disk instead of keeping them in memory.
546
 
        self.failIf(osutils.lexists('dir'))
 
608
        self.assertFalse(osutils.lexists('dir'))
547
609
        self.assertIsInstance(tree, memorytree.MemoryTree)
548
610
        self.assertEqual(format.repository_format.__class__,
549
611
            tree.branch.repository._format.__class__)
553
615
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
554
616
        # Guard against regression into MemoryTransport leaking
555
617
        # files to disk instead of keeping them in memory.
556
 
        self.failIf(osutils.lexists('dir'))
 
618
        self.assertFalse(osutils.lexists('dir'))
557
619
 
558
620
    def test_make_branch_builder_with_format(self):
559
621
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
560
622
        # that the format objects are used.
561
623
        format = bzrdir.BzrDirMetaFormat1()
562
 
        repo_format = weaverepo.RepositoryFormat7()
 
624
        repo_format = repository.format_registry.get_default()
563
625
        format.repository_format = repo_format
564
626
        builder = self.make_branch_builder('dir', format=format)
565
627
        the_branch = builder.get_branch()
566
628
        # Guard against regression into MemoryTransport leaking
567
629
        # files to disk instead of keeping them in memory.
568
 
        self.failIf(osutils.lexists('dir'))
 
630
        self.assertFalse(osutils.lexists('dir'))
569
631
        self.assertEqual(format.repository_format.__class__,
570
632
                         the_branch.repository._format.__class__)
571
633
        self.assertEqual(repo_format.get_format_string(),
577
639
        the_branch = builder.get_branch()
578
640
        # Guard against regression into MemoryTransport leaking
579
641
        # files to disk instead of keeping them in memory.
580
 
        self.failIf(osutils.lexists('dir'))
581
 
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
 
642
        self.assertFalse(osutils.lexists('dir'))
 
643
        dir_format = controldir.format_registry.make_bzrdir('knit')
582
644
        self.assertEqual(dir_format.repository_format.__class__,
583
645
                         the_branch.repository._format.__class__)
584
646
        self.assertEqual('Bazaar-NG Knit Repository Format 1',
585
647
                         self.get_transport().get_bytes(
586
648
                            'dir/.bzr/repository/format'))
587
649
 
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):
 
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):
603
669
    """Tests for the convenience functions TestCaseWithTransport introduces."""
604
670
 
605
671
    def test_get_readonly_url_none(self):
606
 
        from bzrlib.transport import get_transport
607
 
        from bzrlib.transport.memory import MemoryServer
608
672
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
609
 
        self.vfs_transport_factory = MemoryServer
 
673
        self.vfs_transport_factory = memory.MemoryServer
610
674
        self.transport_readonly_server = None
611
675
        # calling get_readonly_transport() constructs a decorator on the url
612
676
        # for the server
613
677
        url = self.get_readonly_url()
614
678
        url2 = self.get_readonly_url('foo/bar')
615
 
        t = get_transport(url)
616
 
        t2 = get_transport(url2)
617
 
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
618
 
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
 
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)
619
683
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
620
684
 
621
685
    def test_get_readonly_url_http(self):
622
686
        from bzrlib.tests.http_server import HttpServer
623
 
        from bzrlib.transport import get_transport
624
 
        from bzrlib.transport.local import LocalURLServer
625
687
        from bzrlib.transport.http import HttpTransportBase
626
 
        self.transport_server = LocalURLServer
 
688
        self.transport_server = test_server.LocalURLServer
627
689
        self.transport_readonly_server = HttpServer
628
690
        # calling get_readonly_transport() gives us a HTTP server instance.
629
691
        url = self.get_readonly_url()
630
692
        url2 = self.get_readonly_url('foo/bar')
631
693
        # the transport returned may be any HttpTransportBase subclass
632
 
        t = get_transport(url)
633
 
        t2 = get_transport(url2)
634
 
        self.failUnless(isinstance(t, HttpTransportBase))
635
 
        self.failUnless(isinstance(t2, HttpTransportBase))
 
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)
636
698
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
637
699
 
638
700
    def test_is_directory(self):
646
708
    def test_make_branch_builder(self):
647
709
        builder = self.make_branch_builder('dir')
648
710
        rev_id = builder.build_commit()
649
 
        self.failUnlessExists('dir')
650
 
        a_dir = bzrdir.BzrDir.open('dir')
 
711
        self.assertPathExists('dir')
 
712
        a_dir = controldir.ControlDir.open('dir')
651
713
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
652
714
        a_branch = a_dir.open_branch()
653
715
        builder_branch = builder.get_branch()
656
718
        self.assertEqual((1, rev_id), a_branch.last_revision_info())
657
719
 
658
720
 
659
 
class TestTestCaseTransports(TestCaseWithTransport):
 
721
class TestTestCaseTransports(tests.TestCaseWithTransport):
660
722
 
661
723
    def setUp(self):
662
724
        super(TestTestCaseTransports, self).setUp()
663
 
        self.vfs_transport_factory = MemoryServer
 
725
        self.vfs_transport_factory = memory.MemoryServer
664
726
 
665
727
    def test_make_bzrdir_preserves_transport(self):
666
728
        t = self.get_transport()
667
729
        result_bzrdir = self.make_bzrdir('subdir')
668
 
        self.assertIsInstance(result_bzrdir.transport, 
669
 
                              MemoryTransport)
 
730
        self.assertIsInstance(result_bzrdir.transport,
 
731
                              memory.MemoryTransport)
670
732
        # should not be on disk, should only be in memory
671
 
        self.failIfExists('subdir')
672
 
 
673
 
 
674
 
class TestChrootedTest(ChrootedTestCase):
 
733
        self.assertPathDoesNotExist('subdir')
 
734
 
 
735
 
 
736
class TestChrootedTest(tests.ChrootedTestCase):
675
737
 
676
738
    def test_root_is_root(self):
677
 
        from bzrlib.transport import get_transport
678
 
        t = get_transport(self.get_readonly_url())
 
739
        t = transport.get_transport_from_url(self.get_readonly_url())
679
740
        url = t.base
680
741
        self.assertEqual(url, t.clone('..').base)
681
742
 
682
743
 
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):
 
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):
707
765
 
708
766
    def check_timing(self, test_case, expected_re):
709
767
        result = bzrlib.tests.TextTestResult(self._log_file,
710
768
                descriptions=0,
711
769
                verbosity=1,
712
770
                )
713
 
        test_case.run(result)
714
 
        timed_string = result._testTimeString(test_case)
 
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)
715
775
        self.assertContainsRe(timed_string, expected_re)
716
776
 
717
777
    def test_test_reporting(self):
718
 
        class ShortDelayTestCase(TestCase):
 
778
        class ShortDelayTestCase(tests.TestCase):
719
779
            def test_short_delay(self):
720
780
                time.sleep(0.003)
721
781
            def test_short_benchmark(self):
722
782
                self.time(time.sleep, 0.003)
723
783
        self.check_timing(ShortDelayTestCase('test_short_delay'),
724
784
                          r"^ +[0-9]+ms$")
725
 
        # if a benchmark time is given, we want a x of y style result.
 
785
        # if a benchmark time is given, we now show just that time followed by
 
786
        # a star
726
787
        self.check_timing(ShortDelayTestCase('test_short_benchmark'),
727
 
                          r"^ +[0-9]+ms/ +[0-9]+ms$")
 
788
                          r"^ +[0-9]+ms\*$")
728
789
 
729
790
    def test_unittest_reporting_unittest_class(self):
730
791
        # getting the time from a non-bzrlib test works ok
733
794
                time.sleep(0.003)
734
795
        self.check_timing(ShortDelayTestCase('test_short_delay'),
735
796
                          r"^ +[0-9]+ms$")
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
 
 
 
797
 
770
798
    def _time_hello_world_encoding(self):
771
799
        """Profile two sleep calls
772
 
        
 
800
 
773
801
        This is used to exercise the test framework.
774
802
        """
775
803
        self.time(unicode, 'hello', errors='replace')
777
805
 
778
806
    def test_lsprofiling(self):
779
807
        """Verbose test result prints lsprof statistics from test cases."""
780
 
        self.requireFeature(test_lsprof.LSProfFeature)
 
808
        self.requireFeature(features.lsprof_feature)
781
809
        result_stream = StringIO()
782
810
        result = bzrlib.tests.VerboseTestResult(
783
 
            unittest._WritelnDecorator(result_stream),
 
811
            result_stream,
784
812
            descriptions=0,
785
813
            verbosity=2,
786
814
            )
793
821
        # execute the test, which should succeed and record profiles
794
822
        example_test_case.run(result)
795
823
        # lsprofile_something()
796
 
        # if this worked we want 
 
824
        # if this worked we want
797
825
        # LSProf output for <built in function unicode> (['hello'], {'errors': 'replace'})
798
826
        #    CallCount    Recursive    Total(ms)   Inline(ms) module:lineno(function)
799
827
        # (the lsprof header)
800
828
        # ... an arbitrary number of lines
801
829
        # and the function call which is time.sleep.
802
 
        #           1        0            ???         ???       ???(sleep) 
 
830
        #           1        0            ???         ???       ???(sleep)
803
831
        # and then repeated but with 'world', rather than 'hello'.
804
832
        # this should appear in the output stream of our test result.
805
833
        output = result_stream.getvalue()
812
840
        self.assertContainsRe(output,
813
841
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
814
842
 
 
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
 
815
858
    def test_known_failure(self):
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
 
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'
822
865
        result = InstrumentedTestResult(None, None, None, None)
823
 
        def test_function():
824
 
            raise KnownFailure('failed!')
825
 
        test = unittest.FunctionTestCase(test_function)
 
866
        class Test(tests.TestCase):
 
867
            def test_function(self):
 
868
                self.knownFailure('failed!')
 
869
        test = Test("test_function")
826
870
        test.run(result)
827
871
        # it should invoke 'report_known_failure'.
828
872
        self.assertEqual(2, len(result._call))
829
 
        self.assertEqual(test, result._call[0])
830
 
        self.assertEqual(KnownFailure, result._call[1][0])
831
 
        self.assertIsInstance(result._call[1][1], KnownFailure)
 
873
        self.assertEqual(test.id(), result._call[0].id())
 
874
        self.assertEqual('known failure', result._call[1])
832
875
        # we dont introspec the traceback, if the rest is ok, it would be
833
876
        # exceptional for it not to be.
834
877
        # it should update the known_failure_count on the object.
840
883
        # verbose test output formatting
841
884
        result_stream = StringIO()
842
885
        result = bzrlib.tests.VerboseTestResult(
843
 
            unittest._WritelnDecorator(result_stream),
 
886
            result_stream,
844
887
            descriptions=0,
845
888
            verbosity=2,
846
889
            )
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:])
 
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)")
896
897
 
897
898
    def get_passing_test(self):
898
899
        """Return a test object that can't be run usefully."""
902
903
 
903
904
    def test_add_not_supported(self):
904
905
        """Test the behaviour of invoking addNotSupported."""
905
 
        class InstrumentedTestResult(ExtendedTestResult):
906
 
            def report_test_start(self, test): pass
 
906
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
907
            def stopTestRun(self): pass
 
908
            def report_tests_starting(self): pass
907
909
            def report_unsupported(self, test, feature):
908
910
                self._call = test, feature
909
911
        result = InstrumentedTestResult(None, None, None, None)
910
912
        test = SampleTestCase('_test_pass')
911
 
        feature = Feature()
 
913
        feature = features.Feature()
912
914
        result.startTest(test)
913
915
        result.addNotSupported(test, feature)
914
916
        # it should invoke 'report_unsupported'.
928
930
        # verbose test output formatting
929
931
        result_stream = StringIO()
930
932
        result = bzrlib.tests.VerboseTestResult(
931
 
            unittest._WritelnDecorator(result_stream),
 
933
            result_stream,
932
934
            descriptions=0,
933
935
            verbosity=2,
934
936
            )
935
937
        test = self.get_passing_test()
936
 
        feature = Feature()
 
938
        feature = features.Feature()
937
939
        result.startTest(test)
938
940
        prefix = len(result_stream.getvalue())
939
941
        result.report_unsupported(test, feature)
940
942
        output = result_stream.getvalue()[prefix:]
941
943
        lines = output.splitlines()
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
 
    
 
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
 
973
949
    def test_unavailable_exception(self):
974
950
        """An UnavailableFeature being raised should invoke addNotSupported."""
975
 
        class InstrumentedTestResult(ExtendedTestResult):
976
 
 
977
 
            def report_test_start(self, test): pass
 
951
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
952
            def stopTestRun(self): pass
 
953
            def report_tests_starting(self): pass
978
954
            def addNotSupported(self, test, feature):
979
955
                self._call = test, feature
980
956
        result = InstrumentedTestResult(None, None, None, None)
981
 
        feature = Feature()
982
 
        def test_function():
983
 
            raise UnavailableFeature(feature)
984
 
        test = unittest.FunctionTestCase(test_function)
 
957
        feature = features.Feature()
 
958
        class Test(tests.TestCase):
 
959
            def test_function(self):
 
960
                raise tests.UnavailableFeature(feature)
 
961
        test = Test("test_function")
985
962
        test.run(result)
986
963
        # it should invoke 'addNotSupported'.
987
964
        self.assertEqual(2, len(result._call))
988
 
        self.assertEqual(test, result._call[0])
 
965
        self.assertEqual(test.id(), result._call[0].id())
989
966
        self.assertEqual(feature, result._call[1])
990
967
        # and not count as an error
991
968
        self.assertEqual(0, result.error_count)
998
975
        result.addNotSupported(test, feature)
999
976
        self.assertFalse(result.wasStrictlySuccessful())
1000
977
        self.assertEqual(None, result._extractBenchmarkTime(test))
1001
 
 
 
978
 
1002
979
    def test_strict_with_known_failure(self):
1003
980
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
1004
981
                                             verbosity=1)
1005
 
        test = self.get_passing_test()
1006
 
        err = (KnownFailure, KnownFailure('foo'), None)
1007
 
        result._addKnownFailure(test, err)
 
982
        test = _get_test("test_xfail")
 
983
        test.run(result)
1008
984
        self.assertFalse(result.wasStrictlySuccessful())
1009
985
        self.assertEqual(None, result._extractBenchmarkTime(test))
1010
986
 
1016
992
        self.assertTrue(result.wasStrictlySuccessful())
1017
993
        self.assertEqual(None, result._extractBenchmarkTime(test))
1018
994
 
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):
 
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):
1030
1022
 
1031
1023
    def dummy_test(self):
1032
1024
        pass
1036
1028
 
1037
1029
        This current saves and restores:
1038
1030
        TestCaseInTempDir.TEST_ROOT
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.
 
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.
1042
1035
        """
1043
 
        old_root = TestCaseInTempDir.TEST_ROOT
 
1036
        old_root = tests.TestCaseInTempDir.TEST_ROOT
1044
1037
        try:
1045
 
            TestCaseInTempDir.TEST_ROOT = None
 
1038
            tests.TestCaseInTempDir.TEST_ROOT = None
1046
1039
            return testrunner.run(test)
1047
1040
        finally:
1048
 
            TestCaseInTempDir.TEST_ROOT = old_root
 
1041
            tests.TestCaseInTempDir.TEST_ROOT = old_root
1049
1042
 
1050
1043
    def test_known_failure_failed_run(self):
1051
1044
        # run a test that generates a known failure which should be printed in
1052
1045
        # the final output when real failures occur.
1053
 
        def known_failure_test():
1054
 
            raise KnownFailure('failed')
 
1046
        class Test(tests.TestCase):
 
1047
            def known_failure_test(self):
 
1048
                self.expectFailure('failed', self.assertTrue, False)
1055
1049
        test = unittest.TestSuite()
1056
 
        test.addTest(unittest.FunctionTestCase(known_failure_test))
 
1050
        test.addTest(Test("known_failure_test"))
1057
1051
        def failing_test():
1058
1052
            raise AssertionError('foo')
1059
1053
        test.addTest(unittest.FunctionTestCase(failing_test))
1060
1054
        stream = StringIO()
1061
 
        runner = TextTestRunner(stream=stream)
 
1055
        runner = tests.TextTestRunner(stream=stream)
1062
1056
        result = self.run_test_runner(runner, test)
1063
1057
        lines = stream.getvalue().splitlines()
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:])
 
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
            )
1077
1072
 
1078
1073
    def test_known_failure_ok_run(self):
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)
 
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")
1083
1080
        stream = StringIO()
1084
 
        runner = TextTestRunner(stream=stream)
 
1081
        runner = tests.TextTestRunner(stream=stream)
1085
1082
        result = self.run_test_runner(runner, test)
1086
1083
        self.assertContainsRe(stream.getvalue(),
1087
1084
            '\n'
1090
1087
            '\n'
1091
1088
            'OK \\(known_failures=1\\)\n')
1092
1089
 
 
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
 
1093
1125
    def test_skipped_test(self):
1094
1126
        # run a test that is skipped, and check the suite as a whole still
1095
1127
        # succeeds.
1096
1128
        # skipping_test must be hidden in here so it's not run as a real 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)
 
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")
1102
1134
        result = self.run_test_runner(runner, test)
1103
1135
        self.assertTrue(result.wasSuccessful())
1104
1136
 
1105
1137
    def test_skipped_from_setup(self):
1106
1138
        calls = []
1107
 
        class SkippedSetupTest(TestCase):
 
1139
        class SkippedSetupTest(tests.TestCase):
1108
1140
 
1109
1141
            def setUp(self):
1110
1142
                calls.append('setUp')
1111
1143
                self.addCleanup(self.cleanup)
1112
 
                raise TestSkipped('skipped setup')
 
1144
                raise tests.TestSkipped('skipped setup')
1113
1145
 
1114
1146
            def test_skip(self):
1115
1147
                self.fail('test reached')
1117
1149
            def cleanup(self):
1118
1150
                calls.append('cleanup')
1119
1151
 
1120
 
        runner = TextTestRunner(stream=self._log_file)
 
1152
        runner = tests.TextTestRunner(stream=self._log_file)
1121
1153
        test = SkippedSetupTest('test_skip')
1122
1154
        result = self.run_test_runner(runner, test)
1123
1155
        self.assertTrue(result.wasSuccessful())
1126
1158
 
1127
1159
    def test_skipped_from_test(self):
1128
1160
        calls = []
1129
 
        class SkippedTest(TestCase):
 
1161
        class SkippedTest(tests.TestCase):
1130
1162
 
1131
1163
            def setUp(self):
 
1164
                super(SkippedTest, self).setUp()
1132
1165
                calls.append('setUp')
1133
1166
                self.addCleanup(self.cleanup)
1134
1167
 
1135
1168
            def test_skip(self):
1136
 
                raise TestSkipped('skipped test')
 
1169
                raise tests.TestSkipped('skipped test')
1137
1170
 
1138
1171
            def cleanup(self):
1139
1172
                calls.append('cleanup')
1140
1173
 
1141
 
        runner = TextTestRunner(stream=self._log_file)
 
1174
        runner = tests.TextTestRunner(stream=self._log_file)
1142
1175
        test = SkippedTest('test_skip')
1143
1176
        result = self.run_test_runner(runner, test)
1144
1177
        self.assertTrue(result.wasSuccessful())
1147
1180
 
1148
1181
    def test_not_applicable(self):
1149
1182
        # run a test that is skipped because it's not applicable
1150
 
        def not_applicable_test():
1151
 
            from bzrlib.tests import TestNotApplicable
1152
 
            raise TestNotApplicable('this test never runs')
 
1183
        class Test(tests.TestCase):
 
1184
            def not_applicable_test(self):
 
1185
                raise tests.TestNotApplicable('this test never runs')
1153
1186
        out = StringIO()
1154
 
        runner = TextTestRunner(stream=out, verbosity=2)
1155
 
        test = unittest.FunctionTestCase(not_applicable_test)
 
1187
        runner = tests.TextTestRunner(stream=out, verbosity=2)
 
1188
        test = Test("not_applicable_test")
1156
1189
        result = self.run_test_runner(runner, test)
1157
1190
        self._log_file.write(out.getvalue())
1158
1191
        self.assertTrue(result.wasSuccessful())
1162
1195
        self.assertContainsRe(out.getvalue(),
1163
1196
                r'(?m)^    this test never runs')
1164
1197
 
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
 
 
1169
1198
    def test_unsupported_features_listed(self):
1170
1199
        """When unsupported features are encountered they are detailed."""
1171
 
        class Feature1(Feature):
 
1200
        class Feature1(features.Feature):
1172
1201
            def _probe(self): return False
1173
 
        class Feature2(Feature):
 
1202
        class Feature2(features.Feature):
1174
1203
            def _probe(self): return False
1175
1204
        # create sample tests
1176
1205
        test1 = SampleTestCase('_test_pass')
1181
1210
        test.addTest(test1)
1182
1211
        test.addTest(test2)
1183
1212
        stream = StringIO()
1184
 
        runner = TextTestRunner(stream=stream)
 
1213
        runner = tests.TextTestRunner(stream=stream)
1185
1214
        result = self.run_test_runner(runner, test)
1186
1215
        lines = stream.getvalue().splitlines()
1187
1216
        self.assertEqual([
1191
1220
            ],
1192
1221
            lines[-3:])
1193
1222
 
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):
 
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):
1324
1282
 
1325
1283
    def _test_pass(self):
1326
1284
        pass
1328
1286
class _TestException(Exception):
1329
1287
    pass
1330
1288
 
1331
 
class TestTestCase(TestCase):
 
1289
 
 
1290
class TestTestCase(tests.TestCase):
1332
1291
    """Tests that test the core bzrlib TestCase."""
1333
1292
 
 
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
 
1334
1336
    def test_debug_flags_sanitised(self):
1335
1337
        """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')
1336
1341
        # we could set something and run a test that will check
1337
1342
        # it gets santised, but this is probably sufficient for now:
1338
1343
        # if someone runs the test with -Dsomething it will error.
1339
 
        self.assertEqual(set(), bzrlib.debug.debug_flags)
 
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)
1340
1418
 
1341
1419
    def inner_test(self):
1342
1420
        # the inner child test
1346
1424
        # the outer child test
1347
1425
        note("outer_start")
1348
1426
        self.inner_test = TestTestCase("inner_child")
1349
 
        result = bzrlib.tests.TextTestResult(self._log_file,
1350
 
                                        descriptions=0,
1351
 
                                        verbosity=1)
 
1427
        result = self.make_test_result()
1352
1428
        self.inner_test.run(result)
1353
1429
        note("outer finish")
 
1430
        self.addCleanup(osutils.delete_any, self._log_file_name)
1354
1431
 
1355
1432
    def test_trace_nesting(self):
1356
1433
        # this tests that each test case nests its trace facility correctly.
1358
1435
        # should setup a new log, log content to it, setup a child case (B),
1359
1436
        # which should log independently, then case (A) should log a trailer
1360
1437
        # and return.
1361
 
        # we do two nested children so that we can verify the state of the 
 
1438
        # we do two nested children so that we can verify the state of the
1362
1439
        # logs after the outer child finishes is correct, which a bad clean
1363
1440
        # up routine in tearDown might trigger a fault in our test with only
1364
1441
        # one child, we should instead see the bad result inside our test with
1366
1443
        # the outer child test
1367
1444
        original_trace = bzrlib.trace._trace_file
1368
1445
        outer_test = TestTestCase("outer_child")
1369
 
        result = bzrlib.tests.TextTestResult(self._log_file,
1370
 
                                        descriptions=0,
1371
 
                                        verbosity=1)
 
1446
        result = self.make_test_result()
1372
1447
        outer_test.run(result)
1373
1448
        self.assertEqual(original_trace, bzrlib.trace._trace_file)
1374
1449
 
1382
1457
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1383
1458
        output_stream = StringIO()
1384
1459
        result = bzrlib.tests.VerboseTestResult(
1385
 
            unittest._WritelnDecorator(output_stream),
 
1460
            output_stream,
1386
1461
            descriptions=0,
1387
 
            verbosity=2,
1388
 
            num_tests=sample_test.countTestCases())
 
1462
            verbosity=2)
1389
1463
        sample_test.run(result)
1390
1464
        self.assertContainsRe(
1391
1465
            output_stream.getvalue(),
1392
 
            r"\d+ms/ +\d+ms\n$")
 
1466
            r"\d+ms\*\n$")
1393
1467
 
1394
1468
    def test_hooks_sanitised(self):
1395
1469
        """The bzrlib hooks should be sanitised by setUp."""
1396
 
        self.assertEqual(bzrlib.branch.BranchHooks(),
1397
 
            bzrlib.branch.Branch.hooks)
1398
 
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
 
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(),
1399
1476
            bzrlib.smart.server.SmartTCPServer.hooks)
 
1477
        self.assertEqual(
 
1478
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
1400
1479
 
1401
1480
    def test__gather_lsprof_in_benchmarks(self):
1402
1481
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1403
 
        
 
1482
 
1404
1483
        Each self.time() call is individually and separately profiled.
1405
1484
        """
1406
 
        self.requireFeature(test_lsprof.LSProfFeature)
1407
 
        # overrides the class member with an instance member so no cleanup 
 
1485
        self.requireFeature(features.lsprof_feature)
 
1486
        # overrides the class member with an instance member so no cleanup
1408
1487
        # needed.
1409
1488
        self._gather_lsprof_in_benchmarks = True
1410
1489
        self.time(time.sleep, 0.000)
1414
1493
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1415
1494
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1416
1495
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
 
1496
        del self._benchcalls[:]
1417
1497
 
1418
1498
    def test_knownFailure(self):
1419
1499
        """Self.knownFailure() should raise a KnownFailure exception."""
1420
 
        self.assertRaises(KnownFailure, self.knownFailure, "A Failure")
 
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)
1421
1518
 
1422
1519
    def test_requireFeature_available(self):
1423
1520
        """self.requireFeature(available) is a no-op."""
1424
 
        class Available(Feature):
 
1521
        class Available(features.Feature):
1425
1522
            def _probe(self):return True
1426
1523
        feature = Available()
1427
1524
        self.requireFeature(feature)
1428
1525
 
1429
1526
    def test_requireFeature_unavailable(self):
1430
1527
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1431
 
        class Unavailable(Feature):
 
1528
        class Unavailable(features.Feature):
1432
1529
            def _probe(self):return False
1433
1530
        feature = Unavailable()
1434
 
        self.assertRaises(UnavailableFeature, self.requireFeature, feature)
 
1531
        self.assertRaises(tests.UnavailableFeature,
 
1532
                          self.requireFeature, feature)
1435
1533
 
1436
1534
    def test_run_no_parameters(self):
1437
1535
        test = SampleTestCase('_test_pass')
1438
1536
        test.run()
1439
 
    
 
1537
 
1440
1538
    def test_run_enabled_unittest_result(self):
1441
1539
        """Test we revert to regular behaviour when the test is enabled."""
1442
1540
        test = SampleTestCase('_test_pass')
1467
1565
        """Test disabled tests behaviour with support aware results."""
1468
1566
        test = SampleTestCase('_test_pass')
1469
1567
        class DisabledFeature(object):
 
1568
            def __eq__(self, other):
 
1569
                return isinstance(other, DisabledFeature)
1470
1570
            def available(self):
1471
1571
                return False
1472
1572
        the_feature = DisabledFeature()
1483
1583
                self.calls.append(('addNotSupported', test, feature))
1484
1584
        result = InstrumentedTestResult()
1485
1585
        test.run(result)
 
1586
        case = result.calls[0][1]
1486
1587
        self.assertEqual([
1487
 
            ('startTest', test),
1488
 
            ('addNotSupported', test, the_feature),
1489
 
            ('stopTest', test),
 
1588
            ('startTest', case),
 
1589
            ('addNotSupported', case, the_feature),
 
1590
            ('stopTest', case),
1490
1591
            ],
1491
1592
            result.calls)
1492
1593
 
 
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
 
1493
1602
    def test_assert_list_raises_on_generator(self):
1494
1603
        def generator_which_will_raise():
1495
1604
            # This will not raise until after the first yield
1545
1654
        self.assertRaises(AssertionError,
1546
1655
            self.assertListRaises, _TestException, success_generator)
1547
1656
 
1548
 
 
1549
 
@symbol_versioning.deprecated_function(zero_eleven)
 
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)))
1550
1883
def sample_deprecated_function():
1551
1884
    """A deprecated function to test applyDeprecated with."""
1552
1885
    return 2
1559
1892
class ApplyDeprecatedHelper(object):
1560
1893
    """A helper class for ApplyDeprecated tests."""
1561
1894
 
1562
 
    @symbol_versioning.deprecated_method(zero_eleven)
 
1895
    @deprecated_method(deprecated_in((0, 11, 0)))
1563
1896
    def sample_deprecated_method(self, param_one):
1564
1897
        """A deprecated method for testing with."""
1565
1898
        return param_one
1567
1900
    def sample_normal_method(self):
1568
1901
        """A undeprecated method."""
1569
1902
 
1570
 
    @symbol_versioning.deprecated_method(zero_ten)
 
1903
    @deprecated_method(deprecated_in((0, 10, 0)))
1571
1904
    def sample_nested_deprecation(self):
1572
1905
        return sample_deprecated_function()
1573
1906
 
1574
1907
 
1575
 
class TestExtraAssertions(TestCase):
 
1908
class TestExtraAssertions(tests.TestCase):
1576
1909
    """Tests for new test assertions in bzrlib test suite"""
1577
1910
 
1578
1911
    def test_assert_isinstance(self):
1579
1912
        self.assertIsInstance(2, int)
1580
1913
        self.assertIsInstance(u'', basestring)
1581
 
        self.assertRaises(AssertionError, self.assertIsInstance, None, int)
 
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'>")
1582
1917
        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")
1583
1923
 
1584
1924
    def test_assertEndsWith(self):
1585
1925
        self.assertEndsWith('foo', 'oo')
1586
1926
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1587
1927
 
 
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
 
1588
1943
    def test_applyDeprecated_not_deprecated(self):
1589
1944
        sample_object = ApplyDeprecatedHelper()
1590
1945
        # calling an undeprecated callable raises an assertion
1591
 
        self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
 
1946
        self.assertRaises(AssertionError, self.applyDeprecated,
 
1947
            deprecated_in((0, 11, 0)),
1592
1948
            sample_object.sample_normal_method)
1593
 
        self.assertRaises(AssertionError, self.applyDeprecated, zero_eleven,
 
1949
        self.assertRaises(AssertionError, self.applyDeprecated,
 
1950
            deprecated_in((0, 11, 0)),
1594
1951
            sample_undeprecated_function, "a param value")
1595
1952
        # calling a deprecated callable (function or method) with the wrong
1596
1953
        # expected deprecation fails.
1597
 
        self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
 
1954
        self.assertRaises(AssertionError, self.applyDeprecated,
 
1955
            deprecated_in((0, 10, 0)),
1598
1956
            sample_object.sample_deprecated_method, "a param value")
1599
 
        self.assertRaises(AssertionError, self.applyDeprecated, zero_ten,
 
1957
        self.assertRaises(AssertionError, self.applyDeprecated,
 
1958
            deprecated_in((0, 10, 0)),
1600
1959
            sample_deprecated_function)
1601
1960
        # calling a deprecated callable (function or method) with the right
1602
1961
        # expected deprecation returns the functions result.
1603
 
        self.assertEqual("a param value", self.applyDeprecated(zero_eleven,
 
1962
        self.assertEqual("a param value",
 
1963
            self.applyDeprecated(deprecated_in((0, 11, 0)),
1604
1964
            sample_object.sample_deprecated_method, "a param value"))
1605
 
        self.assertEqual(2, self.applyDeprecated(zero_eleven,
 
1965
        self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 11, 0)),
1606
1966
            sample_deprecated_function))
1607
1967
        # calling a nested deprecation with the wrong deprecation version
1608
 
        # fails even if a deeper nested function was deprecated with the 
 
1968
        # fails even if a deeper nested function was deprecated with the
1609
1969
        # supplied version.
1610
1970
        self.assertRaises(AssertionError, self.applyDeprecated,
1611
 
            zero_eleven, sample_object.sample_nested_deprecation)
 
1971
            deprecated_in((0, 11, 0)), sample_object.sample_nested_deprecation)
1612
1972
        # calling a nested deprecation with the right deprecation value
1613
1973
        # returns the calls result.
1614
 
        self.assertEqual(2, self.applyDeprecated(zero_ten,
 
1974
        self.assertEqual(2, self.applyDeprecated(deprecated_in((0, 10, 0)),
1615
1975
            sample_object.sample_nested_deprecation))
1616
1976
 
1617
1977
    def test_callDeprecated(self):
1618
1978
        def testfunc(be_deprecated, result=None):
1619
1979
            if be_deprecated is True:
1620
 
                symbol_versioning.warn('i am deprecated', DeprecationWarning, 
 
1980
                symbol_versioning.warn('i am deprecated', DeprecationWarning,
1621
1981
                                       stacklevel=1)
1622
1982
            return result
1623
1983
        result = self.callDeprecated(['i am deprecated'], testfunc, True)
1628
1988
        self.callDeprecated([], testfunc, be_deprecated=False)
1629
1989
 
1630
1990
 
1631
 
class TestWarningTests(TestCase):
 
1991
class TestWarningTests(tests.TestCase):
1632
1992
    """Tests for calling methods that raise warnings."""
1633
1993
 
1634
1994
    def test_callCatchWarnings(self):
1636
1996
            warnings.warn("this is your last warning")
1637
1997
            return a + b
1638
1998
        wlist, result = self.callCatchWarnings(meth, 1, 2)
1639
 
        self.assertEquals(3, result)
 
1999
        self.assertEqual(3, result)
1640
2000
        # would like just to compare them, but UserWarning doesn't implement
1641
2001
        # eq well
1642
2002
        w0, = wlist
1643
2003
        self.assertIsInstance(w0, UserWarning)
1644
 
        self.assertEquals("this is your last warning", str(w0))
1645
 
 
1646
 
 
1647
 
class TestConvenienceMakers(TestCaseWithTransport):
 
2004
        self.assertEqual("this is your last warning", str(w0))
 
2005
 
 
2006
 
 
2007
class TestConvenienceMakers(tests.TestCaseWithTransport):
1648
2008
    """Test for the make_* convenience functions."""
1649
2009
 
1650
2010
    def test_make_branch_and_tree_with_format(self):
1651
2011
        # we should be able to supply a format to make_branch_and_tree
1652
2012
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
1653
 
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
1654
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
 
2013
        self.assertIsInstance(bzrlib.controldir.ControlDir.open('a')._format,
1655
2014
                              bzrlib.bzrdir.BzrDirMetaFormat1)
1656
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
1657
 
                              bzrlib.bzrdir.BzrDirFormat6)
1658
2015
 
1659
2016
    def test_make_branch_and_memory_tree(self):
1660
2017
        # we should be able to get a new branch and a mutable tree from
1662
2019
        tree = self.make_branch_and_memory_tree('a')
1663
2020
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1664
2021
 
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
 
 
 
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://'))
1671
2028
        tree = self.make_branch_and_tree('t1')
1672
2029
        base = tree.bzrdir.root_transport.base
1673
 
        self.failIf(base.startswith('sftp'),
1674
 
                'base %r is on sftp but should be local' % base)
1675
 
        self.assertEquals(tree.bzrdir.root_transport,
 
2030
        self.assertStartsWith(base, 'file://')
 
2031
        self.assertEqual(tree.bzrdir.root_transport,
1676
2032
                tree.branch.bzrdir.root_transport)
1677
 
        self.assertEquals(tree.bzrdir.root_transport,
 
2033
        self.assertEqual(tree.bzrdir.root_transport,
1678
2034
                tree.branch.repository.bzrdir.root_transport)
1679
2035
 
1680
2036
 
1681
 
class TestSelftest(TestCase):
 
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):
1682
2055
    """Tests of bzrlib.tests.selftest."""
1683
2056
 
1684
2057
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
1685
2058
        factory_called = []
1686
2059
        def factory():
1687
2060
            factory_called.append(True)
1688
 
            return TestSuite()
 
2061
            return TestUtil.TestSuite()
1689
2062
        out = StringIO()
1690
2063
        err = StringIO()
1691
 
        self.apply_redirected(out, err, None, bzrlib.tests.selftest, 
 
2064
        self.apply_redirected(out, err, None, bzrlib.tests.selftest,
1692
2065
            test_suite_factory=factory)
1693
2066
        self.assertEqual([True], factory_called)
1694
2067
 
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])
 
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
1714
2483
        else:
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):
 
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):
1760
2669
 
1761
2670
    def setUp(self):
 
2671
        super(TestSelftestFiltering, self).setUp()
1762
2672
        self.suite = TestUtil.TestSuite()
1763
2673
        self.loader = TestUtil.TestLoader()
1764
 
        self.suite.addTest(self.loader.loadTestsFromModuleNames([
1765
 
            'bzrlib.tests.test_selftest']))
 
2674
        self.suite.addTest(self.loader.loadTestsFromModule(
 
2675
            sys.modules['bzrlib.tests.test_selftest']))
1766
2676
        self.all_names = _test_ids(self.suite)
1767
2677
 
1768
2678
    def test_condition_id_re(self):
1769
2679
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
1770
2680
            'test_condition_id_re')
1771
 
        filtered_suite = filter_suite_by_condition(self.suite,
1772
 
            condition_id_re('test_condition_id_re'))
 
2681
        filtered_suite = tests.filter_suite_by_condition(
 
2682
            self.suite, tests.condition_id_re('test_condition_id_re'))
1773
2683
        self.assertEqual([test_name], _test_ids(filtered_suite))
1774
2684
 
1775
2685
    def test_condition_id_in_list(self):
1776
2686
        test_names = ['bzrlib.tests.test_selftest.TestSelftestFiltering.'
1777
2687
                      'test_condition_id_in_list']
1778
2688
        id_list = tests.TestIdList(test_names)
1779
 
        filtered_suite = filter_suite_by_condition(
 
2689
        filtered_suite = tests.filter_suite_by_condition(
1780
2690
            self.suite, tests.condition_id_in_list(id_list))
1781
2691
        my_pattern = 'TestSelftestFiltering.*test_condition_id_in_list'
1782
 
        re_filtered = filter_suite_by_re(self.suite, my_pattern)
 
2692
        re_filtered = tests.filter_suite_by_re(self.suite, my_pattern)
1783
2693
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
1784
2694
 
1785
2695
    def test_condition_id_startswith(self):
1789
2699
        test_names = [ klass + 'test_condition_id_in_list',
1790
2700
                      klass + 'test_condition_id_startswith',
1791
2701
                     ]
1792
 
        filtered_suite = filter_suite_by_condition(
 
2702
        filtered_suite = tests.filter_suite_by_condition(
1793
2703
            self.suite, tests.condition_id_startswith([start1, start2]))
1794
2704
        self.assertEqual(test_names, _test_ids(filtered_suite))
1795
2705
 
1796
2706
    def test_condition_isinstance(self):
1797
 
        filtered_suite = filter_suite_by_condition(self.suite,
1798
 
            condition_isinstance(self.__class__))
 
2707
        filtered_suite = tests.filter_suite_by_condition(
 
2708
            self.suite, tests.condition_isinstance(self.__class__))
1799
2709
        class_pattern = 'bzrlib.tests.test_selftest.TestSelftestFiltering.'
1800
 
        re_filtered = filter_suite_by_re(self.suite, class_pattern)
 
2710
        re_filtered = tests.filter_suite_by_re(self.suite, class_pattern)
1801
2711
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
1802
2712
 
1803
2713
    def test_exclude_tests_by_condition(self):
1804
2714
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
1805
2715
            'test_exclude_tests_by_condition')
1806
 
        filtered_suite = exclude_tests_by_condition(self.suite,
 
2716
        filtered_suite = tests.exclude_tests_by_condition(self.suite,
1807
2717
            lambda x:x.id() == excluded_name)
1808
2718
        self.assertEqual(len(self.all_names) - 1,
1809
2719
            filtered_suite.countTestCases())
1814
2724
 
1815
2725
    def test_exclude_tests_by_re(self):
1816
2726
        self.all_names = _test_ids(self.suite)
1817
 
        filtered_suite = exclude_tests_by_re(self.suite, 'exclude_tests_by_re')
 
2727
        filtered_suite = tests.exclude_tests_by_re(self.suite,
 
2728
                                                   'exclude_tests_by_re')
1818
2729
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
1819
2730
            'test_exclude_tests_by_re')
1820
2731
        self.assertEqual(len(self.all_names) - 1,
1827
2738
    def test_filter_suite_by_condition(self):
1828
2739
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
1829
2740
            'test_filter_suite_by_condition')
1830
 
        filtered_suite = filter_suite_by_condition(self.suite,
 
2741
        filtered_suite = tests.filter_suite_by_condition(self.suite,
1831
2742
            lambda x:x.id() == test_name)
1832
2743
        self.assertEqual([test_name], _test_ids(filtered_suite))
1833
2744
 
1834
2745
    def test_filter_suite_by_re(self):
1835
 
        filtered_suite = filter_suite_by_re(self.suite, 'test_filter_suite_by_r')
 
2746
        filtered_suite = tests.filter_suite_by_re(self.suite,
 
2747
                                                  'test_filter_suite_by_r')
1836
2748
        filtered_names = _test_ids(filtered_suite)
1837
2749
        self.assertEqual(filtered_names, ['bzrlib.tests.test_selftest.'
1838
2750
            'TestSelftestFiltering.test_filter_suite_by_re'])
1866
2778
 
1867
2779
    def test_preserve_input(self):
1868
2780
        # NB: Surely this is something in the stdlib to do this?
1869
 
        self.assertTrue(self.suite is preserve_input(self.suite))
1870
 
        self.assertTrue("@#$" is preserve_input("@#$"))
 
2781
        self.assertTrue(self.suite is tests.preserve_input(self.suite))
 
2782
        self.assertTrue("@#$" is tests.preserve_input("@#$"))
1871
2783
 
1872
2784
    def test_randomize_suite(self):
1873
 
        randomized_suite = randomize_suite(self.suite)
 
2785
        randomized_suite = tests.randomize_suite(self.suite)
1874
2786
        # randomizing should not add or remove test names.
1875
2787
        self.assertEqual(set(_test_ids(self.suite)),
1876
2788
                         set(_test_ids(randomized_suite)))
1886
2798
 
1887
2799
    def test_split_suit_by_condition(self):
1888
2800
        self.all_names = _test_ids(self.suite)
1889
 
        condition = condition_id_re('test_filter_suite_by_r')
1890
 
        split_suite = split_suite_by_condition(self.suite, condition)
 
2801
        condition = tests.condition_id_re('test_filter_suite_by_r')
 
2802
        split_suite = tests.split_suite_by_condition(self.suite, condition)
1891
2803
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
1892
2804
            'test_filter_suite_by_re')
1893
2805
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
1898
2810
 
1899
2811
    def test_split_suit_by_re(self):
1900
2812
        self.all_names = _test_ids(self.suite)
1901
 
        split_suite = split_suite_by_re(self.suite, 'test_filter_suite_by_r')
 
2813
        split_suite = tests.split_suite_by_re(self.suite,
 
2814
                                              'test_filter_suite_by_r')
1902
2815
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
1903
2816
            'test_filter_suite_by_re')
1904
2817
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
1908
2821
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
1909
2822
 
1910
2823
 
1911
 
class TestCheckInventoryShape(TestCaseWithTransport):
 
2824
class TestCheckTreeShape(tests.TestCaseWithTransport):
1912
2825
 
1913
 
    def test_check_inventory_shape(self):
 
2826
    def test_check_tree_shape(self):
1914
2827
        files = ['a', 'b/', 'b/c']
1915
2828
        tree = self.make_branch_and_tree('.')
1916
2829
        self.build_tree(files)
1917
2830
        tree.add(files)
1918
2831
        tree.lock_read()
1919
2832
        try:
1920
 
            self.check_inventory_shape(tree.inventory, files)
 
2833
            self.check_tree_shape(tree, files)
1921
2834
        finally:
1922
2835
            tree.unlock()
1923
2836
 
1924
2837
 
1925
 
class TestBlackboxSupport(TestCase):
 
2838
class TestBlackboxSupport(tests.TestCase):
1926
2839
    """Tests for testsuite blackbox features."""
1927
2840
 
1928
2841
    def test_run_bzr_failure_not_caught(self):
1934
2847
            self.run_bzr, ['assert-fail'])
1935
2848
        # make sure we got the real thing, not an error from somewhere else in
1936
2849
        # the test framework
1937
 
        self.assertEquals('always fails', str(e))
 
2850
        self.assertEqual('always fails', str(e))
1938
2851
        # check that there's no traceback in the test log
1939
 
        self.assertNotContainsRe(self._get_log(keep_log_file=True),
1940
 
            r'Traceback')
 
2852
        self.assertNotContainsRe(self.get_log(), r'Traceback')
1941
2853
 
1942
2854
    def test_run_bzr_user_error_caught(self):
1943
2855
        # Running bzr in blackbox mode, normal/expected/user errors should be
1944
2856
        # caught in the regular way and turned into an error message plus exit
1945
2857
        # code.
1946
 
        out, err = self.run_bzr(["log", "/nonexistantpath"], retcode=3)
 
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)
1947
2864
        self.assertEqual(out, '')
1948
2865
        self.assertContainsRe(err,
1949
2866
            'bzr: ERROR: Not a branch: ".*nonexistantpath/".\n')
1950
2867
 
1951
2868
 
1952
 
class TestTestLoader(TestCase):
 
2869
class TestTestLoader(tests.TestCase):
1953
2870
    """Tests for the test loader."""
1954
2871
 
1955
2872
    def _get_loader_and_module(self):
1956
2873
        """Gets a TestLoader and a module with one test in it."""
1957
2874
        loader = TestUtil.TestLoader()
1958
2875
        module = {}
1959
 
        class Stub(TestCase):
 
2876
        class Stub(tests.TestCase):
1960
2877
            def test_foo(self):
1961
2878
                pass
1962
2879
        class MyModule(object):
1975
2892
        # load_tests do not need that :)
1976
2893
        def load_tests(self, standard_tests, module, loader):
1977
2894
            result = loader.suiteClass()
1978
 
            for test in iter_suite_tests(standard_tests):
 
2895
            for test in tests.iter_suite_tests(standard_tests):
1979
2896
                result.addTests([test, test])
1980
2897
            return result
1981
2898
        # add a load_tests() method which multiplies the tests from the module.
1985
2902
    def test_load_tests_from_module_name_smoke_test(self):
1986
2903
        loader = TestUtil.TestLoader()
1987
2904
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
1988
 
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
 
2905
        self.assertEqual(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
1989
2906
                          _test_ids(suite))
1990
2907
 
1991
2908
    def test_load_tests_from_module_name_with_bogus_module_name(self):
2000
2917
 
2001
2918
    def _create_suite(self, test_id_list):
2002
2919
 
2003
 
        class Stub(TestCase):
 
2920
        class Stub(tests.TestCase):
2004
2921
            def test_foo(self):
2005
2922
                pass
2006
2923
 
2016
2933
 
2017
2934
    def _test_ids(self, test_suite):
2018
2935
        """Get the ids for the tests in a test suite."""
2019
 
        return [t.id() for t in iter_suite_tests(test_suite)]
 
2936
        return [t.id() for t in tests.iter_suite_tests(test_suite)]
2020
2937
 
2021
2938
    def test_empty_list(self):
2022
2939
        id_list = self._create_id_list([])
2023
 
        self.assertEquals({}, id_list.tests)
2024
 
        self.assertEquals({}, id_list.modules)
 
2940
        self.assertEqual({}, id_list.tests)
 
2941
        self.assertEqual({}, id_list.modules)
2025
2942
 
2026
2943
    def test_valid_list(self):
2027
2944
        id_list = self._create_id_list(
2048
2965
        self.assertTrue(id_list.refers_to('mod.class'))
2049
2966
        self.assertTrue(id_list.refers_to('mod.class.meth'))
2050
2967
 
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
 
 
2069
2968
    def test_test_suite_matches_id_list_with_unknown(self):
2070
2969
        loader = TestUtil.TestLoader()
2071
2970
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2072
2971
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',
2073
2972
                     'bogus']
2074
2973
        not_found, duplicates = tests.suite_matches_id_list(suite, test_list)
2075
 
        self.assertEquals(['bogus'], not_found)
2076
 
        self.assertEquals([], duplicates)
 
2974
        self.assertEqual(['bogus'], not_found)
 
2975
        self.assertEqual([], duplicates)
2077
2976
 
2078
2977
    def test_suite_matches_id_list_with_duplicates(self):
2079
2978
        loader = TestUtil.TestLoader()
2080
2979
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2081
2980
        dupes = loader.suiteClass()
2082
 
        for test in iter_suite_tests(suite):
 
2981
        for test in tests.iter_suite_tests(suite):
2083
2982
            dupes.addTest(test)
2084
2983
            dupes.addTest(test) # Add it again
2085
2984
 
2086
2985
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing',]
2087
2986
        not_found, duplicates = tests.suite_matches_id_list(
2088
2987
            dupes, test_list)
2089
 
        self.assertEquals([], not_found)
2090
 
        self.assertEquals(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
 
2988
        self.assertEqual([], not_found)
 
2989
        self.assertEqual(['bzrlib.tests.test_sampler.DemoTest.test_nothing'],
2091
2990
                          duplicates)
2092
2991
 
2093
2992
 
 
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
 
2094
3071
class TestLoadTestIdList(tests.TestCaseInTempDir):
2095
3072
 
2096
3073
    def _create_test_list_file(self, file_name, content):
2107
3084
        self._create_test_list_file(test_list_fname,
2108
3085
                                    'mod1.cl1.meth1\nmod2.cl2.meth2\n')
2109
3086
        tlist = tests.load_test_id_list(test_list_fname)
2110
 
        self.assertEquals(2, len(tlist))
2111
 
        self.assertEquals('mod1.cl1.meth1', tlist[0])
2112
 
        self.assertEquals('mod2.cl2.meth2', tlist[1])
 
3087
        self.assertEqual(2, len(tlist))
 
3088
        self.assertEqual('mod1.cl1.meth1', tlist[0])
 
3089
        self.assertEqual('mod2.cl2.meth2', tlist[1])
2113
3090
 
2114
3091
    def test_load_dirty_file(self):
2115
3092
        test_list_fname = 'test.list'
2117
3094
                                    '  mod1.cl1.meth1\n\nmod2.cl2.meth2  \n'
2118
3095
                                    'bar baz\n')
2119
3096
        tlist = tests.load_test_id_list(test_list_fname)
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])
 
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])
2125
3102
 
2126
3103
 
2127
3104
class TestFilteredByModuleTestLoader(tests.TestCase):
2134
3111
    def test_load_tests(self):
2135
3112
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
2136
3113
        loader = self._create_loader(test_list)
2137
 
 
2138
3114
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2139
 
        self.assertEquals(test_list, _test_ids(suite))
 
3115
        self.assertEqual(test_list, _test_ids(suite))
2140
3116
 
2141
3117
    def test_exclude_tests(self):
2142
3118
        test_list = ['bogus']
2143
3119
        loader = self._create_loader(test_list)
2144
 
 
2145
3120
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2146
 
        self.assertEquals([], _test_ids(suite))
 
3121
        self.assertEqual([], _test_ids(suite))
2147
3122
 
2148
3123
 
2149
3124
class TestFilteredByNameStartTestLoader(tests.TestCase):
2159
3134
        loader = self._create_loader('bzrlib.tests.test_samp')
2160
3135
 
2161
3136
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2162
 
        self.assertEquals(test_list, _test_ids(suite))
 
3137
        self.assertEqual(test_list, _test_ids(suite))
2163
3138
 
2164
3139
    def test_load_tests_inside_module(self):
2165
3140
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
2166
3141
        loader = self._create_loader('bzrlib.tests.test_sampler.Demo')
2167
3142
 
2168
3143
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2169
 
        self.assertEquals(test_list, _test_ids(suite))
 
3144
        self.assertEqual(test_list, _test_ids(suite))
2170
3145
 
2171
3146
    def test_exclude_tests(self):
2172
3147
        test_list = ['bogus']
2173
3148
        loader = self._create_loader('bogus')
2174
3149
 
2175
3150
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2176
 
        self.assertEquals([], _test_ids(suite))
 
3151
        self.assertEqual([], _test_ids(suite))
2177
3152
 
2178
3153
 
2179
3154
class TestTestPrefixRegistry(tests.TestCase):
2185
3160
    def test_register_new_prefix(self):
2186
3161
        tpr = self._get_registry()
2187
3162
        tpr.register('foo', 'fff.ooo.ooo')
2188
 
        self.assertEquals('fff.ooo.ooo', tpr.get('foo'))
 
3163
        self.assertEqual('fff.ooo.ooo', tpr.get('foo'))
2189
3164
 
2190
3165
    def test_register_existing_prefix(self):
2191
3166
        tpr = self._get_registry()
2192
3167
        tpr.register('bar', 'bbb.aaa.rrr')
2193
3168
        tpr.register('bar', 'bBB.aAA.rRR')
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')
 
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))
2197
3173
 
2198
3174
    def test_get_unknown_prefix(self):
2199
3175
        tpr = self._get_registry()
2202
3178
    def test_resolve_prefix(self):
2203
3179
        tpr = self._get_registry()
2204
3180
        tpr.register('bar', 'bb.aa.rr')
2205
 
        self.assertEquals('bb.aa.rr', tpr.resolve_alias('bar'))
 
3181
        self.assertEqual('bb.aa.rr', tpr.resolve_alias('bar'))
2206
3182
 
2207
3183
    def test_resolve_unknown_alias(self):
2208
3184
        tpr = self._get_registry()
2211
3187
 
2212
3188
    def test_predefined_prefixes(self):
2213
3189
        tpr = tests.test_prefix_alias_registry
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'))
 
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')