~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-02-02 10:38:47 UTC
  • mfrom: (4996.2.1 integration)
  • Revision ID: pqm@pqm.ubuntu.com-20100202103847-rqhr25k4s3r5wfer
merge 2.1 to trunk, including fix for bug 514090

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