~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2011-08-17 18:13:57 UTC
  • mfrom: (5268.7.29 transport-segments)
  • Revision ID: pqm@pqm.ubuntu.com-20110817181357-y5q5eth1hk8bl3om
(jelmer) Allow specifying the colocated branch to use in the branch URL,
 and retrieving the branch name using ControlDir._get_selected_branch.
 (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

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