/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Robert Collins
  • Date: 2007-04-19 02:27:44 UTC
  • mto: This revision was merged to the branch mainline in revision 2426.
  • Revision ID: robertc@robertcollins.net-20070419022744-pfdqz42kp1wizh43
``make docs`` now creates a man page at ``man1/bzr.1`` fixing bug 107388.
(Robert Collins)

Show diffs side-by-side

added added

removed removed

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