/brz/remove-bazaar

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

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: John Arbash Meinel
  • Date: 2009-10-17 04:43:14 UTC
  • mto: This revision was merged to the branch mainline in revision 4756.
  • Revision ID: john@arbash-meinel.com-20091017044314-nlvrrqnz0f2wzcp4
change the GroupcompressBlock code a bit.
If the first decompress request is big enough, just decompress everything.
And when we do that, let go of the decompressobj.

After digging through the zlib code, it looks like 1 zlib stream object
contains a 5kB internal state, and another 4*64kB buffers. (about 260kB
of total state.)
That turns out to be quite a lot if you think about it.


In the case of branching a copy of 'bzr.dev' locally, this turned out
to be 383MB w/ bzr.dev and 345MB w/ only this patch. (So ~11% of peak).

Also, this was 'unreferenced' memory, because it is hidden in the
zlib internal state in working buffers. So it wasn't memory that Meliae
could find. \o/.

Show diffs side-by-side

added added

removed removed

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