~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

cleanup of globster code.

Show diffs side-by-side

added added

removed removed

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