~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Vincent Ladeuil
  • Date: 2010-10-26 08:08:23 UTC
  • mfrom: (5514.1.1 665100-content-type)
  • mto: This revision was merged to the branch mainline in revision 5516.
  • Revision ID: v.ladeuil+lp@free.fr-20101026080823-3wggo03b7cpn9908
Correctly set the Content-Type header when POSTing http requests

Show diffs side-by-side

added added

removed removed

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