~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: John Arbash Meinel
  • Date: 2006-08-18 16:48:53 UTC
  • mto: (1946.2.6 reduce-knit-churn)
  • mto: This revision was merged to the branch mainline in revision 1948.
  • Revision ID: john@arbash-meinel.com-20060818164853-eb25e4db3385e216
Add a couple more sftp benchmarks

Show diffs side-by-side

added added

removed removed

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