~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Martin Packman
  • Date: 2011-12-23 19:38:22 UTC
  • mto: This revision was merged to the branch mainline in revision 6405.
  • Revision ID: martin.packman@canonical.com-20111223193822-hesheea4o8aqwexv
Accept and document passing the medium rather than transport for smart connections

Show diffs side-by-side

added added

removed removed

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