~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Martin Pool
  • Date: 2005-08-18 05:52:29 UTC
  • Revision ID: mbp@sourcefrog.net-20050818055229-cac46ebce364d04c
- avoid compiling REs at module load time

Show diffs side-by-side

added added

removed removed

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