~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Ian Clatworthy
  • Date: 2007-08-13 14:33:10 UTC
  • mto: (2733.1.1 ianc-integration)
  • mto: This revision was merged to the branch mainline in revision 2734.
  • Revision ID: ian.clatworthy@internode.on.net-20070813143310-twhj4la0qnupvze8
Added Quick Start Summary

Show diffs side-by-side

added added

removed removed

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