~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Robert Collins
  • Date: 2009-03-27 04:10:25 UTC
  • mfrom: (4208 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4216.
  • Revision ID: robertc@robertcollins.net-20090327041025-rgutx4q03xo4pq6l
Resolve NEWS conflicts.

Show diffs side-by-side

added added

removed removed

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