~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

Major code cleanup.

Show diffs side-by-side

added added

removed removed

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