~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Patch Queue Manager
  • Date: 2012-09-17 11:14:25 UTC
  • mfrom: (6554.1.1 stack-remove-unknown)
  • Revision ID: pqm@pqm.ubuntu.com-20120917111425-4i6r0ze0v9zm33bu
(vila) Fix obscure and misleading warning when trying to delete an unknown
 config option. (Vincent Ladeuil)

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