~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-06-18 20:25:52 UTC
  • mfrom: (4413.5.15 1.16-chk-direct)
  • Revision ID: pqm@pqm.ubuntu.com-20090618202552-xyl6tcvbxtm8bupf
(jam) Improve initial commit performance by creating a CHKMap in bulk,
        rather than via O(tree) map() calls.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2010 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
 
from doctest import ELLIPSIS
 
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 MultiTestResult
30
 
from testtools.content import Content
31
 
from testtools.content_type import ContentType
32
 
from testtools.matchers import (
33
 
    DocTestMatches,
34
 
    Equals,
35
 
    )
36
 
import testtools.tests.helpers
37
 
 
38
27
import bzrlib
39
28
from bzrlib import (
40
29
    branchbuilder,
48
37
    repository,
49
38
    symbol_versioning,
50
39
    tests,
51
 
    transport,
52
40
    workingtree,
53
41
    )
 
42
from bzrlib.progress import _BaseProgressBar
54
43
from bzrlib.repofmt import (
55
 
    groupcompress_repo,
56
44
    pack_repo,
57
45
    weaverepo,
58
46
    )
62
50
    deprecated_method,
63
51
    )
64
52
from bzrlib.tests import (
65
 
    features,
66
 
    test_lsprof,
67
 
    test_server,
68
 
    test_sftp_transport,
69
 
    TestUtil,
70
 
    )
71
 
from bzrlib.trace import note, mutter
72
 
from bzrlib.transport import memory
 
53
                          ChrootedTestCase,
 
54
                          ExtendedTestResult,
 
55
                          Feature,
 
56
                          KnownFailure,
 
57
                          TestCase,
 
58
                          TestCaseInTempDir,
 
59
                          TestCaseWithMemoryTransport,
 
60
                          TestCaseWithTransport,
 
61
                          TestNotApplicable,
 
62
                          TestSkipped,
 
63
                          TestSuite,
 
64
                          TestUtil,
 
65
                          TextTestRunner,
 
66
                          UnavailableFeature,
 
67
                          condition_id_re,
 
68
                          condition_isinstance,
 
69
                          exclude_tests_by_condition,
 
70
                          exclude_tests_by_re,
 
71
                          filter_suite_by_condition,
 
72
                          filter_suite_by_re,
 
73
                          iter_suite_tests,
 
74
                          preserve_input,
 
75
                          randomize_suite,
 
76
                          run_suite,
 
77
                          split_suite_by_condition,
 
78
                          split_suite_by_re,
 
79
                          test_lsprof,
 
80
                          test_suite,
 
81
                          )
 
82
from bzrlib.tests.test_sftp_transport import TestCaseWithSFTPServer
 
83
from bzrlib.tests.TestUtil import _load_module_by_name
 
84
from bzrlib.trace import note
 
85
from bzrlib.transport.memory import MemoryServer, MemoryTransport
73
86
from bzrlib.version import _get_bzr_source_tree
74
87
 
75
88
 
76
89
def _test_ids(test_suite):
77
90
    """Get the ids for the tests in a test suite."""
78
 
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
79
 
 
80
 
 
81
 
class SelftestTests(tests.TestCase):
 
91
    return [t.id() for t in iter_suite_tests(test_suite)]
 
92
 
 
93
 
 
94
class SelftestTests(TestCase):
82
95
 
83
96
    def test_import_tests(self):
84
 
        mod = TestUtil._load_module_by_name('bzrlib.tests.test_selftest')
 
97
        mod = _load_module_by_name('bzrlib.tests.test_selftest')
85
98
        self.assertEqual(mod.SelftestTests, SelftestTests)
86
99
 
87
100
    def test_import_test_failure(self):
88
101
        self.assertRaises(ImportError,
89
 
                          TestUtil._load_module_by_name,
 
102
                          _load_module_by_name,
90
103
                          'bzrlib.no-name-yet')
91
104
 
92
 
 
93
 
class MetaTestLog(tests.TestCase):
 
105
class MetaTestLog(TestCase):
94
106
 
95
107
    def test_logging(self):
96
108
        """Test logs are captured when a test fails."""
97
109
        self.log('a test message')
98
 
        details = self.getDetails()
99
 
        log = details['log']
100
 
        self.assertThat(log.content_type, Equals(ContentType(
101
 
            "text", "plain", {"charset": "utf8"})))
102
 
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
103
 
        self.assertThat(self.get_log(),
104
 
            DocTestMatches(u"...a test message\n", ELLIPSIS))
105
 
 
106
 
 
107
 
class TestUnicodeFilename(tests.TestCase):
 
110
        self._log_file.flush()
 
111
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
112
                              'a test message\n')
 
113
 
 
114
 
 
115
class TestUnicodeFilename(TestCase):
108
116
 
109
117
    def test_probe_passes(self):
110
118
        """UnicodeFilename._probe passes."""
113
121
        tests.UnicodeFilename._probe()
114
122
 
115
123
 
116
 
class TestTreeShape(tests.TestCaseInTempDir):
 
124
class TestTreeShape(TestCaseInTempDir):
117
125
 
118
126
    def test_unicode_paths(self):
119
127
        self.requireFeature(tests.UnicodeFilename)
123
131
        self.failUnlessExists(filename)
124
132
 
125
133
 
126
 
class TestClassesAvailable(tests.TestCase):
127
 
    """As a convenience we expose Test* classes from bzrlib.tests"""
128
 
 
129
 
    def test_test_case(self):
130
 
        from bzrlib.tests import TestCase
131
 
 
132
 
    def test_test_loader(self):
133
 
        from bzrlib.tests import TestLoader
134
 
 
135
 
    def test_test_suite(self):
136
 
        from bzrlib.tests import TestSuite
137
 
 
138
 
 
139
 
class TestTransportScenarios(tests.TestCase):
 
134
class TestTransportScenarios(TestCase):
140
135
    """A group of tests that test the transport implementation adaption core.
141
136
 
142
137
    This is a meta test that the tests are applied to all available
153
148
            def get_test_permutations(self):
154
149
                return sample_permutation
155
150
        sample_permutation = [(1,2), (3,4)]
156
 
        from bzrlib.tests.per_transport import get_transport_test_permutations
 
151
        from bzrlib.tests.test_transport_implementations \
 
152
            import get_transport_test_permutations
157
153
        self.assertEqual(sample_permutation,
158
154
                         get_transport_test_permutations(MockModule()))
159
155
 
160
 
    def test_scenarios_include_all_modules(self):
 
156
    def test_scenarios_invlude_all_modules(self):
161
157
        # this checks that the scenario generator returns as many permutations
162
158
        # as there are in all the registered transport modules - we assume if
163
159
        # this matches its probably doing the right thing especially in
164
160
        # combination with the tests for setting the right classes below.
165
 
        from bzrlib.tests.per_transport import transport_test_permutations
 
161
        from bzrlib.tests.test_transport_implementations \
 
162
            import transport_test_permutations
166
163
        from bzrlib.transport import _get_transport_modules
167
164
        modules = _get_transport_modules()
168
165
        permutation_count = 0
180
177
        # This test used to know about all the possible transports and the
181
178
        # order they were returned but that seems overly brittle (mbp
182
179
        # 20060307)
183
 
        from bzrlib.tests.per_transport import transport_test_permutations
 
180
        from bzrlib.tests.test_transport_implementations \
 
181
            import transport_test_permutations
184
182
        scenarios = transport_test_permutations()
185
183
        # there are at least that many builtin transports
186
184
        self.assertTrue(len(scenarios) > 6)
192
190
                                   bzrlib.transport.Server))
193
191
 
194
192
 
195
 
class TestBranchScenarios(tests.TestCase):
 
193
class TestBranchScenarios(TestCase):
196
194
 
197
195
    def test_scenarios(self):
198
196
        # check that constructor parameters are passed through to the adapted
199
197
        # test.
200
 
        from bzrlib.tests.per_branch import make_scenarios
 
198
        from bzrlib.tests.branch_implementations import make_scenarios
201
199
        server1 = "a"
202
200
        server2 = "b"
203
201
        formats = [("c", "C"), ("d", "D")]
217
215
            scenarios)
218
216
 
219
217
 
220
 
class TestBzrDirScenarios(tests.TestCase):
 
218
class TestBzrDirScenarios(TestCase):
221
219
 
222
220
    def test_scenarios(self):
223
221
        # check that constructor parameters are passed through to the adapted
224
222
        # test.
225
 
        from bzrlib.tests.per_controldir import make_scenarios
 
223
        from bzrlib.tests.bzrdir_implementations import make_scenarios
226
224
        vfs_factory = "v"
227
225
        server1 = "a"
228
226
        server2 = "b"
242
240
            scenarios)
243
241
 
244
242
 
245
 
class TestRepositoryScenarios(tests.TestCase):
 
243
class TestRepositoryScenarios(TestCase):
246
244
 
247
245
    def test_formats_to_scenarios(self):
248
246
        from bzrlib.tests.per_repository import formats_to_scenarios
249
247
        formats = [("(c)", remote.RemoteRepositoryFormat()),
250
248
                   ("(d)", repository.format_registry.get(
251
 
                    'Bazaar repository format 2a (needs bzr 1.16 or later)\n'))]
 
249
                        'Bazaar pack repository format 1 (needs bzr 0.92)\n'))]
252
250
        no_vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
253
251
            None)
254
252
        vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
255
253
            vfs_transport_factory="vfs")
256
254
        # no_vfs generate scenarios without vfs_transport_factory
257
 
        expected = [
 
255
        self.assertEqual([
258
256
            ('RemoteRepositoryFormat(c)',
259
257
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
260
258
              'repository_format': remote.RemoteRepositoryFormat(),
261
259
              'transport_readonly_server': 'readonly',
262
260
              'transport_server': 'server'}),
263
 
            ('RepositoryFormat2a(d)',
 
261
            ('RepositoryFormatKnitPack1(d)',
264
262
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
265
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
263
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
266
264
              'transport_readonly_server': 'readonly',
267
 
              'transport_server': 'server'})]
268
 
        self.assertEqual(expected, no_vfs_scenarios)
 
265
              'transport_server': 'server'})],
 
266
            no_vfs_scenarios)
269
267
        self.assertEqual([
270
268
            ('RemoteRepositoryFormat(c)',
271
269
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
273
271
              'transport_readonly_server': 'readonly',
274
272
              'transport_server': 'server',
275
273
              'vfs_transport_factory': 'vfs'}),
276
 
            ('RepositoryFormat2a(d)',
 
274
            ('RepositoryFormatKnitPack1(d)',
277
275
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
278
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
276
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
279
277
              'transport_readonly_server': 'readonly',
280
278
              'transport_server': 'server',
281
279
              'vfs_transport_factory': 'vfs'})],
282
280
            vfs_scenarios)
283
281
 
284
282
 
285
 
class TestTestScenarioApplication(tests.TestCase):
 
283
class TestTestScenarioApplication(TestCase):
286
284
    """Tests for the test adaption facilities."""
287
285
 
288
286
    def test_apply_scenario(self):
318
316
            adapted_test2.id())
319
317
 
320
318
 
321
 
class TestInterRepositoryScenarios(tests.TestCase):
 
319
class TestInterRepositoryScenarios(TestCase):
322
320
 
323
321
    def test_scenarios(self):
324
322
        # check that constructor parameters are passed through to the adapted
325
323
        # test.
326
 
        from bzrlib.tests.per_interrepository import make_scenarios
 
324
        from bzrlib.tests.interrepository_implementations import \
 
325
            make_scenarios
327
326
        server1 = "a"
328
327
        server2 = "b"
329
 
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
 
328
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
330
329
        scenarios = make_scenarios(server1, server2, formats)
331
330
        self.assertEqual([
332
 
            ('C0,str,str',
333
 
             {'repository_format': 'C1',
 
331
            ('str,str,str',
 
332
             {'interrepo_class': str,
 
333
              'repository_format': 'C1',
334
334
              'repository_format_to': 'C2',
335
335
              'transport_readonly_server': 'b',
336
 
              'transport_server': 'a',
337
 
              'extra_setup': 'C3'}),
338
 
            ('D0,str,str',
339
 
             {'repository_format': 'D1',
 
336
              'transport_server': 'a'}),
 
337
            ('int,str,str',
 
338
             {'interrepo_class': int,
 
339
              'repository_format': 'D1',
340
340
              'repository_format_to': 'D2',
341
341
              'transport_readonly_server': 'b',
342
 
              'transport_server': 'a',
343
 
              'extra_setup': 'D3'})],
 
342
              'transport_server': 'a'})],
344
343
            scenarios)
345
344
 
346
345
 
347
 
class TestWorkingTreeScenarios(tests.TestCase):
 
346
class TestWorkingTreeScenarios(TestCase):
348
347
 
349
348
    def test_scenarios(self):
350
349
        # check that constructor parameters are passed through to the adapted
351
350
        # test.
352
 
        from bzrlib.tests.per_workingtree import make_scenarios
 
351
        from bzrlib.tests.workingtree_implementations \
 
352
            import make_scenarios
353
353
        server1 = "a"
354
354
        server2 = "b"
355
355
        formats = [workingtree.WorkingTreeFormat2(),
369
369
            scenarios)
370
370
 
371
371
 
372
 
class TestTreeScenarios(tests.TestCase):
 
372
class TestTreeScenarios(TestCase):
373
373
 
374
374
    def test_scenarios(self):
375
375
        # the tree implementation scenario generator is meant to setup one
379
379
        # workingtree_to_test_tree attribute set to 'return_parameter' and the
380
380
        # revision one set to revision_tree_from_workingtree.
381
381
 
382
 
        from bzrlib.tests.per_tree import (
 
382
        from bzrlib.tests.tree_implementations import (
383
383
            _dirstate_tree_from_workingtree,
384
384
            make_scenarios,
385
385
            preview_tree_pre,
448
448
        self.assertEqual(expected_scenarios, scenarios)
449
449
 
450
450
 
451
 
class TestInterTreeScenarios(tests.TestCase):
 
451
class TestInterTreeScenarios(TestCase):
452
452
    """A group of tests that test the InterTreeTestAdapter."""
453
453
 
454
454
    def test_scenarios(self):
461
461
        # unlike the TestProviderAdapter we dont want to automatically add a
462
462
        # parameterized one for WorkingTree - the optimisers will tell us what
463
463
        # ones to add.
464
 
        from bzrlib.tests.per_tree import (
 
464
        from bzrlib.tests.tree_implementations import (
465
465
            return_parameter,
466
466
            revision_tree_from_workingtree
467
467
            )
468
 
        from bzrlib.tests.per_intertree import (
 
468
        from bzrlib.tests.intertree_implementations import (
469
469
            make_scenarios,
470
470
            )
471
471
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
504
504
        self.assertEqual(scenarios, expected_scenarios)
505
505
 
506
506
 
507
 
class TestTestCaseInTempDir(tests.TestCaseInTempDir):
 
507
class TestTestCaseInTempDir(TestCaseInTempDir):
508
508
 
509
509
    def test_home_is_not_working(self):
510
510
        self.assertNotEqual(self.test_dir, self.test_home_dir)
521
521
        self.assertEqualStat(real, fake)
522
522
 
523
523
    def test_assertEqualStat_notequal(self):
524
 
        self.build_tree(["foo", "longname"])
 
524
        self.build_tree(["foo", "bar"])
525
525
        self.assertRaises(AssertionError, self.assertEqualStat,
526
 
            os.lstat("foo"), os.lstat("longname"))
527
 
 
528
 
 
529
 
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
 
526
            os.lstat("foo"), os.lstat("bar"))
 
527
 
 
528
 
 
529
class TestTestCaseWithMemoryTransport(TestCaseWithMemoryTransport):
530
530
 
531
531
    def test_home_is_non_existant_dir_under_root(self):
532
532
        """The test_home_dir for TestCaseWithMemoryTransport is missing.
547
547
        cwd = osutils.getcwd()
548
548
        self.assertIsSameRealPath(self.test_dir, cwd)
549
549
 
550
 
    def test_BZR_HOME_and_HOME_are_bytestrings(self):
551
 
        """The $BZR_HOME and $HOME environment variables should not be unicode.
552
 
 
553
 
        See https://bugs.launchpad.net/bzr/+bug/464174
554
 
        """
555
 
        self.assertIsInstance(os.environ['BZR_HOME'], str)
556
 
        self.assertIsInstance(os.environ['HOME'], str)
557
 
 
558
550
    def test_make_branch_and_memory_tree(self):
559
551
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
560
552
 
616
608
                         self.get_transport().get_bytes(
617
609
                            'dir/.bzr/repository/format'))
618
610
 
 
611
    def test_safety_net(self):
 
612
        """No test should modify the safety .bzr directory.
 
613
 
 
614
        We just test that the _check_safety_net private method raises
 
615
        AssertionError, it's easier than building a test suite with the same
 
616
        test.
 
617
        """
 
618
        # Oops, a commit in the current directory (i.e. without local .bzr
 
619
        # directory) will crawl up the hierarchy to find a .bzr directory.
 
620
        self.run_bzr(['commit', '-mfoo', '--unchanged'])
 
621
        # But we have a safety net in place.
 
622
        self.assertRaises(AssertionError, self._check_safety_net)
 
623
 
619
624
    def test_dangling_locks_cause_failures(self):
620
 
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
 
625
        class TestDanglingLock(TestCaseWithMemoryTransport):
621
626
            def test_function(self):
622
627
                t = self.get_transport('.')
623
628
                l = lockdir.LockDir(t, 'lock')
625
630
                l.attempt_lock()
626
631
        test = TestDanglingLock('test_function')
627
632
        result = test.run()
628
 
        total_failures = result.errors + result.failures
629
 
        if self._lock_check_thorough:
630
 
            self.assertEqual(1, len(total_failures))
631
 
        else:
632
 
            # When _lock_check_thorough is disabled, then we don't trigger a
633
 
            # failure
634
 
            self.assertEqual(0, len(total_failures))
635
 
 
636
 
 
637
 
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
 
633
        self.assertEqual(1, len(result.errors))
 
634
 
 
635
 
 
636
class TestTestCaseWithTransport(TestCaseWithTransport):
638
637
    """Tests for the convenience functions TestCaseWithTransport introduces."""
639
638
 
640
639
    def test_get_readonly_url_none(self):
 
640
        from bzrlib.transport import get_transport
 
641
        from bzrlib.transport.memory import MemoryServer
641
642
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
642
 
        self.vfs_transport_factory = memory.MemoryServer
 
643
        self.vfs_transport_factory = MemoryServer
643
644
        self.transport_readonly_server = None
644
645
        # calling get_readonly_transport() constructs a decorator on the url
645
646
        # for the server
646
647
        url = self.get_readonly_url()
647
648
        url2 = self.get_readonly_url('foo/bar')
648
 
        t = transport.get_transport(url)
649
 
        t2 = transport.get_transport(url2)
 
649
        t = get_transport(url)
 
650
        t2 = get_transport(url2)
650
651
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
651
652
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
652
653
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
653
654
 
654
655
    def test_get_readonly_url_http(self):
655
656
        from bzrlib.tests.http_server import HttpServer
 
657
        from bzrlib.transport import get_transport
 
658
        from bzrlib.transport.local import LocalURLServer
656
659
        from bzrlib.transport.http import HttpTransportBase
657
 
        self.transport_server = test_server.LocalURLServer
 
660
        self.transport_server = LocalURLServer
658
661
        self.transport_readonly_server = HttpServer
659
662
        # calling get_readonly_transport() gives us a HTTP server instance.
660
663
        url = self.get_readonly_url()
661
664
        url2 = self.get_readonly_url('foo/bar')
662
665
        # the transport returned may be any HttpTransportBase subclass
663
 
        t = transport.get_transport(url)
664
 
        t2 = transport.get_transport(url2)
 
666
        t = get_transport(url)
 
667
        t2 = get_transport(url2)
665
668
        self.failUnless(isinstance(t, HttpTransportBase))
666
669
        self.failUnless(isinstance(t2, HttpTransportBase))
667
670
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
687
690
        self.assertEqual((1, rev_id), a_branch.last_revision_info())
688
691
 
689
692
 
690
 
class TestTestCaseTransports(tests.TestCaseWithTransport):
 
693
class TestTestCaseTransports(TestCaseWithTransport):
691
694
 
692
695
    def setUp(self):
693
696
        super(TestTestCaseTransports, self).setUp()
694
 
        self.vfs_transport_factory = memory.MemoryServer
 
697
        self.vfs_transport_factory = MemoryServer
695
698
 
696
699
    def test_make_bzrdir_preserves_transport(self):
697
700
        t = self.get_transport()
698
701
        result_bzrdir = self.make_bzrdir('subdir')
699
702
        self.assertIsInstance(result_bzrdir.transport,
700
 
                              memory.MemoryTransport)
 
703
                              MemoryTransport)
701
704
        # should not be on disk, should only be in memory
702
705
        self.failIfExists('subdir')
703
706
 
704
707
 
705
 
class TestChrootedTest(tests.ChrootedTestCase):
 
708
class TestChrootedTest(ChrootedTestCase):
706
709
 
707
710
    def test_root_is_root(self):
708
 
        t = transport.get_transport(self.get_readonly_url())
 
711
        from bzrlib.transport import get_transport
 
712
        t = get_transport(self.get_readonly_url())
709
713
        url = t.base
710
714
        self.assertEqual(url, t.clone('..').base)
711
715
 
712
716
 
713
 
class TestProfileResult(tests.TestCase):
714
 
 
715
 
    def test_profiles_tests(self):
716
 
        self.requireFeature(test_lsprof.LSProfFeature)
717
 
        terminal = testtools.tests.helpers.ExtendedTestResult()
718
 
        result = tests.ProfileResult(terminal)
719
 
        class Sample(tests.TestCase):
720
 
            def a(self):
721
 
                self.sample_function()
722
 
            def sample_function(self):
723
 
                pass
724
 
        test = Sample("a")
725
 
        test.run(result)
726
 
        case = terminal._events[0][1]
727
 
        self.assertLength(1, case._benchcalls)
728
 
        # We must be able to unpack it as the test reporting code wants
729
 
        (_, _, _), stats = case._benchcalls[0]
730
 
        self.assertTrue(callable(stats.pprint))
731
 
 
732
 
 
733
 
class TestTestResult(tests.TestCase):
 
717
class MockProgress(_BaseProgressBar):
 
718
    """Progress-bar standin that records calls.
 
719
 
 
720
    Useful for testing pb using code.
 
721
    """
 
722
 
 
723
    def __init__(self):
 
724
        _BaseProgressBar.__init__(self)
 
725
        self.calls = []
 
726
 
 
727
    def tick(self):
 
728
        self.calls.append(('tick',))
 
729
 
 
730
    def update(self, msg=None, current=None, total=None):
 
731
        self.calls.append(('update', msg, current, total))
 
732
 
 
733
    def clear(self):
 
734
        self.calls.append(('clear',))
 
735
 
 
736
    def note(self, msg, *args):
 
737
        self.calls.append(('note', msg, args))
 
738
 
 
739
 
 
740
class TestTestResult(TestCase):
734
741
 
735
742
    def check_timing(self, test_case, expected_re):
736
743
        result = bzrlib.tests.TextTestResult(self._log_file,
737
744
                descriptions=0,
738
745
                verbosity=1,
739
746
                )
740
 
        capture = testtools.tests.helpers.ExtendedTestResult()
741
 
        test_case.run(MultiTestResult(result, capture))
742
 
        run_case = capture._events[0][1]
743
 
        timed_string = result._testTimeString(run_case)
 
747
        test_case.run(result)
 
748
        timed_string = result._testTimeString(test_case)
744
749
        self.assertContainsRe(timed_string, expected_re)
745
750
 
746
751
    def test_test_reporting(self):
747
 
        class ShortDelayTestCase(tests.TestCase):
 
752
        class ShortDelayTestCase(TestCase):
748
753
            def test_short_delay(self):
749
754
                time.sleep(0.003)
750
755
            def test_short_benchmark(self):
751
756
                self.time(time.sleep, 0.003)
752
757
        self.check_timing(ShortDelayTestCase('test_short_delay'),
753
758
                          r"^ +[0-9]+ms$")
754
 
        # if a benchmark time is given, we now show just that time followed by
755
 
        # a star
 
759
        # if a benchmark time is given, we want a x of y style result.
756
760
        self.check_timing(ShortDelayTestCase('test_short_benchmark'),
757
 
                          r"^ +[0-9]+ms\*$")
 
761
                          r"^ +[0-9]+ms/ +[0-9]+ms$")
758
762
 
759
763
    def test_unittest_reporting_unittest_class(self):
760
764
        # getting the time from a non-bzrlib test works ok
764
768
        self.check_timing(ShortDelayTestCase('test_short_delay'),
765
769
                          r"^ +[0-9]+ms$")
766
770
 
767
 
    def _patch_get_bzr_source_tree(self):
768
 
        # Reading from the actual source tree breaks isolation, but we don't
769
 
        # want to assume that thats *all* that would happen.
770
 
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', lambda: None)
771
 
 
772
771
    def test_assigned_benchmark_file_stores_date(self):
773
 
        self._patch_get_bzr_source_tree()
774
772
        output = StringIO()
775
773
        result = bzrlib.tests.TextTestResult(self._log_file,
776
774
                                        descriptions=0,
784
782
        self.assertContainsRe(output_string, "--date [0-9.]+")
785
783
 
786
784
    def test_benchhistory_records_test_times(self):
787
 
        self._patch_get_bzr_source_tree()
788
785
        result_stream = StringIO()
789
786
        result = bzrlib.tests.TextTestResult(
790
787
            self._log_file,
817
814
        self.requireFeature(test_lsprof.LSProfFeature)
818
815
        result_stream = StringIO()
819
816
        result = bzrlib.tests.VerboseTestResult(
820
 
            result_stream,
 
817
            unittest._WritelnDecorator(result_stream),
821
818
            descriptions=0,
822
819
            verbosity=2,
823
820
            )
851
848
 
852
849
    def test_known_failure(self):
853
850
        """A KnownFailure being raised should trigger several result actions."""
854
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
855
 
            def stopTestRun(self): pass
856
 
            def report_tests_starting(self): pass
857
 
            def report_known_failure(self, test, err=None, details=None):
858
 
                self._call = test, 'known failure'
 
851
        class InstrumentedTestResult(ExtendedTestResult):
 
852
            def done(self): pass
 
853
            def startTests(self): pass
 
854
            def report_test_start(self, test): pass
 
855
            def report_known_failure(self, test, err):
 
856
                self._call = test, err
859
857
        result = InstrumentedTestResult(None, None, None, None)
860
 
        class Test(tests.TestCase):
861
 
            def test_function(self):
862
 
                raise tests.KnownFailure('failed!')
863
 
        test = Test("test_function")
 
858
        def test_function():
 
859
            raise KnownFailure('failed!')
 
860
        test = unittest.FunctionTestCase(test_function)
864
861
        test.run(result)
865
862
        # it should invoke 'report_known_failure'.
866
863
        self.assertEqual(2, len(result._call))
867
 
        self.assertEqual(test.id(), result._call[0].id())
868
 
        self.assertEqual('known failure', result._call[1])
 
864
        self.assertEqual(test, result._call[0])
 
865
        self.assertEqual(KnownFailure, result._call[1][0])
 
866
        self.assertIsInstance(result._call[1][1], KnownFailure)
869
867
        # we dont introspec the traceback, if the rest is ok, it would be
870
868
        # exceptional for it not to be.
871
869
        # it should update the known_failure_count on the object.
877
875
        # verbose test output formatting
878
876
        result_stream = StringIO()
879
877
        result = bzrlib.tests.VerboseTestResult(
880
 
            result_stream,
 
878
            unittest._WritelnDecorator(result_stream),
881
879
            descriptions=0,
882
880
            verbosity=2,
883
881
            )
888
886
        # (class, exception object, traceback)
889
887
        # KnownFailures dont get their tracebacks shown though, so we
890
888
        # can skip that.
891
 
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
 
889
        err = (KnownFailure, KnownFailure('foo'), None)
892
890
        result.report_known_failure(test, err)
893
891
        output = result_stream.getvalue()[prefix:]
894
892
        lines = output.splitlines()
895
893
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
896
 
        if sys.version_info > (2, 7):
897
 
            self.expectFailure("_ExpectedFailure on 2.7 loses the message",
898
 
                self.assertNotEqual, lines[1], '    ')
899
894
        self.assertEqual(lines[1], '    foo')
900
895
        self.assertEqual(2, len(lines))
901
896
 
 
897
    def test_text_report_known_failure(self):
 
898
        # text test output formatting
 
899
        pb = MockProgress()
 
900
        result = bzrlib.tests.TextTestResult(
 
901
            StringIO(),
 
902
            descriptions=0,
 
903
            verbosity=1,
 
904
            pb=pb,
 
905
            )
 
906
        test = self.get_passing_test()
 
907
        # this seeds the state to handle reporting the test.
 
908
        result.startTest(test)
 
909
        # the err parameter has the shape:
 
910
        # (class, exception object, traceback)
 
911
        # KnownFailures dont get their tracebacks shown though, so we
 
912
        # can skip that.
 
913
        err = (KnownFailure, KnownFailure('foo'), None)
 
914
        result.report_known_failure(test, err)
 
915
        self.assertEqual(
 
916
            [
 
917
            ('update', '[1 in 0s] passing_test', None, None),
 
918
            ('note', 'XFAIL: %s\n%s\n', ('passing_test', err[1]))
 
919
            ],
 
920
            pb.calls)
 
921
        # known_failures should be printed in the summary, so if we run a test
 
922
        # after there are some known failures, the update prefix should match
 
923
        # this.
 
924
        result.known_failure_count = 3
 
925
        test.run(result)
 
926
        self.assertEqual(
 
927
            [
 
928
            ('update', '[2 in 0s] passing_test', None, None),
 
929
            ],
 
930
            pb.calls[2:])
 
931
 
902
932
    def get_passing_test(self):
903
933
        """Return a test object that can't be run usefully."""
904
934
        def passing_test():
907
937
 
908
938
    def test_add_not_supported(self):
909
939
        """Test the behaviour of invoking addNotSupported."""
910
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
911
 
            def stopTestRun(self): pass
912
 
            def report_tests_starting(self): pass
 
940
        class InstrumentedTestResult(ExtendedTestResult):
 
941
            def done(self): pass
 
942
            def startTests(self): pass
 
943
            def report_test_start(self, test): pass
913
944
            def report_unsupported(self, test, feature):
914
945
                self._call = test, feature
915
946
        result = InstrumentedTestResult(None, None, None, None)
916
947
        test = SampleTestCase('_test_pass')
917
 
        feature = tests.Feature()
 
948
        feature = Feature()
918
949
        result.startTest(test)
919
950
        result.addNotSupported(test, feature)
920
951
        # it should invoke 'report_unsupported'.
934
965
        # verbose test output formatting
935
966
        result_stream = StringIO()
936
967
        result = bzrlib.tests.VerboseTestResult(
937
 
            result_stream,
 
968
            unittest._WritelnDecorator(result_stream),
938
969
            descriptions=0,
939
970
            verbosity=2,
940
971
            )
941
972
        test = self.get_passing_test()
942
 
        feature = tests.Feature()
 
973
        feature = Feature()
943
974
        result.startTest(test)
944
975
        prefix = len(result_stream.getvalue())
945
976
        result.report_unsupported(test, feature)
946
977
        output = result_stream.getvalue()[prefix:]
947
978
        lines = output.splitlines()
948
 
        # We don't check for the final '0ms' since it may fail on slow hosts
949
 
        self.assertStartsWith(lines[0], 'NODEP')
950
 
        self.assertEqual(lines[1],
951
 
                         "    The feature 'Feature' is not available.")
 
979
        self.assertEqual(lines, ['NODEP                   0ms', "    The feature 'Feature' is not available."])
 
980
 
 
981
    def test_text_report_unsupported(self):
 
982
        # text test output formatting
 
983
        pb = MockProgress()
 
984
        result = bzrlib.tests.TextTestResult(
 
985
            StringIO(),
 
986
            descriptions=0,
 
987
            verbosity=1,
 
988
            pb=pb,
 
989
            )
 
990
        test = self.get_passing_test()
 
991
        feature = Feature()
 
992
        # this seeds the state to handle reporting the test.
 
993
        result.startTest(test)
 
994
        result.report_unsupported(test, feature)
 
995
        # no output on unsupported features
 
996
        self.assertEqual(
 
997
            [('update', '[1 in 0s] passing_test', None, None)
 
998
            ],
 
999
            pb.calls)
 
1000
        # the number of missing features should be printed in the progress
 
1001
        # summary, so check for that.
 
1002
        result.unsupported = {'foo':0, 'bar':0}
 
1003
        test.run(result)
 
1004
        self.assertEqual(
 
1005
            [
 
1006
            ('update', '[2 in 0s, 2 missing] passing_test', None, None),
 
1007
            ],
 
1008
            pb.calls[1:])
952
1009
 
953
1010
    def test_unavailable_exception(self):
954
1011
        """An UnavailableFeature being raised should invoke addNotSupported."""
955
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
956
 
            def stopTestRun(self): pass
957
 
            def report_tests_starting(self): pass
 
1012
        class InstrumentedTestResult(ExtendedTestResult):
 
1013
            def done(self): pass
 
1014
            def startTests(self): pass
 
1015
            def report_test_start(self, test): pass
958
1016
            def addNotSupported(self, test, feature):
959
1017
                self._call = test, feature
960
1018
        result = InstrumentedTestResult(None, None, None, None)
961
 
        feature = tests.Feature()
962
 
        class Test(tests.TestCase):
963
 
            def test_function(self):
964
 
                raise tests.UnavailableFeature(feature)
965
 
        test = Test("test_function")
 
1019
        feature = Feature()
 
1020
        def test_function():
 
1021
            raise UnavailableFeature(feature)
 
1022
        test = unittest.FunctionTestCase(test_function)
966
1023
        test.run(result)
967
1024
        # it should invoke 'addNotSupported'.
968
1025
        self.assertEqual(2, len(result._call))
969
 
        self.assertEqual(test.id(), result._call[0].id())
 
1026
        self.assertEqual(test, result._call[0])
970
1027
        self.assertEqual(feature, result._call[1])
971
1028
        # and not count as an error
972
1029
        self.assertEqual(0, result.error_count)
984
1041
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
985
1042
                                             verbosity=1)
986
1043
        test = self.get_passing_test()
987
 
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
988
 
        result.addExpectedFailure(test, err)
 
1044
        err = (KnownFailure, KnownFailure('foo'), None)
 
1045
        result._addKnownFailure(test, err)
989
1046
        self.assertFalse(result.wasStrictlySuccessful())
990
1047
        self.assertEqual(None, result._extractBenchmarkTime(test))
991
1048
 
999
1056
 
1000
1057
    def test_startTests(self):
1001
1058
        """Starting the first test should trigger startTests."""
1002
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
1059
        class InstrumentedTestResult(ExtendedTestResult):
1003
1060
            calls = 0
1004
1061
            def startTests(self): self.calls += 1
 
1062
            def report_test_start(self, test): pass
1005
1063
        result = InstrumentedTestResult(None, None, None, None)
1006
1064
        def test_function():
1007
1065
            pass
1009
1067
        test.run(result)
1010
1068
        self.assertEquals(1, result.calls)
1011
1069
 
1012
 
    def test_startTests_only_once(self):
1013
 
        """With multiple tests startTests should still only be called once"""
1014
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
1015
 
            calls = 0
1016
 
            def startTests(self): self.calls += 1
1017
 
        result = InstrumentedTestResult(None, None, None, None)
1018
 
        suite = unittest.TestSuite([
1019
 
            unittest.FunctionTestCase(lambda: None),
1020
 
            unittest.FunctionTestCase(lambda: None)])
1021
 
        suite.run(result)
1022
 
        self.assertEquals(1, result.calls)
1023
 
        self.assertEquals(2, result.count)
1024
 
 
1025
 
 
1026
 
class TestUnicodeFilenameFeature(tests.TestCase):
 
1070
 
 
1071
class TestUnicodeFilenameFeature(TestCase):
1027
1072
 
1028
1073
    def test_probe_passes(self):
1029
1074
        """UnicodeFilenameFeature._probe passes."""
1032
1077
        tests.UnicodeFilenameFeature._probe()
1033
1078
 
1034
1079
 
1035
 
class TestRunner(tests.TestCase):
 
1080
class TestRunner(TestCase):
1036
1081
 
1037
1082
    def dummy_test(self):
1038
1083
        pass
1043
1088
        This current saves and restores:
1044
1089
        TestCaseInTempDir.TEST_ROOT
1045
1090
 
1046
 
        There should be no tests in this file that use
1047
 
        bzrlib.tests.TextTestRunner without using this convenience method,
1048
 
        because of our use of global state.
 
1091
        There should be no tests in this file that use bzrlib.tests.TextTestRunner
 
1092
        without using this convenience method, because of our use of global state.
1049
1093
        """
1050
 
        old_root = tests.TestCaseInTempDir.TEST_ROOT
 
1094
        old_root = TestCaseInTempDir.TEST_ROOT
1051
1095
        try:
1052
 
            tests.TestCaseInTempDir.TEST_ROOT = None
 
1096
            TestCaseInTempDir.TEST_ROOT = None
1053
1097
            return testrunner.run(test)
1054
1098
        finally:
1055
 
            tests.TestCaseInTempDir.TEST_ROOT = old_root
 
1099
            TestCaseInTempDir.TEST_ROOT = old_root
1056
1100
 
1057
1101
    def test_known_failure_failed_run(self):
1058
1102
        # run a test that generates a known failure which should be printed in
1059
1103
        # the final output when real failures occur.
1060
 
        class Test(tests.TestCase):
1061
 
            def known_failure_test(self):
1062
 
                self.expectFailure('failed', self.assertTrue, False)
 
1104
        def known_failure_test():
 
1105
            raise KnownFailure('failed')
1063
1106
        test = unittest.TestSuite()
1064
 
        test.addTest(Test("known_failure_test"))
 
1107
        test.addTest(unittest.FunctionTestCase(known_failure_test))
1065
1108
        def failing_test():
1066
 
            self.fail('foo')
 
1109
            raise AssertionError('foo')
1067
1110
        test.addTest(unittest.FunctionTestCase(failing_test))
1068
1111
        stream = StringIO()
1069
 
        runner = tests.TextTestRunner(stream=stream)
 
1112
        runner = TextTestRunner(stream=stream)
1070
1113
        result = self.run_test_runner(runner, test)
1071
1114
        lines = stream.getvalue().splitlines()
1072
 
        self.assertContainsRe(stream.getvalue(),
1073
 
            '(?sm)^bzr selftest.*$'
1074
 
            '.*'
1075
 
            '^======================================================================\n'
1076
 
            '^FAIL: failing_test\n'
1077
 
            '^----------------------------------------------------------------------\n'
1078
 
            'Traceback \\(most recent call last\\):\n'
1079
 
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1080
 
            '    self.fail\\(\'foo\'\\)\n'
1081
 
            '.*'
1082
 
            '^----------------------------------------------------------------------\n'
1083
 
            '.*'
1084
 
            'FAILED \\(failures=1, known_failure_count=1\\)'
1085
 
            )
 
1115
        self.assertEqual([
 
1116
            '',
 
1117
            '======================================================================',
 
1118
            'FAIL: unittest.FunctionTestCase (failing_test)',
 
1119
            '----------------------------------------------------------------------',
 
1120
            'Traceback (most recent call last):',
 
1121
            '    raise AssertionError(\'foo\')',
 
1122
            'AssertionError: foo',
 
1123
            '',
 
1124
            '----------------------------------------------------------------------',
 
1125
            '',
 
1126
            'FAILED (failures=1, known_failure_count=1)'],
 
1127
            lines[3:8] + lines[9:13] + lines[14:])
1086
1128
 
1087
1129
    def test_known_failure_ok_run(self):
1088
 
        # run a test that generates a known failure which should be printed in
1089
 
        # the final output.
1090
 
        class Test(tests.TestCase):
1091
 
            def known_failure_test(self):
1092
 
                self.expectFailure('failed', self.assertTrue, False)
1093
 
        test = Test("known_failure_test")
 
1130
        # run a test that generates a known failure which should be printed in the final output.
 
1131
        def known_failure_test():
 
1132
            raise KnownFailure('failed')
 
1133
        test = unittest.FunctionTestCase(known_failure_test)
1094
1134
        stream = StringIO()
1095
 
        runner = tests.TextTestRunner(stream=stream)
 
1135
        runner = TextTestRunner(stream=stream)
1096
1136
        result = self.run_test_runner(runner, test)
1097
1137
        self.assertContainsRe(stream.getvalue(),
1098
1138
            '\n'
1101
1141
            '\n'
1102
1142
            'OK \\(known_failures=1\\)\n')
1103
1143
 
1104
 
    def test_result_decorator(self):
1105
 
        # decorate results
1106
 
        calls = []
1107
 
        class LoggingDecorator(tests.ForwardingResult):
1108
 
            def startTest(self, test):
1109
 
                tests.ForwardingResult.startTest(self, test)
1110
 
                calls.append('start')
1111
 
        test = unittest.FunctionTestCase(lambda:None)
1112
 
        stream = StringIO()
1113
 
        runner = tests.TextTestRunner(stream=stream,
1114
 
            result_decorators=[LoggingDecorator])
1115
 
        result = self.run_test_runner(runner, test)
1116
 
        self.assertLength(1, calls)
1117
 
 
1118
1144
    def test_skipped_test(self):
1119
1145
        # run a test that is skipped, and check the suite as a whole still
1120
1146
        # succeeds.
1121
1147
        # skipping_test must be hidden in here so it's not run as a real test
1122
 
        class SkippingTest(tests.TestCase):
 
1148
        class SkippingTest(TestCase):
1123
1149
            def skipping_test(self):
1124
 
                raise tests.TestSkipped('test intentionally skipped')
1125
 
        runner = tests.TextTestRunner(stream=self._log_file)
 
1150
                raise TestSkipped('test intentionally skipped')
 
1151
        runner = TextTestRunner(stream=self._log_file)
1126
1152
        test = SkippingTest("skipping_test")
1127
1153
        result = self.run_test_runner(runner, test)
1128
1154
        self.assertTrue(result.wasSuccessful())
1129
1155
 
1130
1156
    def test_skipped_from_setup(self):
1131
1157
        calls = []
1132
 
        class SkippedSetupTest(tests.TestCase):
 
1158
        class SkippedSetupTest(TestCase):
1133
1159
 
1134
1160
            def setUp(self):
1135
1161
                calls.append('setUp')
1136
1162
                self.addCleanup(self.cleanup)
1137
 
                raise tests.TestSkipped('skipped setup')
 
1163
                raise TestSkipped('skipped setup')
1138
1164
 
1139
1165
            def test_skip(self):
1140
1166
                self.fail('test reached')
1142
1168
            def cleanup(self):
1143
1169
                calls.append('cleanup')
1144
1170
 
1145
 
        runner = tests.TextTestRunner(stream=self._log_file)
 
1171
        runner = TextTestRunner(stream=self._log_file)
1146
1172
        test = SkippedSetupTest('test_skip')
1147
1173
        result = self.run_test_runner(runner, test)
1148
1174
        self.assertTrue(result.wasSuccessful())
1151
1177
 
1152
1178
    def test_skipped_from_test(self):
1153
1179
        calls = []
1154
 
        class SkippedTest(tests.TestCase):
 
1180
        class SkippedTest(TestCase):
1155
1181
 
1156
1182
            def setUp(self):
1157
 
                tests.TestCase.setUp(self)
 
1183
                TestCase.setUp(self)
1158
1184
                calls.append('setUp')
1159
1185
                self.addCleanup(self.cleanup)
1160
1186
 
1161
1187
            def test_skip(self):
1162
 
                raise tests.TestSkipped('skipped test')
 
1188
                raise TestSkipped('skipped test')
1163
1189
 
1164
1190
            def cleanup(self):
1165
1191
                calls.append('cleanup')
1166
1192
 
1167
 
        runner = tests.TextTestRunner(stream=self._log_file)
 
1193
        runner = TextTestRunner(stream=self._log_file)
1168
1194
        test = SkippedTest('test_skip')
1169
1195
        result = self.run_test_runner(runner, test)
1170
1196
        self.assertTrue(result.wasSuccessful())
1173
1199
 
1174
1200
    def test_not_applicable(self):
1175
1201
        # run a test that is skipped because it's not applicable
1176
 
        class Test(tests.TestCase):
1177
 
            def not_applicable_test(self):
1178
 
                raise tests.TestNotApplicable('this test never runs')
 
1202
        def not_applicable_test():
 
1203
            from bzrlib.tests import TestNotApplicable
 
1204
            raise TestNotApplicable('this test never runs')
1179
1205
        out = StringIO()
1180
 
        runner = tests.TextTestRunner(stream=out, verbosity=2)
1181
 
        test = Test("not_applicable_test")
 
1206
        runner = TextTestRunner(stream=out, verbosity=2)
 
1207
        test = unittest.FunctionTestCase(not_applicable_test)
1182
1208
        result = self.run_test_runner(runner, test)
1183
1209
        self._log_file.write(out.getvalue())
1184
1210
        self.assertTrue(result.wasSuccessful())
1188
1214
        self.assertContainsRe(out.getvalue(),
1189
1215
                r'(?m)^    this test never runs')
1190
1216
 
 
1217
    def test_not_applicable_demo(self):
 
1218
        # just so you can see it in the test output
 
1219
        raise TestNotApplicable('this test is just a demonstation')
 
1220
 
1191
1221
    def test_unsupported_features_listed(self):
1192
1222
        """When unsupported features are encountered they are detailed."""
1193
 
        class Feature1(tests.Feature):
 
1223
        class Feature1(Feature):
1194
1224
            def _probe(self): return False
1195
 
        class Feature2(tests.Feature):
 
1225
        class Feature2(Feature):
1196
1226
            def _probe(self): return False
1197
1227
        # create sample tests
1198
1228
        test1 = SampleTestCase('_test_pass')
1203
1233
        test.addTest(test1)
1204
1234
        test.addTest(test2)
1205
1235
        stream = StringIO()
1206
 
        runner = tests.TextTestRunner(stream=stream)
 
1236
        runner = TextTestRunner(stream=stream)
1207
1237
        result = self.run_test_runner(runner, test)
1208
1238
        lines = stream.getvalue().splitlines()
1209
1239
        self.assertEqual([
1213
1243
            ],
1214
1244
            lines[-3:])
1215
1245
 
1216
 
    def _patch_get_bzr_source_tree(self):
1217
 
        # Reading from the actual source tree breaks isolation, but we don't
1218
 
        # want to assume that thats *all* that would happen.
1219
 
        self._get_source_tree_calls = []
1220
 
        def new_get():
1221
 
            self._get_source_tree_calls.append("called")
1222
 
            return None
1223
 
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree',  new_get)
1224
 
 
1225
1246
    def test_bench_history(self):
1226
 
        # tests that the running the benchmark passes bench_history into
1227
 
        # the test result object. We can tell that happens if
1228
 
        # _get_bzr_source_tree is called.
1229
 
        self._patch_get_bzr_source_tree()
 
1247
        # tests that the running the benchmark produces a history file
 
1248
        # containing a timestamp and the revision id of the bzrlib source which
 
1249
        # was tested.
 
1250
        workingtree = _get_bzr_source_tree()
1230
1251
        test = TestRunner('dummy_test')
1231
1252
        output = StringIO()
1232
 
        runner = tests.TextTestRunner(stream=self._log_file,
1233
 
                                      bench_history=output)
 
1253
        runner = TextTestRunner(stream=self._log_file, bench_history=output)
1234
1254
        result = self.run_test_runner(runner, test)
1235
1255
        output_string = output.getvalue()
1236
1256
        self.assertContainsRe(output_string, "--date [0-9.]+")
1237
 
        self.assertLength(1, self._get_source_tree_calls)
1238
 
 
1239
 
    def test_verbose_test_count(self):
1240
 
        """A verbose test run reports the right test count at the start"""
1241
 
        suite = TestUtil.TestSuite([
1242
 
            unittest.FunctionTestCase(lambda:None),
1243
 
            unittest.FunctionTestCase(lambda:None)])
1244
 
        self.assertEqual(suite.countTestCases(), 2)
1245
 
        stream = StringIO()
1246
 
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
1247
 
        # Need to use the CountingDecorator as that's what sets num_tests
1248
 
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
1249
 
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
1250
 
 
1251
 
    def test_startTestRun(self):
1252
 
        """run should call result.startTestRun()"""
1253
 
        calls = []
1254
 
        class LoggingDecorator(tests.ForwardingResult):
1255
 
            def startTestRun(self):
1256
 
                tests.ForwardingResult.startTestRun(self)
1257
 
                calls.append('startTestRun')
1258
 
        test = unittest.FunctionTestCase(lambda:None)
1259
 
        stream = StringIO()
1260
 
        runner = tests.TextTestRunner(stream=stream,
1261
 
            result_decorators=[LoggingDecorator])
1262
 
        result = self.run_test_runner(runner, test)
1263
 
        self.assertLength(1, calls)
1264
 
 
1265
 
    def test_stopTestRun(self):
1266
 
        """run should call result.stopTestRun()"""
1267
 
        calls = []
1268
 
        class LoggingDecorator(tests.ForwardingResult):
1269
 
            def stopTestRun(self):
1270
 
                tests.ForwardingResult.stopTestRun(self)
1271
 
                calls.append('stopTestRun')
1272
 
        test = unittest.FunctionTestCase(lambda:None)
1273
 
        stream = StringIO()
1274
 
        runner = tests.TextTestRunner(stream=stream,
1275
 
            result_decorators=[LoggingDecorator])
1276
 
        result = self.run_test_runner(runner, test)
1277
 
        self.assertLength(1, calls)
1278
 
 
1279
 
 
1280
 
class SampleTestCase(tests.TestCase):
 
1257
        if workingtree is not None:
 
1258
            revision_id = workingtree.get_parent_ids()[0]
 
1259
            self.assertEndsWith(output_string.rstrip(), revision_id)
 
1260
 
 
1261
    def assertLogDeleted(self, test):
 
1262
        log = test._get_log()
 
1263
        self.assertEqual("DELETED log file to reduce memory footprint", log)
 
1264
        self.assertEqual('', test._log_contents)
 
1265
        self.assertIs(None, test._log_file_name)
 
1266
 
 
1267
    def test_success_log_deleted(self):
 
1268
        """Successful tests have their log deleted"""
 
1269
 
 
1270
        class LogTester(TestCase):
 
1271
 
 
1272
            def test_success(self):
 
1273
                self.log('this will be removed\n')
 
1274
 
 
1275
        sio = cStringIO.StringIO()
 
1276
        runner = TextTestRunner(stream=sio)
 
1277
        test = LogTester('test_success')
 
1278
        result = self.run_test_runner(runner, test)
 
1279
 
 
1280
        self.assertLogDeleted(test)
 
1281
 
 
1282
    def test_skipped_log_deleted(self):
 
1283
        """Skipped tests have their log deleted"""
 
1284
 
 
1285
        class LogTester(TestCase):
 
1286
 
 
1287
            def test_skipped(self):
 
1288
                self.log('this will be removed\n')
 
1289
                raise tests.TestSkipped()
 
1290
 
 
1291
        sio = cStringIO.StringIO()
 
1292
        runner = TextTestRunner(stream=sio)
 
1293
        test = LogTester('test_skipped')
 
1294
        result = self.run_test_runner(runner, test)
 
1295
 
 
1296
        self.assertLogDeleted(test)
 
1297
 
 
1298
    def test_not_aplicable_log_deleted(self):
 
1299
        """Not applicable tests have their log deleted"""
 
1300
 
 
1301
        class LogTester(TestCase):
 
1302
 
 
1303
            def test_not_applicable(self):
 
1304
                self.log('this will be removed\n')
 
1305
                raise tests.TestNotApplicable()
 
1306
 
 
1307
        sio = cStringIO.StringIO()
 
1308
        runner = TextTestRunner(stream=sio)
 
1309
        test = LogTester('test_not_applicable')
 
1310
        result = self.run_test_runner(runner, test)
 
1311
 
 
1312
        self.assertLogDeleted(test)
 
1313
 
 
1314
    def test_known_failure_log_deleted(self):
 
1315
        """Know failure tests have their log deleted"""
 
1316
 
 
1317
        class LogTester(TestCase):
 
1318
 
 
1319
            def test_known_failure(self):
 
1320
                self.log('this will be removed\n')
 
1321
                raise tests.KnownFailure()
 
1322
 
 
1323
        sio = cStringIO.StringIO()
 
1324
        runner = TextTestRunner(stream=sio)
 
1325
        test = LogTester('test_known_failure')
 
1326
        result = self.run_test_runner(runner, test)
 
1327
 
 
1328
        self.assertLogDeleted(test)
 
1329
 
 
1330
    def test_fail_log_kept(self):
 
1331
        """Failed tests have their log kept"""
 
1332
 
 
1333
        class LogTester(TestCase):
 
1334
 
 
1335
            def test_fail(self):
 
1336
                self.log('this will be kept\n')
 
1337
                self.fail('this test fails')
 
1338
 
 
1339
        sio = cStringIO.StringIO()
 
1340
        runner = TextTestRunner(stream=sio)
 
1341
        test = LogTester('test_fail')
 
1342
        result = self.run_test_runner(runner, test)
 
1343
 
 
1344
        text = sio.getvalue()
 
1345
        self.assertContainsRe(text, 'this will be kept')
 
1346
        self.assertContainsRe(text, 'this test fails')
 
1347
 
 
1348
        log = test._get_log()
 
1349
        self.assertContainsRe(log, 'this will be kept')
 
1350
        self.assertEqual(log, test._log_contents)
 
1351
 
 
1352
    def test_error_log_kept(self):
 
1353
        """Tests with errors have their log kept"""
 
1354
 
 
1355
        class LogTester(TestCase):
 
1356
 
 
1357
            def test_error(self):
 
1358
                self.log('this will be kept\n')
 
1359
                raise ValueError('random exception raised')
 
1360
 
 
1361
        sio = cStringIO.StringIO()
 
1362
        runner = TextTestRunner(stream=sio)
 
1363
        test = LogTester('test_error')
 
1364
        result = self.run_test_runner(runner, test)
 
1365
 
 
1366
        text = sio.getvalue()
 
1367
        self.assertContainsRe(text, 'this will be kept')
 
1368
        self.assertContainsRe(text, 'random exception raised')
 
1369
 
 
1370
        log = test._get_log()
 
1371
        self.assertContainsRe(log, 'this will be kept')
 
1372
        self.assertEqual(log, test._log_contents)
 
1373
 
 
1374
 
 
1375
class SampleTestCase(TestCase):
1281
1376
 
1282
1377
    def _test_pass(self):
1283
1378
        pass
1285
1380
class _TestException(Exception):
1286
1381
    pass
1287
1382
 
1288
 
 
1289
 
class TestTestCase(tests.TestCase):
 
1383
class TestTestCase(TestCase):
1290
1384
    """Tests that test the core bzrlib TestCase."""
1291
1385
 
1292
1386
    def test_assertLength_matches_empty(self):
1309
1403
            exception.args[0])
1310
1404
 
1311
1405
    def test_base_setUp_not_called_causes_failure(self):
1312
 
        class TestCaseWithBrokenSetUp(tests.TestCase):
 
1406
        class TestCaseWithBrokenSetUp(TestCase):
1313
1407
            def setUp(self):
1314
1408
                pass # does not call TestCase.setUp
1315
1409
            def test_foo(self):
1321
1415
        self.assertEqual(1, result.testsRun)
1322
1416
 
1323
1417
    def test_base_tearDown_not_called_causes_failure(self):
1324
 
        class TestCaseWithBrokenTearDown(tests.TestCase):
 
1418
        class TestCaseWithBrokenTearDown(TestCase):
1325
1419
            def tearDown(self):
1326
1420
                pass # does not call TestCase.tearDown
1327
1421
            def test_foo(self):
1335
1429
    def test_debug_flags_sanitised(self):
1336
1430
        """The bzrlib debug flags should be sanitised by setUp."""
1337
1431
        if 'allow_debug' in tests.selftest_debug_flags:
1338
 
            raise tests.TestNotApplicable(
 
1432
            raise TestNotApplicable(
1339
1433
                '-Eallow_debug option prevents debug flag sanitisation')
1340
1434
        # we could set something and run a test that will check
1341
1435
        # it gets santised, but this is probably sufficient for now:
1342
1436
        # if someone runs the test with -Dsomething it will error.
1343
 
        flags = set()
1344
 
        if self._lock_check_thorough:
1345
 
            flags.add('strict_locks')
1346
 
        self.assertEqual(flags, bzrlib.debug.debug_flags)
 
1437
        self.assertEqual(set(), bzrlib.debug.debug_flags)
1347
1438
 
1348
1439
    def change_selftest_debug_flags(self, new_flags):
1349
 
        self.overrideAttr(tests, 'selftest_debug_flags', set(new_flags))
 
1440
        orig_selftest_flags = tests.selftest_debug_flags
 
1441
        self.addCleanup(self._restore_selftest_debug_flags, orig_selftest_flags)
 
1442
        tests.selftest_debug_flags = set(new_flags)
 
1443
 
 
1444
    def _restore_selftest_debug_flags(self, flags):
 
1445
        tests.selftest_debug_flags = flags
1350
1446
 
1351
1447
    def test_allow_debug_flag(self):
1352
1448
        """The -Eallow_debug flag prevents bzrlib.debug.debug_flags from being
1354
1450
        """
1355
1451
        self.change_selftest_debug_flags(set(['allow_debug']))
1356
1452
        bzrlib.debug.debug_flags = set(['a-flag'])
1357
 
        class TestThatRecordsFlags(tests.TestCase):
1358
 
            def test_foo(nested_self):
1359
 
                self.flags = set(bzrlib.debug.debug_flags)
1360
 
        test = TestThatRecordsFlags('test_foo')
1361
 
        test.run(self.make_test_result())
1362
 
        flags = set(['a-flag'])
1363
 
        if 'disable_lock_checks' not in tests.selftest_debug_flags:
1364
 
            flags.add('strict_locks')
1365
 
        self.assertEqual(flags, self.flags)
1366
 
 
1367
 
    def test_disable_lock_checks(self):
1368
 
        """The -Edisable_lock_checks flag disables thorough checks."""
1369
 
        class TestThatRecordsFlags(tests.TestCase):
1370
 
            def test_foo(nested_self):
1371
 
                self.flags = set(bzrlib.debug.debug_flags)
1372
 
                self.test_lock_check_thorough = nested_self._lock_check_thorough
1373
 
        self.change_selftest_debug_flags(set())
1374
 
        test = TestThatRecordsFlags('test_foo')
1375
 
        test.run(self.make_test_result())
1376
 
        # By default we do strict lock checking and thorough lock/unlock
1377
 
        # tracking.
1378
 
        self.assertTrue(self.test_lock_check_thorough)
1379
 
        self.assertEqual(set(['strict_locks']), self.flags)
1380
 
        # Now set the disable_lock_checks flag, and show that this changed.
1381
 
        self.change_selftest_debug_flags(set(['disable_lock_checks']))
1382
 
        test = TestThatRecordsFlags('test_foo')
1383
 
        test.run(self.make_test_result())
1384
 
        self.assertFalse(self.test_lock_check_thorough)
1385
 
        self.assertEqual(set(), self.flags)
1386
 
 
1387
 
    def test_this_fails_strict_lock_check(self):
1388
 
        class TestThatRecordsFlags(tests.TestCase):
1389
 
            def test_foo(nested_self):
1390
 
                self.flags1 = set(bzrlib.debug.debug_flags)
1391
 
                self.thisFailsStrictLockCheck()
1392
 
                self.flags2 = set(bzrlib.debug.debug_flags)
1393
 
        # Make sure lock checking is active
1394
 
        self.change_selftest_debug_flags(set())
1395
 
        test = TestThatRecordsFlags('test_foo')
1396
 
        test.run(self.make_test_result())
1397
 
        self.assertEqual(set(['strict_locks']), self.flags1)
1398
 
        self.assertEqual(set(), self.flags2)
 
1453
        class TestThatRecordsFlags(TestCase):
 
1454
            def test_foo(nested_self):
 
1455
                self.flags = set(bzrlib.debug.debug_flags)
 
1456
        test = TestThatRecordsFlags('test_foo')
 
1457
        test.run(self.make_test_result())
 
1458
        self.assertEqual(set(['a-flag']), self.flags)
1399
1459
 
1400
1460
    def test_debug_flags_restored(self):
1401
1461
        """The bzrlib debug flags should be restored to their original state
1404
1464
        self.change_selftest_debug_flags(set(['allow_debug']))
1405
1465
        # Now run a test that modifies debug.debug_flags.
1406
1466
        bzrlib.debug.debug_flags = set(['original-state'])
1407
 
        class TestThatModifiesFlags(tests.TestCase):
 
1467
        class TestThatModifiesFlags(TestCase):
1408
1468
            def test_foo(self):
1409
1469
                bzrlib.debug.debug_flags = set(['modified'])
1410
1470
        test = TestThatModifiesFlags('test_foo')
1412
1472
        self.assertEqual(set(['original-state']), bzrlib.debug.debug_flags)
1413
1473
 
1414
1474
    def make_test_result(self):
1415
 
        """Get a test result that writes to the test log file."""
1416
 
        return tests.TextTestResult(self._log_file, descriptions=0, verbosity=1)
 
1475
        return bzrlib.tests.TextTestResult(
 
1476
            self._log_file, descriptions=0, verbosity=1)
1417
1477
 
1418
1478
    def inner_test(self):
1419
1479
        # the inner child test
1426
1486
        result = self.make_test_result()
1427
1487
        self.inner_test.run(result)
1428
1488
        note("outer finish")
1429
 
        self.addCleanup(osutils.delete_any, self._log_file_name)
1430
1489
 
1431
1490
    def test_trace_nesting(self):
1432
1491
        # this tests that each test case nests its trace facility correctly.
1456
1515
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1457
1516
        output_stream = StringIO()
1458
1517
        result = bzrlib.tests.VerboseTestResult(
1459
 
            output_stream,
 
1518
            unittest._WritelnDecorator(output_stream),
1460
1519
            descriptions=0,
1461
 
            verbosity=2)
 
1520
            verbosity=2,
 
1521
            num_tests=sample_test.countTestCases())
1462
1522
        sample_test.run(result)
1463
1523
        self.assertContainsRe(
1464
1524
            output_stream.getvalue(),
1465
 
            r"\d+ms\*\n$")
 
1525
            r"\d+ms/ +\d+ms\n$")
1466
1526
 
1467
1527
    def test_hooks_sanitised(self):
1468
1528
        """The bzrlib hooks should be sanitised by setUp."""
1492
1552
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1493
1553
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1494
1554
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
1495
 
        del self._benchcalls[:]
1496
1555
 
1497
1556
    def test_knownFailure(self):
1498
1557
        """Self.knownFailure() should raise a KnownFailure exception."""
1499
 
        self.assertRaises(tests.KnownFailure, self.knownFailure, "A Failure")
1500
 
 
1501
 
    def test_open_bzrdir_safe_roots(self):
1502
 
        # even a memory transport should fail to open when its url isn't 
1503
 
        # permitted.
1504
 
        # Manually set one up (TestCase doesn't and shouldn't provide magic
1505
 
        # machinery)
1506
 
        transport_server = memory.MemoryServer()
1507
 
        transport_server.start_server()
1508
 
        self.addCleanup(transport_server.stop_server)
1509
 
        t = transport.get_transport(transport_server.get_url())
1510
 
        bzrdir.BzrDir.create(t.base)
1511
 
        self.assertRaises(errors.BzrError,
1512
 
            bzrdir.BzrDir.open_from_transport, t)
1513
 
        # But if we declare this as safe, we can open the bzrdir.
1514
 
        self.permit_url(t.base)
1515
 
        self._bzr_selftest_roots.append(t.base)
1516
 
        bzrdir.BzrDir.open_from_transport(t)
 
1558
        self.assertRaises(KnownFailure, self.knownFailure, "A Failure")
1517
1559
 
1518
1560
    def test_requireFeature_available(self):
1519
1561
        """self.requireFeature(available) is a no-op."""
1520
 
        class Available(tests.Feature):
 
1562
        class Available(Feature):
1521
1563
            def _probe(self):return True
1522
1564
        feature = Available()
1523
1565
        self.requireFeature(feature)
1524
1566
 
1525
1567
    def test_requireFeature_unavailable(self):
1526
1568
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1527
 
        class Unavailable(tests.Feature):
 
1569
        class Unavailable(Feature):
1528
1570
            def _probe(self):return False
1529
1571
        feature = Unavailable()
1530
 
        self.assertRaises(tests.UnavailableFeature,
1531
 
                          self.requireFeature, feature)
 
1572
        self.assertRaises(UnavailableFeature, self.requireFeature, feature)
1532
1573
 
1533
1574
    def test_run_no_parameters(self):
1534
1575
        test = SampleTestCase('_test_pass')
1564
1605
        """Test disabled tests behaviour with support aware results."""
1565
1606
        test = SampleTestCase('_test_pass')
1566
1607
        class DisabledFeature(object):
1567
 
            def __eq__(self, other):
1568
 
                return isinstance(other, DisabledFeature)
1569
1608
            def available(self):
1570
1609
                return False
1571
1610
        the_feature = DisabledFeature()
1582
1621
                self.calls.append(('addNotSupported', test, feature))
1583
1622
        result = InstrumentedTestResult()
1584
1623
        test.run(result)
1585
 
        case = result.calls[0][1]
1586
1624
        self.assertEqual([
1587
 
            ('startTest', case),
1588
 
            ('addNotSupported', case, the_feature),
1589
 
            ('stopTest', case),
 
1625
            ('startTest', test),
 
1626
            ('addNotSupported', test, the_feature),
 
1627
            ('stopTest', test),
1590
1628
            ],
1591
1629
            result.calls)
1592
1630
 
1593
 
    def test_start_server_registers_url(self):
1594
 
        transport_server = memory.MemoryServer()
1595
 
        # A little strict, but unlikely to be changed soon.
1596
 
        self.assertEqual([], self._bzr_selftest_roots)
1597
 
        self.start_server(transport_server)
1598
 
        self.assertSubset([transport_server.get_url()],
1599
 
            self._bzr_selftest_roots)
1600
 
 
1601
1631
    def test_assert_list_raises_on_generator(self):
1602
1632
        def generator_which_will_raise():
1603
1633
            # This will not raise until after the first yield
1653
1683
        self.assertRaises(AssertionError,
1654
1684
            self.assertListRaises, _TestException, success_generator)
1655
1685
 
1656
 
    def test_overrideAttr_without_value(self):
1657
 
        self.test_attr = 'original' # Define a test attribute
1658
 
        obj = self # Make 'obj' visible to the embedded test
1659
 
        class Test(tests.TestCase):
1660
 
 
1661
 
            def setUp(self):
1662
 
                tests.TestCase.setUp(self)
1663
 
                self.orig = self.overrideAttr(obj, 'test_attr')
1664
 
 
1665
 
            def test_value(self):
1666
 
                self.assertEqual('original', self.orig)
1667
 
                self.assertEqual('original', obj.test_attr)
1668
 
                obj.test_attr = 'modified'
1669
 
                self.assertEqual('modified', obj.test_attr)
1670
 
 
1671
 
        test = Test('test_value')
1672
 
        test.run(unittest.TestResult())
1673
 
        self.assertEqual('original', obj.test_attr)
1674
 
 
1675
 
    def test_overrideAttr_with_value(self):
1676
 
        self.test_attr = 'original' # Define a test attribute
1677
 
        obj = self # Make 'obj' visible to the embedded test
1678
 
        class Test(tests.TestCase):
1679
 
 
1680
 
            def setUp(self):
1681
 
                tests.TestCase.setUp(self)
1682
 
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1683
 
 
1684
 
            def test_value(self):
1685
 
                self.assertEqual('original', self.orig)
1686
 
                self.assertEqual('modified', obj.test_attr)
1687
 
 
1688
 
        test = Test('test_value')
1689
 
        test.run(unittest.TestResult())
1690
 
        self.assertEqual('original', obj.test_attr)
1691
 
 
1692
 
 
1693
 
class _MissingFeature(tests.Feature):
1694
 
    def _probe(self):
1695
 
        return False
1696
 
missing_feature = _MissingFeature()
1697
 
 
1698
 
 
1699
 
def _get_test(name):
1700
 
    """Get an instance of a specific example test.
1701
 
 
1702
 
    We protect this in a function so that they don't auto-run in the test
1703
 
    suite.
1704
 
    """
1705
 
 
1706
 
    class ExampleTests(tests.TestCase):
1707
 
 
1708
 
        def test_fail(self):
1709
 
            mutter('this was a failing test')
1710
 
            self.fail('this test will fail')
1711
 
 
1712
 
        def test_error(self):
1713
 
            mutter('this test errored')
1714
 
            raise RuntimeError('gotcha')
1715
 
 
1716
 
        def test_missing_feature(self):
1717
 
            mutter('missing the feature')
1718
 
            self.requireFeature(missing_feature)
1719
 
 
1720
 
        def test_skip(self):
1721
 
            mutter('this test will be skipped')
1722
 
            raise tests.TestSkipped('reason')
1723
 
 
1724
 
        def test_success(self):
1725
 
            mutter('this test succeeds')
1726
 
 
1727
 
        def test_xfail(self):
1728
 
            mutter('test with expected failure')
1729
 
            self.knownFailure('this_fails')
1730
 
 
1731
 
        def test_unexpected_success(self):
1732
 
            mutter('test with unexpected success')
1733
 
            self.expectFailure('should_fail', lambda: None)
1734
 
 
1735
 
    return ExampleTests(name)
1736
 
 
1737
 
 
1738
 
class TestTestCaseLogDetails(tests.TestCase):
1739
 
 
1740
 
    def _run_test(self, test_name):
1741
 
        test = _get_test(test_name)
1742
 
        result = testtools.TestResult()
1743
 
        test.run(result)
1744
 
        return result
1745
 
 
1746
 
    def test_fail_has_log(self):
1747
 
        result = self._run_test('test_fail')
1748
 
        self.assertEqual(1, len(result.failures))
1749
 
        result_content = result.failures[0][1]
1750
 
        self.assertContainsRe(result_content, 'Text attachment: log')
1751
 
        self.assertContainsRe(result_content, 'this was a failing test')
1752
 
 
1753
 
    def test_error_has_log(self):
1754
 
        result = self._run_test('test_error')
1755
 
        self.assertEqual(1, len(result.errors))
1756
 
        result_content = result.errors[0][1]
1757
 
        self.assertContainsRe(result_content, 'Text attachment: log')
1758
 
        self.assertContainsRe(result_content, 'this test errored')
1759
 
 
1760
 
    def test_skip_has_no_log(self):
1761
 
        result = self._run_test('test_skip')
1762
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
1763
 
        skips = result.skip_reasons['reason']
1764
 
        self.assertEqual(1, len(skips))
1765
 
        test = skips[0]
1766
 
        self.assertFalse('log' in test.getDetails())
1767
 
 
1768
 
    def test_missing_feature_has_no_log(self):
1769
 
        # testtools doesn't know about addNotSupported, so it just gets
1770
 
        # considered as a skip
1771
 
        result = self._run_test('test_missing_feature')
1772
 
        self.assertEqual([missing_feature], result.skip_reasons.keys())
1773
 
        skips = result.skip_reasons[missing_feature]
1774
 
        self.assertEqual(1, len(skips))
1775
 
        test = skips[0]
1776
 
        self.assertFalse('log' in test.getDetails())
1777
 
 
1778
 
    def test_xfail_has_no_log(self):
1779
 
        result = self._run_test('test_xfail')
1780
 
        self.assertEqual(1, len(result.expectedFailures))
1781
 
        result_content = result.expectedFailures[0][1]
1782
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
1783
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
1784
 
 
1785
 
    def test_unexpected_success_has_log(self):
1786
 
        result = self._run_test('test_unexpected_success')
1787
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
1788
 
        # Inconsistency, unexpectedSuccesses is a list of tests,
1789
 
        # expectedFailures is a list of reasons?
1790
 
        test = result.unexpectedSuccesses[0]
1791
 
        details = test.getDetails()
1792
 
        self.assertTrue('log' in details)
1793
 
 
1794
 
 
1795
 
class TestTestCloning(tests.TestCase):
1796
 
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
1797
 
 
1798
 
    def test_cloned_testcase_does_not_share_details(self):
1799
 
        """A TestCase cloned with clone_test does not share mutable attributes
1800
 
        such as details or cleanups.
1801
 
        """
1802
 
        class Test(tests.TestCase):
1803
 
            def test_foo(self):
1804
 
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1805
 
        orig_test = Test('test_foo')
1806
 
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1807
 
        orig_test.run(unittest.TestResult())
1808
 
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1809
 
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
1810
 
 
1811
 
    def test_double_apply_scenario_preserves_first_scenario(self):
1812
 
        """Applying two levels of scenarios to a test preserves the attributes
1813
 
        added by both scenarios.
1814
 
        """
1815
 
        class Test(tests.TestCase):
1816
 
            def test_foo(self):
1817
 
                pass
1818
 
        test = Test('test_foo')
1819
 
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1820
 
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1821
 
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1822
 
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1823
 
        all_tests = list(tests.iter_suite_tests(suite))
1824
 
        self.assertLength(4, all_tests)
1825
 
        all_xys = sorted((t.x, t.y) for t in all_tests)
1826
 
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1827
 
 
1828
1686
 
1829
1687
# NB: Don't delete this; it's not actually from 0.11!
1830
1688
@deprecated_function(deprecated_in((0, 11, 0)))
1853
1711
        return sample_deprecated_function()
1854
1712
 
1855
1713
 
1856
 
class TestExtraAssertions(tests.TestCase):
 
1714
class TestExtraAssertions(TestCase):
1857
1715
    """Tests for new test assertions in bzrlib test suite"""
1858
1716
 
1859
1717
    def test_assert_isinstance(self):
1860
1718
        self.assertIsInstance(2, int)
1861
1719
        self.assertIsInstance(u'', basestring)
1862
 
        e = self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1863
 
        self.assertEquals(str(e),
1864
 
            "None is an instance of <type 'NoneType'> rather than <type 'int'>")
 
1720
        self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1865
1721
        self.assertRaises(AssertionError, self.assertIsInstance, 23.3, int)
1866
 
        e = self.assertRaises(AssertionError,
1867
 
            self.assertIsInstance, None, int, "it's just not")
1868
 
        self.assertEquals(str(e),
1869
 
            "None is an instance of <type 'NoneType'> rather than <type 'int'>"
1870
 
            ": it's just not")
1871
1722
 
1872
1723
    def test_assertEndsWith(self):
1873
1724
        self.assertEndsWith('foo', 'oo')
1874
1725
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1875
1726
 
1876
 
    def test_assertEqualDiff(self):
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
 
                          'first string is missing a final newline.\n+ \n')
1882
 
        e = self.assertRaises(AssertionError,
1883
 
                              self.assertEqualDiff, '\n', '')
1884
 
        self.assertEquals(str(e),
1885
 
                          # Don't blink ! The '-' applies to the second string
1886
 
                          'second string is missing a final newline.\n- \n')
1887
 
 
1888
 
 
1889
 
class TestDeprecations(tests.TestCase):
1890
 
 
1891
1727
    def test_applyDeprecated_not_deprecated(self):
1892
1728
        sample_object = ApplyDeprecatedHelper()
1893
1729
        # calling an undeprecated callable raises an assertion
1936
1772
        self.callDeprecated([], testfunc, be_deprecated=False)
1937
1773
 
1938
1774
 
1939
 
class TestWarningTests(tests.TestCase):
 
1775
class TestWarningTests(TestCase):
1940
1776
    """Tests for calling methods that raise warnings."""
1941
1777
 
1942
1778
    def test_callCatchWarnings(self):
1952
1788
        self.assertEquals("this is your last warning", str(w0))
1953
1789
 
1954
1790
 
1955
 
class TestConvenienceMakers(tests.TestCaseWithTransport):
 
1791
class TestConvenienceMakers(TestCaseWithTransport):
1956
1792
    """Test for the make_* convenience functions."""
1957
1793
 
1958
1794
    def test_make_branch_and_tree_with_format(self):
1970
1806
        tree = self.make_branch_and_memory_tree('a')
1971
1807
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1972
1808
 
1973
 
    def test_make_tree_for_local_vfs_backed_transport(self):
1974
 
        # make_branch_and_tree has to use local branch and repositories
1975
 
        # when the vfs transport and local disk are colocated, even if
1976
 
        # a different transport is in use for url generation.
1977
 
        self.transport_server = test_server.FakeVFATServer
1978
 
        self.assertFalse(self.get_url('t1').startswith('file://'))
 
1809
 
 
1810
class TestSFTPMakeBranchAndTree(TestCaseWithSFTPServer):
 
1811
 
 
1812
    def test_make_tree_for_sftp_branch(self):
 
1813
        """Transports backed by local directories create local trees."""
 
1814
 
1979
1815
        tree = self.make_branch_and_tree('t1')
1980
1816
        base = tree.bzrdir.root_transport.base
1981
 
        self.assertStartsWith(base, 'file://')
 
1817
        self.failIf(base.startswith('sftp'),
 
1818
                'base %r is on sftp but should be local' % base)
1982
1819
        self.assertEquals(tree.bzrdir.root_transport,
1983
1820
                tree.branch.bzrdir.root_transport)
1984
1821
        self.assertEquals(tree.bzrdir.root_transport,
1985
1822
                tree.branch.repository.bzrdir.root_transport)
1986
1823
 
1987
1824
 
1988
 
class SelfTestHelper(object):
1989
 
 
1990
 
    def run_selftest(self, **kwargs):
1991
 
        """Run selftest returning its output."""
1992
 
        output = StringIO()
1993
 
        old_transport = bzrlib.tests.default_transport
1994
 
        old_root = tests.TestCaseWithMemoryTransport.TEST_ROOT
1995
 
        tests.TestCaseWithMemoryTransport.TEST_ROOT = None
1996
 
        try:
1997
 
            self.assertEqual(True, tests.selftest(stream=output, **kwargs))
1998
 
        finally:
1999
 
            bzrlib.tests.default_transport = old_transport
2000
 
            tests.TestCaseWithMemoryTransport.TEST_ROOT = old_root
2001
 
        output.seek(0)
2002
 
        return output
2003
 
 
2004
 
 
2005
 
class TestSelftest(tests.TestCase, SelfTestHelper):
 
1825
class TestSelftest(TestCase):
2006
1826
    """Tests of bzrlib.tests.selftest."""
2007
1827
 
2008
1828
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
2009
1829
        factory_called = []
2010
1830
        def factory():
2011
1831
            factory_called.append(True)
2012
 
            return TestUtil.TestSuite()
 
1832
            return TestSuite()
2013
1833
        out = StringIO()
2014
1834
        err = StringIO()
2015
1835
        self.apply_redirected(out, err, None, bzrlib.tests.selftest,
2016
1836
            test_suite_factory=factory)
2017
1837
        self.assertEqual([True], factory_called)
2018
1838
 
2019
 
    def factory(self):
2020
 
        """A test suite factory."""
2021
 
        class Test(tests.TestCase):
2022
 
            def a(self):
2023
 
                pass
2024
 
            def b(self):
2025
 
                pass
2026
 
            def c(self):
2027
 
                pass
2028
 
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
2029
 
 
2030
 
    def test_list_only(self):
2031
 
        output = self.run_selftest(test_suite_factory=self.factory,
2032
 
            list_only=True)
2033
 
        self.assertEqual(3, len(output.readlines()))
2034
 
 
2035
 
    def test_list_only_filtered(self):
2036
 
        output = self.run_selftest(test_suite_factory=self.factory,
2037
 
            list_only=True, pattern="Test.b")
2038
 
        self.assertEndsWith(output.getvalue(), "Test.b\n")
2039
 
        self.assertLength(1, output.readlines())
2040
 
 
2041
 
    def test_list_only_excludes(self):
2042
 
        output = self.run_selftest(test_suite_factory=self.factory,
2043
 
            list_only=True, exclude_pattern="Test.b")
2044
 
        self.assertNotContainsRe("Test.b", output.getvalue())
2045
 
        self.assertLength(2, output.readlines())
2046
 
 
2047
 
    def test_lsprof_tests(self):
2048
 
        self.requireFeature(test_lsprof.LSProfFeature)
2049
 
        calls = []
2050
 
        class Test(object):
2051
 
            def __call__(test, result):
2052
 
                test.run(result)
2053
 
            def run(test, result):
2054
 
                self.assertIsInstance(result, tests.ForwardingResult)
2055
 
                calls.append("called")
2056
 
            def countTestCases(self):
2057
 
                return 1
2058
 
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
2059
 
        self.assertLength(1, calls)
2060
 
 
2061
 
    def test_random(self):
2062
 
        # test randomising by listing a number of tests.
2063
 
        output_123 = self.run_selftest(test_suite_factory=self.factory,
2064
 
            list_only=True, random_seed="123")
2065
 
        output_234 = self.run_selftest(test_suite_factory=self.factory,
2066
 
            list_only=True, random_seed="234")
2067
 
        self.assertNotEqual(output_123, output_234)
2068
 
        # "Randominzing test order..\n\n
2069
 
        self.assertLength(5, output_123.readlines())
2070
 
        self.assertLength(5, output_234.readlines())
2071
 
 
2072
 
    def test_random_reuse_is_same_order(self):
2073
 
        # test randomising by listing a number of tests.
2074
 
        expected = self.run_selftest(test_suite_factory=self.factory,
2075
 
            list_only=True, random_seed="123")
2076
 
        repeated = self.run_selftest(test_suite_factory=self.factory,
2077
 
            list_only=True, random_seed="123")
2078
 
        self.assertEqual(expected.getvalue(), repeated.getvalue())
2079
 
 
2080
 
    def test_runner_class(self):
2081
 
        self.requireFeature(features.subunit)
2082
 
        from subunit import ProtocolTestCase
2083
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2084
 
            test_suite_factory=self.factory)
2085
 
        test = ProtocolTestCase(stream)
2086
 
        result = unittest.TestResult()
2087
 
        test.run(result)
2088
 
        self.assertEqual(3, result.testsRun)
2089
 
 
2090
 
    def test_starting_with_single_argument(self):
2091
 
        output = self.run_selftest(test_suite_factory=self.factory,
2092
 
            starting_with=['bzrlib.tests.test_selftest.Test.a'],
2093
 
            list_only=True)
2094
 
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n',
2095
 
            output.getvalue())
2096
 
 
2097
 
    def test_starting_with_multiple_argument(self):
2098
 
        output = self.run_selftest(test_suite_factory=self.factory,
2099
 
            starting_with=['bzrlib.tests.test_selftest.Test.a',
2100
 
                'bzrlib.tests.test_selftest.Test.b'],
2101
 
            list_only=True)
2102
 
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n'
2103
 
            'bzrlib.tests.test_selftest.Test.b\n',
2104
 
            output.getvalue())
2105
 
 
2106
 
    def check_transport_set(self, transport_server):
2107
 
        captured_transport = []
2108
 
        def seen_transport(a_transport):
2109
 
            captured_transport.append(a_transport)
2110
 
        class Capture(tests.TestCase):
2111
 
            def a(self):
2112
 
                seen_transport(bzrlib.tests.default_transport)
2113
 
        def factory():
2114
 
            return TestUtil.TestSuite([Capture("a")])
2115
 
        self.run_selftest(transport=transport_server, test_suite_factory=factory)
2116
 
        self.assertEqual(transport_server, captured_transport[0])
2117
 
 
2118
 
    def test_transport_sftp(self):
2119
 
        self.requireFeature(features.paramiko)
2120
 
        from bzrlib.tests import stub_sftp
2121
 
        self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
2122
 
 
2123
 
    def test_transport_memory(self):
2124
 
        self.check_transport_set(memory.MemoryServer)
2125
 
 
2126
 
 
2127
 
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
2128
 
    # Does IO: reads test.list
2129
 
 
2130
 
    def test_load_list(self):
2131
 
        # Provide a list with one test - this test.
2132
 
        test_id_line = '%s\n' % self.id()
2133
 
        self.build_tree_contents([('test.list', test_id_line)])
2134
 
        # And generate a list of the tests in  the suite.
2135
 
        stream = self.run_selftest(load_list='test.list', list_only=True)
2136
 
        self.assertEqual(test_id_line, stream.getvalue())
2137
 
 
2138
 
    def test_load_unknown(self):
2139
 
        # Provide a list with one test - this test.
2140
 
        # And generate a list of the tests in  the suite.
2141
 
        err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
2142
 
            load_list='missing file name', list_only=True)
2143
 
 
2144
 
 
2145
 
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2146
 
 
2147
 
    _test_needs_features = [features.subunit]
2148
 
 
2149
 
    def run_subunit_stream(self, test_name):
2150
 
        from subunit import ProtocolTestCase
2151
 
        def factory():
2152
 
            return TestUtil.TestSuite([_get_test(test_name)])
2153
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2154
 
            test_suite_factory=factory)
2155
 
        test = ProtocolTestCase(stream)
2156
 
        result = testtools.TestResult()
2157
 
        test.run(result)
2158
 
        content = stream.getvalue()
2159
 
        return content, result
2160
 
 
2161
 
    def test_fail_has_log(self):
2162
 
        content, result = self.run_subunit_stream('test_fail')
2163
 
        self.assertEqual(1, len(result.failures))
2164
 
        self.assertContainsRe(content, '(?m)^log$')
2165
 
        self.assertContainsRe(content, 'this test will fail')
2166
 
 
2167
 
    def test_error_has_log(self):
2168
 
        content, result = self.run_subunit_stream('test_error')
2169
 
        self.assertContainsRe(content, '(?m)^log$')
2170
 
        self.assertContainsRe(content, 'this test errored')
2171
 
 
2172
 
    def test_skip_has_no_log(self):
2173
 
        content, result = self.run_subunit_stream('test_skip')
2174
 
        self.assertNotContainsRe(content, '(?m)^log$')
2175
 
        self.assertNotContainsRe(content, 'this test will be skipped')
2176
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
2177
 
        skips = result.skip_reasons['reason']
2178
 
        self.assertEqual(1, len(skips))
2179
 
        test = skips[0]
2180
 
        # RemotedTestCase doesn't preserve the "details"
2181
 
        ## self.assertFalse('log' in test.getDetails())
2182
 
 
2183
 
    def test_missing_feature_has_no_log(self):
2184
 
        content, result = self.run_subunit_stream('test_missing_feature')
2185
 
        self.assertNotContainsRe(content, '(?m)^log$')
2186
 
        self.assertNotContainsRe(content, 'missing the feature')
2187
 
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
2188
 
        skips = result.skip_reasons['_MissingFeature\n']
2189
 
        self.assertEqual(1, len(skips))
2190
 
        test = skips[0]
2191
 
        # RemotedTestCase doesn't preserve the "details"
2192
 
        ## self.assertFalse('log' in test.getDetails())
2193
 
 
2194
 
    def test_xfail_has_no_log(self):
2195
 
        content, result = self.run_subunit_stream('test_xfail')
2196
 
        self.assertNotContainsRe(content, '(?m)^log$')
2197
 
        self.assertNotContainsRe(content, 'test with expected failure')
2198
 
        self.assertEqual(1, len(result.expectedFailures))
2199
 
        result_content = result.expectedFailures[0][1]
2200
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
2201
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
2202
 
 
2203
 
    def test_unexpected_success_has_log(self):
2204
 
        content, result = self.run_subunit_stream('test_unexpected_success')
2205
 
        self.assertContainsRe(content, '(?m)^log$')
2206
 
        self.assertContainsRe(content, 'test with unexpected success')
2207
 
        self.expectFailure('subunit treats "unexpectedSuccess"'
2208
 
                           ' as a plain success',
2209
 
            self.assertEqual, 1, len(result.unexpectedSuccesses))
2210
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
2211
 
        test = result.unexpectedSuccesses[0]
2212
 
        # RemotedTestCase doesn't preserve the "details"
2213
 
        ## self.assertTrue('log' in test.getDetails())
2214
 
 
2215
 
    def test_success_has_no_log(self):
2216
 
        content, result = self.run_subunit_stream('test_success')
2217
 
        self.assertEqual(1, result.testsRun)
2218
 
        self.assertNotContainsRe(content, '(?m)^log$')
2219
 
        self.assertNotContainsRe(content, 'this test succeeds')
2220
 
 
2221
 
 
2222
 
class TestRunBzr(tests.TestCase):
2223
 
 
2224
 
    out = ''
2225
 
    err = ''
2226
 
 
2227
 
    def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
2228
 
                         working_dir=None):
2229
 
        """Override _run_bzr_core to test how it is invoked by run_bzr.
2230
 
 
2231
 
        Attempts to run bzr from inside this class don't actually run it.
2232
 
 
2233
 
        We test how run_bzr actually invokes bzr in another location.  Here we
2234
 
        only need to test that it passes the right parameters to run_bzr.
2235
 
        """
2236
 
        self.argv = list(argv)
2237
 
        self.retcode = retcode
2238
 
        self.encoding = encoding
2239
 
        self.stdin = stdin
2240
 
        self.working_dir = working_dir
2241
 
        return self.retcode, self.out, self.err
2242
 
 
2243
 
    def test_run_bzr_error(self):
2244
 
        self.out = "It sure does!\n"
2245
 
        out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
2246
 
        self.assertEqual(['rocks'], self.argv)
2247
 
        self.assertEqual(34, self.retcode)
2248
 
        self.assertEqual('It sure does!\n', out)
2249
 
        self.assertEquals(out, self.out)
2250
 
        self.assertEqual('', err)
2251
 
        self.assertEquals(err, self.err)
2252
 
 
2253
 
    def test_run_bzr_error_regexes(self):
2254
 
        self.out = ''
2255
 
        self.err = "bzr: ERROR: foobarbaz is not versioned"
2256
 
        out, err = self.run_bzr_error(
2257
 
            ["bzr: ERROR: foobarbaz is not versioned"],
2258
 
            ['file-id', 'foobarbaz'])
2259
 
 
2260
 
    def test_encoding(self):
2261
 
        """Test that run_bzr passes encoding to _run_bzr_core"""
2262
 
        self.run_bzr('foo bar')
2263
 
        self.assertEqual(None, self.encoding)
2264
 
        self.assertEqual(['foo', 'bar'], self.argv)
2265
 
 
2266
 
        self.run_bzr('foo bar', encoding='baz')
2267
 
        self.assertEqual('baz', self.encoding)
2268
 
        self.assertEqual(['foo', 'bar'], self.argv)
2269
 
 
2270
 
    def test_retcode(self):
2271
 
        """Test that run_bzr passes retcode to _run_bzr_core"""
2272
 
        # Default is retcode == 0
2273
 
        self.run_bzr('foo bar')
2274
 
        self.assertEqual(0, self.retcode)
2275
 
        self.assertEqual(['foo', 'bar'], self.argv)
2276
 
 
2277
 
        self.run_bzr('foo bar', retcode=1)
2278
 
        self.assertEqual(1, self.retcode)
2279
 
        self.assertEqual(['foo', 'bar'], self.argv)
2280
 
 
2281
 
        self.run_bzr('foo bar', retcode=None)
2282
 
        self.assertEqual(None, self.retcode)
2283
 
        self.assertEqual(['foo', 'bar'], self.argv)
2284
 
 
2285
 
        self.run_bzr(['foo', 'bar'], retcode=3)
2286
 
        self.assertEqual(3, self.retcode)
2287
 
        self.assertEqual(['foo', 'bar'], self.argv)
2288
 
 
2289
 
    def test_stdin(self):
2290
 
        # test that the stdin keyword to run_bzr is passed through to
2291
 
        # _run_bzr_core as-is. We do this by overriding
2292
 
        # _run_bzr_core in this class, and then calling run_bzr,
2293
 
        # which is a convenience function for _run_bzr_core, so
2294
 
        # should invoke it.
2295
 
        self.run_bzr('foo bar', stdin='gam')
2296
 
        self.assertEqual('gam', self.stdin)
2297
 
        self.assertEqual(['foo', 'bar'], self.argv)
2298
 
 
2299
 
        self.run_bzr('foo bar', stdin='zippy')
2300
 
        self.assertEqual('zippy', self.stdin)
2301
 
        self.assertEqual(['foo', 'bar'], self.argv)
2302
 
 
2303
 
    def test_working_dir(self):
2304
 
        """Test that run_bzr passes working_dir to _run_bzr_core"""
2305
 
        self.run_bzr('foo bar')
2306
 
        self.assertEqual(None, self.working_dir)
2307
 
        self.assertEqual(['foo', 'bar'], self.argv)
2308
 
 
2309
 
        self.run_bzr('foo bar', working_dir='baz')
2310
 
        self.assertEqual('baz', self.working_dir)
2311
 
        self.assertEqual(['foo', 'bar'], self.argv)
2312
 
 
2313
 
    def test_reject_extra_keyword_arguments(self):
2314
 
        self.assertRaises(TypeError, self.run_bzr, "foo bar",
2315
 
                          error_regex=['error message'])
2316
 
 
2317
 
 
2318
 
class TestRunBzrCaptured(tests.TestCaseWithTransport):
2319
 
    # Does IO when testing the working_dir parameter.
2320
 
 
2321
 
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
2322
 
                         a_callable=None, *args, **kwargs):
2323
 
        self.stdin = stdin
2324
 
        self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
2325
 
        self.factory = bzrlib.ui.ui_factory
2326
 
        self.working_dir = osutils.getcwd()
2327
 
        stdout.write('foo\n')
2328
 
        stderr.write('bar\n')
2329
 
        return 0
2330
 
 
2331
 
    def test_stdin(self):
2332
 
        # test that the stdin keyword to _run_bzr_core is passed through to
2333
 
        # apply_redirected as a StringIO. We do this by overriding
2334
 
        # apply_redirected in this class, and then calling _run_bzr_core,
2335
 
        # which calls apply_redirected.
2336
 
        self.run_bzr(['foo', 'bar'], stdin='gam')
2337
 
        self.assertEqual('gam', self.stdin.read())
2338
 
        self.assertTrue(self.stdin is self.factory_stdin)
2339
 
        self.run_bzr(['foo', 'bar'], stdin='zippy')
2340
 
        self.assertEqual('zippy', self.stdin.read())
2341
 
        self.assertTrue(self.stdin is self.factory_stdin)
2342
 
 
2343
 
    def test_ui_factory(self):
2344
 
        # each invocation of self.run_bzr should get its
2345
 
        # own UI factory, which is an instance of TestUIFactory,
2346
 
        # with stdin, stdout and stderr attached to the stdin,
2347
 
        # stdout and stderr of the invoked run_bzr
2348
 
        current_factory = bzrlib.ui.ui_factory
2349
 
        self.run_bzr(['foo'])
2350
 
        self.failIf(current_factory is self.factory)
2351
 
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2352
 
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2353
 
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2354
 
        self.assertEqual('bar\n', self.factory.stderr.getvalue())
2355
 
        self.assertIsInstance(self.factory, tests.TestUIFactory)
2356
 
 
2357
 
    def test_working_dir(self):
2358
 
        self.build_tree(['one/', 'two/'])
2359
 
        cwd = osutils.getcwd()
2360
 
 
2361
 
        # Default is to work in the current directory
2362
 
        self.run_bzr(['foo', 'bar'])
2363
 
        self.assertEqual(cwd, self.working_dir)
2364
 
 
2365
 
        self.run_bzr(['foo', 'bar'], working_dir=None)
2366
 
        self.assertEqual(cwd, self.working_dir)
2367
 
 
2368
 
        # The function should be run in the alternative directory
2369
 
        # but afterwards the current working dir shouldn't be changed
2370
 
        self.run_bzr(['foo', 'bar'], working_dir='one')
2371
 
        self.assertNotEqual(cwd, self.working_dir)
2372
 
        self.assertEndsWith(self.working_dir, 'one')
2373
 
        self.assertEqual(cwd, osutils.getcwd())
2374
 
 
2375
 
        self.run_bzr(['foo', 'bar'], working_dir='two')
2376
 
        self.assertNotEqual(cwd, self.working_dir)
2377
 
        self.assertEndsWith(self.working_dir, 'two')
2378
 
        self.assertEqual(cwd, osutils.getcwd())
2379
 
 
2380
 
 
2381
 
class StubProcess(object):
2382
 
    """A stub process for testing run_bzr_subprocess."""
2383
 
    
2384
 
    def __init__(self, out="", err="", retcode=0):
2385
 
        self.out = out
2386
 
        self.err = err
2387
 
        self.returncode = retcode
2388
 
 
2389
 
    def communicate(self):
2390
 
        return self.out, self.err
2391
 
 
2392
 
 
2393
 
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
2394
 
    """Base class for tests testing how we might run bzr."""
2395
 
 
2396
 
    def setUp(self):
2397
 
        tests.TestCaseWithTransport.setUp(self)
2398
 
        self.subprocess_calls = []
2399
 
 
2400
 
    def start_bzr_subprocess(self, process_args, env_changes=None,
2401
 
                             skip_if_plan_to_signal=False,
2402
 
                             working_dir=None,
2403
 
                             allow_plugins=False):
2404
 
        """capture what run_bzr_subprocess tries to do."""
2405
 
        self.subprocess_calls.append({'process_args':process_args,
2406
 
            'env_changes':env_changes,
2407
 
            'skip_if_plan_to_signal':skip_if_plan_to_signal,
2408
 
            'working_dir':working_dir, 'allow_plugins':allow_plugins})
2409
 
        return self.next_subprocess
2410
 
 
2411
 
 
2412
 
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
2413
 
 
2414
 
    def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
2415
 
        """Run run_bzr_subprocess with args and kwargs using a stubbed process.
2416
 
 
2417
 
        Inside TestRunBzrSubprocessCommands we use a stub start_bzr_subprocess
2418
 
        that will return static results. This assertion method populates those
2419
 
        results and also checks the arguments run_bzr_subprocess generates.
2420
 
        """
2421
 
        self.next_subprocess = process
2422
 
        try:
2423
 
            result = self.run_bzr_subprocess(*args, **kwargs)
2424
 
        except:
2425
 
            self.next_subprocess = None
2426
 
            for key, expected in expected_args.iteritems():
2427
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2428
 
            raise
 
1839
 
 
1840
class TestKnownFailure(TestCase):
 
1841
 
 
1842
    def test_known_failure(self):
 
1843
        """Check that KnownFailure is defined appropriately."""
 
1844
        # a KnownFailure is an assertion error for compatability with unaware
 
1845
        # runners.
 
1846
        self.assertIsInstance(KnownFailure(""), AssertionError)
 
1847
 
 
1848
    def test_expect_failure(self):
 
1849
        try:
 
1850
            self.expectFailure("Doomed to failure", self.assertTrue, False)
 
1851
        except KnownFailure, e:
 
1852
            self.assertEqual('Doomed to failure', e.args[0])
 
1853
        try:
 
1854
            self.expectFailure("Doomed to failure", self.assertTrue, True)
 
1855
        except AssertionError, e:
 
1856
            self.assertEqual('Unexpected success.  Should have failed:'
 
1857
                             ' Doomed to failure', e.args[0])
2429
1858
        else:
2430
 
            self.next_subprocess = None
2431
 
            for key, expected in expected_args.iteritems():
2432
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2433
 
            return result
2434
 
 
2435
 
    def test_run_bzr_subprocess(self):
2436
 
        """The run_bzr_helper_external command behaves nicely."""
2437
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2438
 
            StubProcess(), '--version')
2439
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2440
 
            StubProcess(), ['--version'])
2441
 
        # retcode=None disables retcode checking
2442
 
        result = self.assertRunBzrSubprocess({},
2443
 
            StubProcess(retcode=3), '--version', retcode=None)
2444
 
        result = self.assertRunBzrSubprocess({},
2445
 
            StubProcess(out="is free software"), '--version')
2446
 
        self.assertContainsRe(result[0], 'is free software')
2447
 
        # Running a subcommand that is missing errors
2448
 
        self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
2449
 
            {'process_args':['--versionn']}, StubProcess(retcode=3),
2450
 
            '--versionn')
2451
 
        # Unless it is told to expect the error from the subprocess
2452
 
        result = self.assertRunBzrSubprocess({},
2453
 
            StubProcess(retcode=3), '--versionn', retcode=3)
2454
 
        # Or to ignore retcode checking
2455
 
        result = self.assertRunBzrSubprocess({},
2456
 
            StubProcess(err="unknown command", retcode=3), '--versionn',
2457
 
            retcode=None)
2458
 
        self.assertContainsRe(result[1], 'unknown command')
2459
 
 
2460
 
    def test_env_change_passes_through(self):
2461
 
        self.assertRunBzrSubprocess(
2462
 
            {'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
2463
 
            StubProcess(), '',
2464
 
            env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
2465
 
 
2466
 
    def test_no_working_dir_passed_as_None(self):
2467
 
        self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
2468
 
 
2469
 
    def test_no_working_dir_passed_through(self):
2470
 
        self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
2471
 
            working_dir='dir')
2472
 
 
2473
 
    def test_run_bzr_subprocess_no_plugins(self):
2474
 
        self.assertRunBzrSubprocess({'allow_plugins': False},
2475
 
            StubProcess(), '')
2476
 
 
2477
 
    def test_allow_plugins(self):
2478
 
        self.assertRunBzrSubprocess({'allow_plugins': True},
2479
 
            StubProcess(), '', allow_plugins=True)
2480
 
 
2481
 
 
2482
 
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2483
 
 
2484
 
    def test_finish_bzr_subprocess_with_error(self):
2485
 
        """finish_bzr_subprocess allows specification of the desired exit code.
2486
 
        """
2487
 
        process = StubProcess(err="unknown command", retcode=3)
2488
 
        result = self.finish_bzr_subprocess(process, retcode=3)
2489
 
        self.assertEqual('', result[0])
2490
 
        self.assertContainsRe(result[1], 'unknown command')
2491
 
 
2492
 
    def test_finish_bzr_subprocess_ignoring_retcode(self):
2493
 
        """finish_bzr_subprocess allows the exit code to be ignored."""
2494
 
        process = StubProcess(err="unknown command", retcode=3)
2495
 
        result = self.finish_bzr_subprocess(process, retcode=None)
2496
 
        self.assertEqual('', result[0])
2497
 
        self.assertContainsRe(result[1], 'unknown command')
2498
 
 
2499
 
    def test_finish_subprocess_with_unexpected_retcode(self):
2500
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2501
 
        not the expected one.
2502
 
        """
2503
 
        process = StubProcess(err="unknown command", retcode=3)
2504
 
        self.assertRaises(self.failureException, self.finish_bzr_subprocess,
2505
 
                          process)
2506
 
 
2507
 
 
2508
 
class _DontSpawnProcess(Exception):
2509
 
    """A simple exception which just allows us to skip unnecessary steps"""
2510
 
 
2511
 
 
2512
 
class TestStartBzrSubProcess(tests.TestCase):
2513
 
 
2514
 
    def check_popen_state(self):
2515
 
        """Replace to make assertions when popen is called."""
2516
 
 
2517
 
    def _popen(self, *args, **kwargs):
2518
 
        """Record the command that is run, so that we can ensure it is correct"""
2519
 
        self.check_popen_state()
2520
 
        self._popen_args = args
2521
 
        self._popen_kwargs = kwargs
2522
 
        raise _DontSpawnProcess()
2523
 
 
2524
 
    def test_run_bzr_subprocess_no_plugins(self):
2525
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2526
 
        command = self._popen_args[0]
2527
 
        self.assertEqual(sys.executable, command[0])
2528
 
        self.assertEqual(self.get_bzr_path(), command[1])
2529
 
        self.assertEqual(['--no-plugins'], command[2:])
2530
 
 
2531
 
    def test_allow_plugins(self):
2532
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2533
 
            allow_plugins=True)
2534
 
        command = self._popen_args[0]
2535
 
        self.assertEqual([], command[2:])
2536
 
 
2537
 
    def test_set_env(self):
2538
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2539
 
        # set in the child
2540
 
        def check_environment():
2541
 
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2542
 
        self.check_popen_state = check_environment
2543
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2544
 
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
2545
 
        # not set in theparent
2546
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2547
 
 
2548
 
    def test_run_bzr_subprocess_env_del(self):
2549
 
        """run_bzr_subprocess can remove environment variables too."""
2550
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2551
 
        def check_environment():
2552
 
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2553
 
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2554
 
        self.check_popen_state = check_environment
2555
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2556
 
            env_changes={'EXISTANT_ENV_VAR':None})
2557
 
        # Still set in parent
2558
 
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2559
 
        del os.environ['EXISTANT_ENV_VAR']
2560
 
 
2561
 
    def test_env_del_missing(self):
2562
 
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2563
 
        def check_environment():
2564
 
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2565
 
        self.check_popen_state = check_environment
2566
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2567
 
            env_changes={'NON_EXISTANT_ENV_VAR':None})
2568
 
 
2569
 
    def test_working_dir(self):
2570
 
        """Test that we can specify the working dir for the child"""
2571
 
        orig_getcwd = osutils.getcwd
2572
 
        orig_chdir = os.chdir
2573
 
        chdirs = []
2574
 
        def chdir(path):
2575
 
            chdirs.append(path)
2576
 
        os.chdir = chdir
2577
 
        try:
2578
 
            def getcwd():
2579
 
                return 'current'
2580
 
            osutils.getcwd = getcwd
2581
 
            try:
2582
 
                self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2583
 
                    working_dir='foo')
2584
 
            finally:
2585
 
                osutils.getcwd = orig_getcwd
2586
 
        finally:
2587
 
            os.chdir = orig_chdir
2588
 
        self.assertEqual(['foo', 'current'], chdirs)
2589
 
 
2590
 
    def test_get_bzr_path_with_cwd_bzrlib(self):
2591
 
        self.get_source_path = lambda: ""
2592
 
        self.overrideAttr(os.path, "isfile", lambda path: True)
2593
 
        self.assertEqual(self.get_bzr_path(), "bzr")
2594
 
 
2595
 
 
2596
 
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2597
 
    """Tests that really need to do things with an external bzr."""
2598
 
 
2599
 
    def test_start_and_stop_bzr_subprocess_send_signal(self):
2600
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2601
 
        not the expected one.
2602
 
        """
2603
 
        self.disable_missing_extensions_warning()
2604
 
        process = self.start_bzr_subprocess(['wait-until-signalled'],
2605
 
                                            skip_if_plan_to_signal=True)
2606
 
        self.assertEqual('running\n', process.stdout.readline())
2607
 
        result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
2608
 
                                            retcode=3)
2609
 
        self.assertEqual('', result[0])
2610
 
        self.assertEqual('bzr: interrupted\n', result[1])
2611
 
 
2612
 
 
2613
 
class TestFeature(tests.TestCase):
 
1859
            self.fail('Assertion not raised')
 
1860
 
 
1861
 
 
1862
class TestFeature(TestCase):
2614
1863
 
2615
1864
    def test_caching(self):
2616
1865
        """Feature._probe is called by the feature at most once."""
2617
 
        class InstrumentedFeature(tests.Feature):
 
1866
        class InstrumentedFeature(Feature):
2618
1867
            def __init__(self):
2619
 
                super(InstrumentedFeature, self).__init__()
 
1868
                Feature.__init__(self)
2620
1869
                self.calls = []
2621
1870
            def _probe(self):
2622
1871
                self.calls.append('_probe')
2629
1878
 
2630
1879
    def test_named_str(self):
2631
1880
        """Feature.__str__ should thunk to feature_name()."""
2632
 
        class NamedFeature(tests.Feature):
 
1881
        class NamedFeature(Feature):
2633
1882
            def feature_name(self):
2634
1883
                return 'symlinks'
2635
1884
        feature = NamedFeature()
2637
1886
 
2638
1887
    def test_default_str(self):
2639
1888
        """Feature.__str__ should default to __class__.__name__."""
2640
 
        class NamedFeature(tests.Feature):
 
1889
        class NamedFeature(Feature):
2641
1890
            pass
2642
1891
        feature = NamedFeature()
2643
1892
        self.assertEqual('NamedFeature', str(feature))
2644
1893
 
2645
1894
 
2646
 
class TestUnavailableFeature(tests.TestCase):
 
1895
class TestUnavailableFeature(TestCase):
2647
1896
 
2648
1897
    def test_access_feature(self):
2649
 
        feature = tests.Feature()
2650
 
        exception = tests.UnavailableFeature(feature)
 
1898
        feature = Feature()
 
1899
        exception = UnavailableFeature(feature)
2651
1900
        self.assertIs(feature, exception.args[0])
2652
1901
 
2653
1902
 
2654
 
simple_thunk_feature = tests._CompatabilityThunkFeature(
2655
 
    deprecated_in((2, 1, 0)),
2656
 
    'bzrlib.tests.test_selftest',
2657
 
    'simple_thunk_feature','UnicodeFilename',
2658
 
    replacement_module='bzrlib.tests'
2659
 
    )
2660
 
 
2661
 
class Test_CompatibilityFeature(tests.TestCase):
2662
 
 
2663
 
    def test_does_thunk(self):
2664
 
        res = self.callDeprecated(
2665
 
            ['bzrlib.tests.test_selftest.simple_thunk_feature was deprecated'
2666
 
             ' in version 2.1.0. Use bzrlib.tests.UnicodeFilename instead.'],
2667
 
            simple_thunk_feature.available)
2668
 
        self.assertEqual(tests.UnicodeFilename.available(), res)
2669
 
 
2670
 
 
2671
 
class TestModuleAvailableFeature(tests.TestCase):
2672
 
 
2673
 
    def test_available_module(self):
2674
 
        feature = tests.ModuleAvailableFeature('bzrlib.tests')
2675
 
        self.assertEqual('bzrlib.tests', feature.module_name)
2676
 
        self.assertEqual('bzrlib.tests', str(feature))
2677
 
        self.assertTrue(feature.available())
2678
 
        self.assertIs(tests, feature.module)
2679
 
 
2680
 
    def test_unavailable_module(self):
2681
 
        feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
2682
 
        self.assertEqual('bzrlib.no_such_module_exists', str(feature))
2683
 
        self.assertFalse(feature.available())
2684
 
        self.assertIs(None, feature.module)
2685
 
 
2686
 
 
2687
 
class TestSelftestFiltering(tests.TestCase):
 
1903
class TestSelftestFiltering(TestCase):
2688
1904
 
2689
1905
    def setUp(self):
2690
 
        tests.TestCase.setUp(self)
 
1906
        TestCase.setUp(self)
2691
1907
        self.suite = TestUtil.TestSuite()
2692
1908
        self.loader = TestUtil.TestLoader()
2693
 
        self.suite.addTest(self.loader.loadTestsFromModule(
2694
 
            sys.modules['bzrlib.tests.test_selftest']))
 
1909
        self.suite.addTest(self.loader.loadTestsFromModuleNames([
 
1910
            'bzrlib.tests.test_selftest']))
2695
1911
        self.all_names = _test_ids(self.suite)
2696
1912
 
2697
1913
    def test_condition_id_re(self):
2698
1914
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2699
1915
            'test_condition_id_re')
2700
 
        filtered_suite = tests.filter_suite_by_condition(
2701
 
            self.suite, tests.condition_id_re('test_condition_id_re'))
 
1916
        filtered_suite = filter_suite_by_condition(self.suite,
 
1917
            condition_id_re('test_condition_id_re'))
2702
1918
        self.assertEqual([test_name], _test_ids(filtered_suite))
2703
1919
 
2704
1920
    def test_condition_id_in_list(self):
2705
1921
        test_names = ['bzrlib.tests.test_selftest.TestSelftestFiltering.'
2706
1922
                      'test_condition_id_in_list']
2707
1923
        id_list = tests.TestIdList(test_names)
2708
 
        filtered_suite = tests.filter_suite_by_condition(
 
1924
        filtered_suite = filter_suite_by_condition(
2709
1925
            self.suite, tests.condition_id_in_list(id_list))
2710
1926
        my_pattern = 'TestSelftestFiltering.*test_condition_id_in_list'
2711
 
        re_filtered = tests.filter_suite_by_re(self.suite, my_pattern)
 
1927
        re_filtered = filter_suite_by_re(self.suite, my_pattern)
2712
1928
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
2713
1929
 
2714
1930
    def test_condition_id_startswith(self):
2718
1934
        test_names = [ klass + 'test_condition_id_in_list',
2719
1935
                      klass + 'test_condition_id_startswith',
2720
1936
                     ]
2721
 
        filtered_suite = tests.filter_suite_by_condition(
 
1937
        filtered_suite = filter_suite_by_condition(
2722
1938
            self.suite, tests.condition_id_startswith([start1, start2]))
2723
1939
        self.assertEqual(test_names, _test_ids(filtered_suite))
2724
1940
 
2725
1941
    def test_condition_isinstance(self):
2726
 
        filtered_suite = tests.filter_suite_by_condition(
2727
 
            self.suite, tests.condition_isinstance(self.__class__))
 
1942
        filtered_suite = filter_suite_by_condition(self.suite,
 
1943
            condition_isinstance(self.__class__))
2728
1944
        class_pattern = 'bzrlib.tests.test_selftest.TestSelftestFiltering.'
2729
 
        re_filtered = tests.filter_suite_by_re(self.suite, class_pattern)
 
1945
        re_filtered = filter_suite_by_re(self.suite, class_pattern)
2730
1946
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
2731
1947
 
2732
1948
    def test_exclude_tests_by_condition(self):
2733
1949
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2734
1950
            'test_exclude_tests_by_condition')
2735
 
        filtered_suite = tests.exclude_tests_by_condition(self.suite,
 
1951
        filtered_suite = exclude_tests_by_condition(self.suite,
2736
1952
            lambda x:x.id() == excluded_name)
2737
1953
        self.assertEqual(len(self.all_names) - 1,
2738
1954
            filtered_suite.countTestCases())
2743
1959
 
2744
1960
    def test_exclude_tests_by_re(self):
2745
1961
        self.all_names = _test_ids(self.suite)
2746
 
        filtered_suite = tests.exclude_tests_by_re(self.suite,
2747
 
                                                   'exclude_tests_by_re')
 
1962
        filtered_suite = exclude_tests_by_re(self.suite, 'exclude_tests_by_re')
2748
1963
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2749
1964
            'test_exclude_tests_by_re')
2750
1965
        self.assertEqual(len(self.all_names) - 1,
2757
1972
    def test_filter_suite_by_condition(self):
2758
1973
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2759
1974
            'test_filter_suite_by_condition')
2760
 
        filtered_suite = tests.filter_suite_by_condition(self.suite,
 
1975
        filtered_suite = filter_suite_by_condition(self.suite,
2761
1976
            lambda x:x.id() == test_name)
2762
1977
        self.assertEqual([test_name], _test_ids(filtered_suite))
2763
1978
 
2764
1979
    def test_filter_suite_by_re(self):
2765
 
        filtered_suite = tests.filter_suite_by_re(self.suite,
2766
 
                                                  'test_filter_suite_by_r')
 
1980
        filtered_suite = filter_suite_by_re(self.suite, 'test_filter_suite_by_r')
2767
1981
        filtered_names = _test_ids(filtered_suite)
2768
1982
        self.assertEqual(filtered_names, ['bzrlib.tests.test_selftest.'
2769
1983
            'TestSelftestFiltering.test_filter_suite_by_re'])
2797
2011
 
2798
2012
    def test_preserve_input(self):
2799
2013
        # NB: Surely this is something in the stdlib to do this?
2800
 
        self.assertTrue(self.suite is tests.preserve_input(self.suite))
2801
 
        self.assertTrue("@#$" is tests.preserve_input("@#$"))
 
2014
        self.assertTrue(self.suite is preserve_input(self.suite))
 
2015
        self.assertTrue("@#$" is preserve_input("@#$"))
2802
2016
 
2803
2017
    def test_randomize_suite(self):
2804
 
        randomized_suite = tests.randomize_suite(self.suite)
 
2018
        randomized_suite = randomize_suite(self.suite)
2805
2019
        # randomizing should not add or remove test names.
2806
2020
        self.assertEqual(set(_test_ids(self.suite)),
2807
2021
                         set(_test_ids(randomized_suite)))
2817
2031
 
2818
2032
    def test_split_suit_by_condition(self):
2819
2033
        self.all_names = _test_ids(self.suite)
2820
 
        condition = tests.condition_id_re('test_filter_suite_by_r')
2821
 
        split_suite = tests.split_suite_by_condition(self.suite, condition)
 
2034
        condition = condition_id_re('test_filter_suite_by_r')
 
2035
        split_suite = split_suite_by_condition(self.suite, condition)
2822
2036
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2823
2037
            'test_filter_suite_by_re')
2824
2038
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
2829
2043
 
2830
2044
    def test_split_suit_by_re(self):
2831
2045
        self.all_names = _test_ids(self.suite)
2832
 
        split_suite = tests.split_suite_by_re(self.suite,
2833
 
                                              'test_filter_suite_by_r')
 
2046
        split_suite = split_suite_by_re(self.suite, 'test_filter_suite_by_r')
2834
2047
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2835
2048
            'test_filter_suite_by_re')
2836
2049
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
2840
2053
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2841
2054
 
2842
2055
 
2843
 
class TestCheckInventoryShape(tests.TestCaseWithTransport):
 
2056
class TestCheckInventoryShape(TestCaseWithTransport):
2844
2057
 
2845
2058
    def test_check_inventory_shape(self):
2846
2059
        files = ['a', 'b/', 'b/c']
2854
2067
            tree.unlock()
2855
2068
 
2856
2069
 
2857
 
class TestBlackboxSupport(tests.TestCase):
 
2070
class TestBlackboxSupport(TestCase):
2858
2071
    """Tests for testsuite blackbox features."""
2859
2072
 
2860
2073
    def test_run_bzr_failure_not_caught(self):
2868
2081
        # the test framework
2869
2082
        self.assertEquals('always fails', str(e))
2870
2083
        # check that there's no traceback in the test log
2871
 
        self.assertNotContainsRe(self.get_log(), r'Traceback')
 
2084
        self.assertNotContainsRe(self._get_log(keep_log_file=True),
 
2085
            r'Traceback')
2872
2086
 
2873
2087
    def test_run_bzr_user_error_caught(self):
2874
2088
        # Running bzr in blackbox mode, normal/expected/user errors should be
2875
2089
        # caught in the regular way and turned into an error message plus exit
2876
2090
        # code.
2877
 
        transport_server = memory.MemoryServer()
2878
 
        transport_server.start_server()
2879
 
        self.addCleanup(transport_server.stop_server)
2880
 
        url = transport_server.get_url()
2881
 
        self.permit_url(url)
2882
 
        out, err = self.run_bzr(["log", "%s/nonexistantpath" % url], retcode=3)
 
2091
        out, err = self.run_bzr(["log", "/nonexistantpath"], retcode=3)
2883
2092
        self.assertEqual(out, '')
2884
2093
        self.assertContainsRe(err,
2885
2094
            'bzr: ERROR: Not a branch: ".*nonexistantpath/".\n')
2886
2095
 
2887
2096
 
2888
 
class TestTestLoader(tests.TestCase):
 
2097
class TestTestLoader(TestCase):
2889
2098
    """Tests for the test loader."""
2890
2099
 
2891
2100
    def _get_loader_and_module(self):
2892
2101
        """Gets a TestLoader and a module with one test in it."""
2893
2102
        loader = TestUtil.TestLoader()
2894
2103
        module = {}
2895
 
        class Stub(tests.TestCase):
 
2104
        class Stub(TestCase):
2896
2105
            def test_foo(self):
2897
2106
                pass
2898
2107
        class MyModule(object):
2911
2120
        # load_tests do not need that :)
2912
2121
        def load_tests(self, standard_tests, module, loader):
2913
2122
            result = loader.suiteClass()
2914
 
            for test in tests.iter_suite_tests(standard_tests):
 
2123
            for test in iter_suite_tests(standard_tests):
2915
2124
                result.addTests([test, test])
2916
2125
            return result
2917
2126
        # add a load_tests() method which multiplies the tests from the module.
2936
2145
 
2937
2146
    def _create_suite(self, test_id_list):
2938
2147
 
2939
 
        class Stub(tests.TestCase):
 
2148
        class Stub(TestCase):
2940
2149
            def test_foo(self):
2941
2150
                pass
2942
2151
 
2952
2161
 
2953
2162
    def _test_ids(self, test_suite):
2954
2163
        """Get the ids for the tests in a test suite."""
2955
 
        return [t.id() for t in tests.iter_suite_tests(test_suite)]
 
2164
        return [t.id() for t in iter_suite_tests(test_suite)]
2956
2165
 
2957
2166
    def test_empty_list(self):
2958
2167
        id_list = self._create_id_list([])
2984
2193
        self.assertTrue(id_list.refers_to('mod.class'))
2985
2194
        self.assertTrue(id_list.refers_to('mod.class.meth'))
2986
2195
 
 
2196
    def test_test_suite(self):
 
2197
        # This test is slow, so we do a single test with one test in each
 
2198
        # category
 
2199
        test_list = [
 
2200
            # testmod_names
 
2201
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
 
2202
            'bzrlib.tests.test_selftest.TestTestIdList.test_test_suite',
 
2203
            # transport implementations
 
2204
            'bzrlib.tests.test_transport_implementations.TransportTests'
 
2205
            '.test_abspath(LocalURLServer)',
 
2206
            # modules_to_doctest
 
2207
            'bzrlib.timestamp.format_highres_date',
 
2208
            # plugins can't be tested that way since selftest may be run with
 
2209
            # --no-plugins
 
2210
            ]
 
2211
        suite = tests.test_suite(test_list)
 
2212
        self.assertEquals(test_list, _test_ids(suite))
 
2213
 
2987
2214
    def test_test_suite_matches_id_list_with_unknown(self):
2988
2215
        loader = TestUtil.TestLoader()
2989
2216
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2997
2224
        loader = TestUtil.TestLoader()
2998
2225
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2999
2226
        dupes = loader.suiteClass()
3000
 
        for test in tests.iter_suite_tests(suite):
 
2227
        for test in iter_suite_tests(suite):
3001
2228
            dupes.addTest(test)
3002
2229
            dupes.addTest(test) # Add it again
3003
2230
 
3009
2236
                          duplicates)
3010
2237
 
3011
2238
 
3012
 
class TestTestSuite(tests.TestCase):
3013
 
 
3014
 
    def test__test_suite_testmod_names(self):
3015
 
        # Test that a plausible list of test module names are returned
3016
 
        # by _test_suite_testmod_names.
3017
 
        test_list = tests._test_suite_testmod_names()
3018
 
        self.assertSubset([
3019
 
            'bzrlib.tests.blackbox',
3020
 
            'bzrlib.tests.per_transport',
3021
 
            'bzrlib.tests.test_selftest',
3022
 
            ],
3023
 
            test_list)
3024
 
 
3025
 
    def test__test_suite_modules_to_doctest(self):
3026
 
        # Test that a plausible list of modules to doctest is returned
3027
 
        # by _test_suite_modules_to_doctest.
3028
 
        test_list = tests._test_suite_modules_to_doctest()
3029
 
        if __doc__ is None:
3030
 
            # When docstrings are stripped, there are no modules to doctest
3031
 
            self.assertEqual([], test_list)
3032
 
            return
3033
 
        self.assertSubset([
3034
 
            'bzrlib.timestamp',
3035
 
            ],
3036
 
            test_list)
3037
 
 
3038
 
    def test_test_suite(self):
3039
 
        # test_suite() loads the entire test suite to operate. To avoid this
3040
 
        # overhead, and yet still be confident that things are happening,
3041
 
        # we temporarily replace two functions used by test_suite with 
3042
 
        # test doubles that supply a few sample tests to load, and check they
3043
 
        # are loaded.
3044
 
        calls = []
3045
 
        def testmod_names():
3046
 
            calls.append("testmod_names")
3047
 
            return [
3048
 
                'bzrlib.tests.blackbox.test_branch',
3049
 
                'bzrlib.tests.per_transport',
3050
 
                'bzrlib.tests.test_selftest',
3051
 
                ]
3052
 
        self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
3053
 
        def doctests():
3054
 
            calls.append("modules_to_doctest")
3055
 
            if __doc__ is None:
3056
 
                return []
3057
 
            return ['bzrlib.timestamp']
3058
 
        self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
3059
 
        expected_test_list = [
3060
 
            # testmod_names
3061
 
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
3062
 
            ('bzrlib.tests.per_transport.TransportTests'
3063
 
             '.test_abspath(LocalTransport,LocalURLServer)'),
3064
 
            'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
3065
 
            # plugins can't be tested that way since selftest may be run with
3066
 
            # --no-plugins
3067
 
            ]
3068
 
        if __doc__ is not None:
3069
 
            expected_test_list.extend([
3070
 
                # modules_to_doctest
3071
 
                'bzrlib.timestamp.format_highres_date',
3072
 
                ])
3073
 
        suite = tests.test_suite()
3074
 
        self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
3075
 
            set(calls))
3076
 
        self.assertSubset(expected_test_list, _test_ids(suite))
3077
 
 
3078
 
    def test_test_suite_list_and_start(self):
3079
 
        # We cannot test this at the same time as the main load, because we want
3080
 
        # to know that starting_with == None works. So a second load is
3081
 
        # incurred - note that the starting_with parameter causes a partial load
3082
 
        # rather than a full load so this test should be pretty quick.
3083
 
        test_list = ['bzrlib.tests.test_selftest.TestTestSuite.test_test_suite']
3084
 
        suite = tests.test_suite(test_list,
3085
 
                                 ['bzrlib.tests.test_selftest.TestTestSuite'])
3086
 
        # test_test_suite_list_and_start is not included 
3087
 
        self.assertEquals(test_list, _test_ids(suite))
3088
 
 
3089
 
 
3090
2239
class TestLoadTestIdList(tests.TestCaseInTempDir):
3091
2240
 
3092
2241
    def _create_test_list_file(self, file_name, content):
3130
2279
    def test_load_tests(self):
3131
2280
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3132
2281
        loader = self._create_loader(test_list)
 
2282
 
3133
2283
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3134
2284
        self.assertEquals(test_list, _test_ids(suite))
3135
2285
 
3136
2286
    def test_exclude_tests(self):
3137
2287
        test_list = ['bogus']
3138
2288
        loader = self._create_loader(test_list)
 
2289
 
3139
2290
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3140
2291
        self.assertEquals([], _test_ids(suite))
3141
2292
 
3186
2337
        tpr.register('bar', 'bbb.aaa.rrr')
3187
2338
        tpr.register('bar', 'bBB.aAA.rRR')
3188
2339
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
3189
 
        self.assertThat(self.get_log(),
3190
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
 
2340
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
2341
                              r'.*bar.*bbb.aaa.rrr.*bBB.aAA.rRR')
3191
2342
 
3192
2343
    def test_get_unknown_prefix(self):
3193
2344
        tpr = self._get_registry()
3213
2364
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3214
2365
 
3215
2366
 
3216
 
class TestThreadLeakDetection(tests.TestCase):
3217
 
    """Ensure when tests leak threads we detect and report it"""
3218
 
 
3219
 
    class LeakRecordingResult(tests.ExtendedTestResult):
3220
 
        def __init__(self):
3221
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3222
 
            self.leaks = []
3223
 
        def _report_thread_leak(self, test, leaks, alive):
3224
 
            self.leaks.append((test, leaks))
3225
 
 
3226
 
    def test_testcase_without_addCleanups(self):
3227
 
        """Check old TestCase instances don't break with leak detection"""
3228
 
        class Test(unittest.TestCase):
3229
 
            def runTest(self):
3230
 
                pass
3231
 
            addCleanup = None # for when on Python 2.7 with native addCleanup
3232
 
        result = self.LeakRecordingResult()
3233
 
        test = Test()
3234
 
        self.assertIs(getattr(test, "addCleanup", None), None)
3235
 
        result.startTestRun()
3236
 
        test.run(result)
3237
 
        result.stopTestRun()
3238
 
        self.assertEqual(result._tests_leaking_threads_count, 0)
3239
 
        self.assertEqual(result.leaks, [])
3240
 
        
3241
 
    def test_thread_leak(self):
3242
 
        """Ensure a thread that outlives the running of a test is reported
3243
 
 
3244
 
        Uses a thread that blocks on an event, and is started by the inner
3245
 
        test case. As the thread outlives the inner case's run, it should be
3246
 
        detected as a leak, but the event is then set so that the thread can
3247
 
        be safely joined in cleanup so it's not leaked for real.
3248
 
        """
3249
 
        event = threading.Event()
3250
 
        thread = threading.Thread(name="Leaker", target=event.wait)
3251
 
        class Test(tests.TestCase):
3252
 
            def test_leak(self):
3253
 
                thread.start()
3254
 
        result = self.LeakRecordingResult()
3255
 
        test = Test("test_leak")
3256
 
        self.addCleanup(thread.join)
3257
 
        self.addCleanup(event.set)
3258
 
        result.startTestRun()
3259
 
        test.run(result)
3260
 
        result.stopTestRun()
3261
 
        self.assertEqual(result._tests_leaking_threads_count, 1)
3262
 
        self.assertEqual(result._first_thread_leaker_id, test.id())
3263
 
        self.assertEqual(result.leaks, [(test, set([thread]))])
3264
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3265
 
 
3266
 
    def test_multiple_leaks(self):
3267
 
        """Check multiple leaks are blamed on the test cases at fault
3268
 
 
3269
 
        Same concept as the previous test, but has one inner test method that
3270
 
        leaks two threads, and one that doesn't leak at all.
3271
 
        """
3272
 
        event = threading.Event()
3273
 
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
3274
 
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
3275
 
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
3276
 
        class Test(tests.TestCase):
3277
 
            def test_first_leak(self):
3278
 
                thread_b.start()
3279
 
            def test_second_no_leak(self):
3280
 
                pass
3281
 
            def test_third_leak(self):
3282
 
                thread_c.start()
3283
 
                thread_a.start()
3284
 
        result = self.LeakRecordingResult()
3285
 
        first_test = Test("test_first_leak")
3286
 
        third_test = Test("test_third_leak")
3287
 
        self.addCleanup(thread_a.join)
3288
 
        self.addCleanup(thread_b.join)
3289
 
        self.addCleanup(thread_c.join)
3290
 
        self.addCleanup(event.set)
3291
 
        result.startTestRun()
3292
 
        unittest.TestSuite(
3293
 
            [first_test, Test("test_second_no_leak"), third_test]
3294
 
            ).run(result)
3295
 
        result.stopTestRun()
3296
 
        self.assertEqual(result._tests_leaking_threads_count, 2)
3297
 
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
3298
 
        self.assertEqual(result.leaks, [
3299
 
            (first_test, set([thread_b])),
3300
 
            (third_test, set([thread_a, thread_c]))])
3301
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3302
 
 
3303
 
 
3304
 
class TestRunSuite(tests.TestCase):
 
2367
class TestRunSuite(TestCase):
3305
2368
 
3306
2369
    def test_runner_class(self):
3307
2370
        """run_suite accepts and uses a runner_class keyword argument."""
3308
 
        class Stub(tests.TestCase):
 
2371
        class Stub(TestCase):
3309
2372
            def test_foo(self):
3310
2373
                pass
3311
2374
        suite = Stub("test_foo")
3312
2375
        calls = []
3313
 
        class MyRunner(tests.TextTestRunner):
 
2376
        class MyRunner(TextTestRunner):
3314
2377
            def run(self, test):
3315
2378
                calls.append(test)
3316
 
                return tests.ExtendedTestResult(self.stream, self.descriptions,
3317
 
                                                self.verbosity)
3318
 
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
3319
 
        self.assertLength(1, calls)
 
2379
                return ExtendedTestResult(self.stream, self.descriptions,
 
2380
                    self.verbosity)
 
2381
        run_suite(suite, runner_class=MyRunner, stream=StringIO())
 
2382
        self.assertEqual(calls, [suite])
 
2383
 
 
2384
    def test_done(self):
 
2385
        """run_suite should call result.done()"""
 
2386
        self.calls = 0
 
2387
        def one_more_call(): self.calls += 1
 
2388
        def test_function():
 
2389
            pass
 
2390
        test = unittest.FunctionTestCase(test_function)
 
2391
        class InstrumentedTestResult(ExtendedTestResult):
 
2392
            def done(self): one_more_call()
 
2393
        class MyRunner(TextTestRunner):
 
2394
            def run(self, test):
 
2395
                return InstrumentedTestResult(self.stream, self.descriptions,
 
2396
                                              self.verbosity)
 
2397
        run_suite(test, runner_class=MyRunner, stream=StringIO())
 
2398
        self.assertEquals(1, self.calls)