~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Vincent Ladeuil
  • Date: 2010-02-09 17:27:46 UTC
  • mto: (5029.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5030.
  • Revision ID: v.ladeuil+lp@free.fr-20100209172746-6f4mvlnr2mac807j
Move NoSmartTransportServer to bzrlib.tests.test_server

Show diffs side-by-side

added added

removed removed

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