~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Jonathan Lange
  • Date: 2009-05-01 06:42:30 UTC
  • mto: This revision was merged to the branch mainline in revision 4320.
  • Revision ID: jml@canonical.com-20090501064230-kyk7v49xt8cevd25
Remove InstallFailed, it's not needed anymore.

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,
41
30
    bzrdir,
42
 
    debug,
43
31
    errors,
44
 
    lockdir,
45
32
    memorytree,
46
33
    osutils,
47
34
    remote,
48
35
    repository,
49
36
    symbol_versioning,
50
37
    tests,
51
 
    transport,
52
38
    workingtree,
53
39
    )
 
40
from bzrlib.progress import _BaseProgressBar
54
41
from bzrlib.repofmt import (
55
 
    groupcompress_repo,
56
42
    pack_repo,
57
43
    weaverepo,
58
44
    )
62
48
    deprecated_method,
63
49
    )
64
50
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
 
51
                          ChrootedTestCase,
 
52
                          ExtendedTestResult,
 
53
                          Feature,
 
54
                          KnownFailure,
 
55
                          TestCase,
 
56
                          TestCaseInTempDir,
 
57
                          TestCaseWithMemoryTransport,
 
58
                          TestCaseWithTransport,
 
59
                          TestNotApplicable,
 
60
                          TestSkipped,
 
61
                          TestSuite,
 
62
                          TestUtil,
 
63
                          TextTestRunner,
 
64
                          UnavailableFeature,
 
65
                          condition_id_re,
 
66
                          condition_isinstance,
 
67
                          exclude_tests_by_condition,
 
68
                          exclude_tests_by_re,
 
69
                          filter_suite_by_condition,
 
70
                          filter_suite_by_re,
 
71
                          iter_suite_tests,
 
72
                          preserve_input,
 
73
                          randomize_suite,
 
74
                          run_suite,
 
75
                          split_suite_by_condition,
 
76
                          split_suite_by_re,
 
77
                          test_lsprof,
 
78
                          test_suite,
 
79
                          )
 
80
from bzrlib.tests.test_sftp_transport import TestCaseWithSFTPServer
 
81
from bzrlib.tests.TestUtil import _load_module_by_name
 
82
from bzrlib.trace import note
 
83
from bzrlib.transport.memory import MemoryServer, MemoryTransport
73
84
from bzrlib.version import _get_bzr_source_tree
74
85
 
75
86
 
76
87
def _test_ids(test_suite):
77
88
    """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):
 
89
    return [t.id() for t in iter_suite_tests(test_suite)]
 
90
 
 
91
 
 
92
class SelftestTests(TestCase):
82
93
 
83
94
    def test_import_tests(self):
84
 
        mod = TestUtil._load_module_by_name('bzrlib.tests.test_selftest')
 
95
        mod = _load_module_by_name('bzrlib.tests.test_selftest')
85
96
        self.assertEqual(mod.SelftestTests, SelftestTests)
86
97
 
87
98
    def test_import_test_failure(self):
88
99
        self.assertRaises(ImportError,
89
 
                          TestUtil._load_module_by_name,
 
100
                          _load_module_by_name,
90
101
                          'bzrlib.no-name-yet')
91
102
 
92
 
 
93
 
class MetaTestLog(tests.TestCase):
 
103
class MetaTestLog(TestCase):
94
104
 
95
105
    def test_logging(self):
96
106
        """Test logs are captured when a test fails."""
97
107
        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):
 
108
        self._log_file.flush()
 
109
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
110
                              'a test message\n')
 
111
 
 
112
 
 
113
class TestUnicodeFilename(TestCase):
108
114
 
109
115
    def test_probe_passes(self):
110
116
        """UnicodeFilename._probe passes."""
113
119
        tests.UnicodeFilename._probe()
114
120
 
115
121
 
116
 
class TestTreeShape(tests.TestCaseInTempDir):
 
122
class TestTreeShape(TestCaseInTempDir):
117
123
 
118
124
    def test_unicode_paths(self):
119
125
        self.requireFeature(tests.UnicodeFilename)
123
129
        self.failUnlessExists(filename)
124
130
 
125
131
 
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):
 
132
class TestTransportScenarios(TestCase):
140
133
    """A group of tests that test the transport implementation adaption core.
141
134
 
142
135
    This is a meta test that the tests are applied to all available
153
146
            def get_test_permutations(self):
154
147
                return sample_permutation
155
148
        sample_permutation = [(1,2), (3,4)]
156
 
        from bzrlib.tests.per_transport import get_transport_test_permutations
 
149
        from bzrlib.tests.test_transport_implementations \
 
150
            import get_transport_test_permutations
157
151
        self.assertEqual(sample_permutation,
158
152
                         get_transport_test_permutations(MockModule()))
159
153
 
160
 
    def test_scenarios_include_all_modules(self):
 
154
    def test_scenarios_invlude_all_modules(self):
161
155
        # this checks that the scenario generator returns as many permutations
162
156
        # as there are in all the registered transport modules - we assume if
163
157
        # this matches its probably doing the right thing especially in
164
158
        # combination with the tests for setting the right classes below.
165
 
        from bzrlib.tests.per_transport import transport_test_permutations
 
159
        from bzrlib.tests.test_transport_implementations \
 
160
            import transport_test_permutations
166
161
        from bzrlib.transport import _get_transport_modules
167
162
        modules = _get_transport_modules()
168
163
        permutation_count = 0
180
175
        # This test used to know about all the possible transports and the
181
176
        # order they were returned but that seems overly brittle (mbp
182
177
        # 20060307)
183
 
        from bzrlib.tests.per_transport import transport_test_permutations
 
178
        from bzrlib.tests.test_transport_implementations \
 
179
            import transport_test_permutations
184
180
        scenarios = transport_test_permutations()
185
181
        # there are at least that many builtin transports
186
182
        self.assertTrue(len(scenarios) > 6)
192
188
                                   bzrlib.transport.Server))
193
189
 
194
190
 
195
 
class TestBranchScenarios(tests.TestCase):
 
191
class TestBranchScenarios(TestCase):
196
192
 
197
193
    def test_scenarios(self):
198
194
        # check that constructor parameters are passed through to the adapted
199
195
        # test.
200
 
        from bzrlib.tests.per_branch import make_scenarios
 
196
        from bzrlib.tests.branch_implementations import make_scenarios
201
197
        server1 = "a"
202
198
        server2 = "b"
203
199
        formats = [("c", "C"), ("d", "D")]
217
213
            scenarios)
218
214
 
219
215
 
220
 
class TestBzrDirScenarios(tests.TestCase):
 
216
class TestBzrDirScenarios(TestCase):
221
217
 
222
218
    def test_scenarios(self):
223
219
        # check that constructor parameters are passed through to the adapted
224
220
        # test.
225
 
        from bzrlib.tests.per_controldir import make_scenarios
 
221
        from bzrlib.tests.bzrdir_implementations import make_scenarios
226
222
        vfs_factory = "v"
227
223
        server1 = "a"
228
224
        server2 = "b"
242
238
            scenarios)
243
239
 
244
240
 
245
 
class TestRepositoryScenarios(tests.TestCase):
 
241
class TestRepositoryScenarios(TestCase):
246
242
 
247
243
    def test_formats_to_scenarios(self):
248
244
        from bzrlib.tests.per_repository import formats_to_scenarios
249
245
        formats = [("(c)", remote.RemoteRepositoryFormat()),
250
246
                   ("(d)", repository.format_registry.get(
251
 
                    'Bazaar repository format 2a (needs bzr 1.16 or later)\n'))]
 
247
                        'Bazaar pack repository format 1 (needs bzr 0.92)\n'))]
252
248
        no_vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
253
249
            None)
254
250
        vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
255
251
            vfs_transport_factory="vfs")
256
252
        # no_vfs generate scenarios without vfs_transport_factory
257
 
        expected = [
 
253
        self.assertEqual([
258
254
            ('RemoteRepositoryFormat(c)',
259
255
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
260
256
              'repository_format': remote.RemoteRepositoryFormat(),
261
257
              'transport_readonly_server': 'readonly',
262
258
              'transport_server': 'server'}),
263
 
            ('RepositoryFormat2a(d)',
 
259
            ('RepositoryFormatKnitPack1(d)',
264
260
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
265
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
261
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
266
262
              'transport_readonly_server': 'readonly',
267
 
              'transport_server': 'server'})]
268
 
        self.assertEqual(expected, no_vfs_scenarios)
 
263
              'transport_server': 'server'})],
 
264
            no_vfs_scenarios)
269
265
        self.assertEqual([
270
266
            ('RemoteRepositoryFormat(c)',
271
267
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
273
269
              'transport_readonly_server': 'readonly',
274
270
              'transport_server': 'server',
275
271
              'vfs_transport_factory': 'vfs'}),
276
 
            ('RepositoryFormat2a(d)',
 
272
            ('RepositoryFormatKnitPack1(d)',
277
273
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
278
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
274
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
279
275
              'transport_readonly_server': 'readonly',
280
276
              'transport_server': 'server',
281
277
              'vfs_transport_factory': 'vfs'})],
282
278
            vfs_scenarios)
283
279
 
284
280
 
285
 
class TestTestScenarioApplication(tests.TestCase):
 
281
class TestTestScenarioApplication(TestCase):
286
282
    """Tests for the test adaption facilities."""
287
283
 
288
284
    def test_apply_scenario(self):
318
314
            adapted_test2.id())
319
315
 
320
316
 
321
 
class TestInterRepositoryScenarios(tests.TestCase):
 
317
class TestInterRepositoryScenarios(TestCase):
322
318
 
323
319
    def test_scenarios(self):
324
320
        # check that constructor parameters are passed through to the adapted
325
321
        # test.
326
 
        from bzrlib.tests.per_interrepository import make_scenarios
 
322
        from bzrlib.tests.interrepository_implementations import \
 
323
            make_scenarios
327
324
        server1 = "a"
328
325
        server2 = "b"
329
 
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
 
326
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
330
327
        scenarios = make_scenarios(server1, server2, formats)
331
328
        self.assertEqual([
332
 
            ('C0,str,str',
333
 
             {'repository_format': 'C1',
 
329
            ('str,str,str',
 
330
             {'interrepo_class': str,
 
331
              'repository_format': 'C1',
334
332
              'repository_format_to': 'C2',
335
333
              'transport_readonly_server': 'b',
336
 
              'transport_server': 'a',
337
 
              'extra_setup': 'C3'}),
338
 
            ('D0,str,str',
339
 
             {'repository_format': 'D1',
 
334
              'transport_server': 'a'}),
 
335
            ('int,str,str',
 
336
             {'interrepo_class': int,
 
337
              'repository_format': 'D1',
340
338
              'repository_format_to': 'D2',
341
339
              'transport_readonly_server': 'b',
342
 
              'transport_server': 'a',
343
 
              'extra_setup': 'D3'})],
 
340
              'transport_server': 'a'})],
344
341
            scenarios)
345
342
 
346
343
 
347
 
class TestWorkingTreeScenarios(tests.TestCase):
 
344
class TestWorkingTreeScenarios(TestCase):
348
345
 
349
346
    def test_scenarios(self):
350
347
        # check that constructor parameters are passed through to the adapted
351
348
        # test.
352
 
        from bzrlib.tests.per_workingtree import make_scenarios
 
349
        from bzrlib.tests.workingtree_implementations \
 
350
            import make_scenarios
353
351
        server1 = "a"
354
352
        server2 = "b"
355
353
        formats = [workingtree.WorkingTreeFormat2(),
369
367
            scenarios)
370
368
 
371
369
 
372
 
class TestTreeScenarios(tests.TestCase):
 
370
class TestTreeScenarios(TestCase):
373
371
 
374
372
    def test_scenarios(self):
375
373
        # the tree implementation scenario generator is meant to setup one
379
377
        # workingtree_to_test_tree attribute set to 'return_parameter' and the
380
378
        # revision one set to revision_tree_from_workingtree.
381
379
 
382
 
        from bzrlib.tests.per_tree import (
 
380
        from bzrlib.tests.tree_implementations import (
383
381
            _dirstate_tree_from_workingtree,
384
382
            make_scenarios,
385
383
            preview_tree_pre,
448
446
        self.assertEqual(expected_scenarios, scenarios)
449
447
 
450
448
 
451
 
class TestInterTreeScenarios(tests.TestCase):
 
449
class TestInterTreeScenarios(TestCase):
452
450
    """A group of tests that test the InterTreeTestAdapter."""
453
451
 
454
452
    def test_scenarios(self):
461
459
        # unlike the TestProviderAdapter we dont want to automatically add a
462
460
        # parameterized one for WorkingTree - the optimisers will tell us what
463
461
        # ones to add.
464
 
        from bzrlib.tests.per_tree import (
 
462
        from bzrlib.tests.tree_implementations import (
465
463
            return_parameter,
466
464
            revision_tree_from_workingtree
467
465
            )
468
 
        from bzrlib.tests.per_intertree import (
 
466
        from bzrlib.tests.intertree_implementations import (
469
467
            make_scenarios,
470
468
            )
471
469
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
504
502
        self.assertEqual(scenarios, expected_scenarios)
505
503
 
506
504
 
507
 
class TestTestCaseInTempDir(tests.TestCaseInTempDir):
 
505
class TestTestCaseInTempDir(TestCaseInTempDir):
508
506
 
509
507
    def test_home_is_not_working(self):
510
508
        self.assertNotEqual(self.test_dir, self.test_home_dir)
521
519
        self.assertEqualStat(real, fake)
522
520
 
523
521
    def test_assertEqualStat_notequal(self):
524
 
        self.build_tree(["foo", "longname"])
 
522
        self.build_tree(["foo", "bar"])
525
523
        self.assertRaises(AssertionError, self.assertEqualStat,
526
 
            os.lstat("foo"), os.lstat("longname"))
527
 
 
528
 
 
529
 
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
 
524
            os.lstat("foo"), os.lstat("bar"))
 
525
 
 
526
 
 
527
class TestTestCaseWithMemoryTransport(TestCaseWithMemoryTransport):
530
528
 
531
529
    def test_home_is_non_existant_dir_under_root(self):
532
530
        """The test_home_dir for TestCaseWithMemoryTransport is missing.
547
545
        cwd = osutils.getcwd()
548
546
        self.assertIsSameRealPath(self.test_dir, cwd)
549
547
 
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
548
    def test_make_branch_and_memory_tree(self):
559
549
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
560
550
 
616
606
                         self.get_transport().get_bytes(
617
607
                            'dir/.bzr/repository/format'))
618
608
 
619
 
    def test_dangling_locks_cause_failures(self):
620
 
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
621
 
            def test_function(self):
622
 
                t = self.get_transport('.')
623
 
                l = lockdir.LockDir(t, 'lock')
624
 
                l.create()
625
 
                l.attempt_lock()
626
 
        test = TestDanglingLock('test_function')
627
 
        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):
 
609
    def test_safety_net(self):
 
610
        """No test should modify the safety .bzr directory.
 
611
 
 
612
        We just test that the _check_safety_net private method raises
 
613
        AssertionError, it's easier than building a test suite with the same
 
614
        test.
 
615
        """
 
616
        # Oops, a commit in the current directory (i.e. without local .bzr
 
617
        # directory) will crawl up the hierarchy to find a .bzr directory.
 
618
        self.run_bzr(['commit', '-mfoo', '--unchanged'])
 
619
        # But we have a safety net in place.
 
620
        self.assertRaises(AssertionError, self._check_safety_net)
 
621
 
 
622
 
 
623
class TestTestCaseWithTransport(TestCaseWithTransport):
638
624
    """Tests for the convenience functions TestCaseWithTransport introduces."""
639
625
 
640
626
    def test_get_readonly_url_none(self):
 
627
        from bzrlib.transport import get_transport
 
628
        from bzrlib.transport.memory import MemoryServer
641
629
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
642
 
        self.vfs_transport_factory = memory.MemoryServer
 
630
        self.vfs_transport_factory = MemoryServer
643
631
        self.transport_readonly_server = None
644
632
        # calling get_readonly_transport() constructs a decorator on the url
645
633
        # for the server
646
634
        url = self.get_readonly_url()
647
635
        url2 = self.get_readonly_url('foo/bar')
648
 
        t = transport.get_transport(url)
649
 
        t2 = transport.get_transport(url2)
 
636
        t = get_transport(url)
 
637
        t2 = get_transport(url2)
650
638
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
651
639
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
652
640
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
653
641
 
654
642
    def test_get_readonly_url_http(self):
655
643
        from bzrlib.tests.http_server import HttpServer
 
644
        from bzrlib.transport import get_transport
 
645
        from bzrlib.transport.local import LocalURLServer
656
646
        from bzrlib.transport.http import HttpTransportBase
657
 
        self.transport_server = test_server.LocalURLServer
 
647
        self.transport_server = LocalURLServer
658
648
        self.transport_readonly_server = HttpServer
659
649
        # calling get_readonly_transport() gives us a HTTP server instance.
660
650
        url = self.get_readonly_url()
661
651
        url2 = self.get_readonly_url('foo/bar')
662
652
        # the transport returned may be any HttpTransportBase subclass
663
 
        t = transport.get_transport(url)
664
 
        t2 = transport.get_transport(url2)
 
653
        t = get_transport(url)
 
654
        t2 = get_transport(url2)
665
655
        self.failUnless(isinstance(t, HttpTransportBase))
666
656
        self.failUnless(isinstance(t2, HttpTransportBase))
667
657
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
687
677
        self.assertEqual((1, rev_id), a_branch.last_revision_info())
688
678
 
689
679
 
690
 
class TestTestCaseTransports(tests.TestCaseWithTransport):
 
680
class TestTestCaseTransports(TestCaseWithTransport):
691
681
 
692
682
    def setUp(self):
693
683
        super(TestTestCaseTransports, self).setUp()
694
 
        self.vfs_transport_factory = memory.MemoryServer
 
684
        self.vfs_transport_factory = MemoryServer
695
685
 
696
686
    def test_make_bzrdir_preserves_transport(self):
697
687
        t = self.get_transport()
698
688
        result_bzrdir = self.make_bzrdir('subdir')
699
689
        self.assertIsInstance(result_bzrdir.transport,
700
 
                              memory.MemoryTransport)
 
690
                              MemoryTransport)
701
691
        # should not be on disk, should only be in memory
702
692
        self.failIfExists('subdir')
703
693
 
704
694
 
705
 
class TestChrootedTest(tests.ChrootedTestCase):
 
695
class TestChrootedTest(ChrootedTestCase):
706
696
 
707
697
    def test_root_is_root(self):
708
 
        t = transport.get_transport(self.get_readonly_url())
 
698
        from bzrlib.transport import get_transport
 
699
        t = get_transport(self.get_readonly_url())
709
700
        url = t.base
710
701
        self.assertEqual(url, t.clone('..').base)
711
702
 
712
703
 
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):
 
704
class MockProgress(_BaseProgressBar):
 
705
    """Progress-bar standin that records calls.
 
706
 
 
707
    Useful for testing pb using code.
 
708
    """
 
709
 
 
710
    def __init__(self):
 
711
        _BaseProgressBar.__init__(self)
 
712
        self.calls = []
 
713
 
 
714
    def tick(self):
 
715
        self.calls.append(('tick',))
 
716
 
 
717
    def update(self, msg=None, current=None, total=None):
 
718
        self.calls.append(('update', msg, current, total))
 
719
 
 
720
    def clear(self):
 
721
        self.calls.append(('clear',))
 
722
 
 
723
    def note(self, msg, *args):
 
724
        self.calls.append(('note', msg, args))
 
725
 
 
726
 
 
727
class TestTestResult(TestCase):
734
728
 
735
729
    def check_timing(self, test_case, expected_re):
736
730
        result = bzrlib.tests.TextTestResult(self._log_file,
737
731
                descriptions=0,
738
732
                verbosity=1,
739
733
                )
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)
 
734
        test_case.run(result)
 
735
        timed_string = result._testTimeString(test_case)
744
736
        self.assertContainsRe(timed_string, expected_re)
745
737
 
746
738
    def test_test_reporting(self):
747
 
        class ShortDelayTestCase(tests.TestCase):
 
739
        class ShortDelayTestCase(TestCase):
748
740
            def test_short_delay(self):
749
741
                time.sleep(0.003)
750
742
            def test_short_benchmark(self):
751
743
                self.time(time.sleep, 0.003)
752
744
        self.check_timing(ShortDelayTestCase('test_short_delay'),
753
745
                          r"^ +[0-9]+ms$")
754
 
        # if a benchmark time is given, we now show just that time followed by
755
 
        # a star
 
746
        # if a benchmark time is given, we want a x of y style result.
756
747
        self.check_timing(ShortDelayTestCase('test_short_benchmark'),
757
 
                          r"^ +[0-9]+ms\*$")
 
748
                          r"^ +[0-9]+ms/ +[0-9]+ms$")
758
749
 
759
750
    def test_unittest_reporting_unittest_class(self):
760
751
        # getting the time from a non-bzrlib test works ok
764
755
        self.check_timing(ShortDelayTestCase('test_short_delay'),
765
756
                          r"^ +[0-9]+ms$")
766
757
 
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
758
    def test_assigned_benchmark_file_stores_date(self):
773
 
        self._patch_get_bzr_source_tree()
774
759
        output = StringIO()
775
760
        result = bzrlib.tests.TextTestResult(self._log_file,
776
761
                                        descriptions=0,
784
769
        self.assertContainsRe(output_string, "--date [0-9.]+")
785
770
 
786
771
    def test_benchhistory_records_test_times(self):
787
 
        self._patch_get_bzr_source_tree()
788
772
        result_stream = StringIO()
789
773
        result = bzrlib.tests.TextTestResult(
790
774
            self._log_file,
817
801
        self.requireFeature(test_lsprof.LSProfFeature)
818
802
        result_stream = StringIO()
819
803
        result = bzrlib.tests.VerboseTestResult(
820
 
            result_stream,
 
804
            unittest._WritelnDecorator(result_stream),
821
805
            descriptions=0,
822
806
            verbosity=2,
823
807
            )
851
835
 
852
836
    def test_known_failure(self):
853
837
        """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'
 
838
        class InstrumentedTestResult(ExtendedTestResult):
 
839
            def done(self): pass
 
840
            def startTests(self): pass
 
841
            def report_test_start(self, test): pass
 
842
            def report_known_failure(self, test, err):
 
843
                self._call = test, err
859
844
        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")
 
845
        def test_function():
 
846
            raise KnownFailure('failed!')
 
847
        test = unittest.FunctionTestCase(test_function)
864
848
        test.run(result)
865
849
        # it should invoke 'report_known_failure'.
866
850
        self.assertEqual(2, len(result._call))
867
 
        self.assertEqual(test.id(), result._call[0].id())
868
 
        self.assertEqual('known failure', result._call[1])
 
851
        self.assertEqual(test, result._call[0])
 
852
        self.assertEqual(KnownFailure, result._call[1][0])
 
853
        self.assertIsInstance(result._call[1][1], KnownFailure)
869
854
        # we dont introspec the traceback, if the rest is ok, it would be
870
855
        # exceptional for it not to be.
871
856
        # it should update the known_failure_count on the object.
877
862
        # verbose test output formatting
878
863
        result_stream = StringIO()
879
864
        result = bzrlib.tests.VerboseTestResult(
880
 
            result_stream,
 
865
            unittest._WritelnDecorator(result_stream),
881
866
            descriptions=0,
882
867
            verbosity=2,
883
868
            )
888
873
        # (class, exception object, traceback)
889
874
        # KnownFailures dont get their tracebacks shown though, so we
890
875
        # can skip that.
891
 
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
 
876
        err = (KnownFailure, KnownFailure('foo'), None)
892
877
        result.report_known_failure(test, err)
893
878
        output = result_stream.getvalue()[prefix:]
894
879
        lines = output.splitlines()
895
880
        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
881
        self.assertEqual(lines[1], '    foo')
900
882
        self.assertEqual(2, len(lines))
901
883
 
 
884
    def test_text_report_known_failure(self):
 
885
        # text test output formatting
 
886
        pb = MockProgress()
 
887
        result = bzrlib.tests.TextTestResult(
 
888
            StringIO(),
 
889
            descriptions=0,
 
890
            verbosity=1,
 
891
            pb=pb,
 
892
            )
 
893
        test = self.get_passing_test()
 
894
        # this seeds the state to handle reporting the test.
 
895
        result.startTest(test)
 
896
        # the err parameter has the shape:
 
897
        # (class, exception object, traceback)
 
898
        # KnownFailures dont get their tracebacks shown though, so we
 
899
        # can skip that.
 
900
        err = (KnownFailure, KnownFailure('foo'), None)
 
901
        result.report_known_failure(test, err)
 
902
        self.assertEqual(
 
903
            [
 
904
            ('update', '[1 in 0s] passing_test', None, None),
 
905
            ('note', 'XFAIL: %s\n%s\n', ('passing_test', err[1]))
 
906
            ],
 
907
            pb.calls)
 
908
        # known_failures should be printed in the summary, so if we run a test
 
909
        # after there are some known failures, the update prefix should match
 
910
        # this.
 
911
        result.known_failure_count = 3
 
912
        test.run(result)
 
913
        self.assertEqual(
 
914
            [
 
915
            ('update', '[2 in 0s] passing_test', None, None),
 
916
            ],
 
917
            pb.calls[2:])
 
918
 
902
919
    def get_passing_test(self):
903
920
        """Return a test object that can't be run usefully."""
904
921
        def passing_test():
907
924
 
908
925
    def test_add_not_supported(self):
909
926
        """Test the behaviour of invoking addNotSupported."""
910
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
911
 
            def stopTestRun(self): pass
912
 
            def report_tests_starting(self): pass
 
927
        class InstrumentedTestResult(ExtendedTestResult):
 
928
            def done(self): pass
 
929
            def startTests(self): pass
 
930
            def report_test_start(self, test): pass
913
931
            def report_unsupported(self, test, feature):
914
932
                self._call = test, feature
915
933
        result = InstrumentedTestResult(None, None, None, None)
916
934
        test = SampleTestCase('_test_pass')
917
 
        feature = tests.Feature()
 
935
        feature = Feature()
918
936
        result.startTest(test)
919
937
        result.addNotSupported(test, feature)
920
938
        # it should invoke 'report_unsupported'.
934
952
        # verbose test output formatting
935
953
        result_stream = StringIO()
936
954
        result = bzrlib.tests.VerboseTestResult(
937
 
            result_stream,
 
955
            unittest._WritelnDecorator(result_stream),
938
956
            descriptions=0,
939
957
            verbosity=2,
940
958
            )
941
959
        test = self.get_passing_test()
942
 
        feature = tests.Feature()
 
960
        feature = Feature()
943
961
        result.startTest(test)
944
962
        prefix = len(result_stream.getvalue())
945
963
        result.report_unsupported(test, feature)
946
964
        output = result_stream.getvalue()[prefix:]
947
965
        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.")
 
966
        self.assertEqual(lines, ['NODEP                   0ms', "    The feature 'Feature' is not available."])
 
967
 
 
968
    def test_text_report_unsupported(self):
 
969
        # text test output formatting
 
970
        pb = MockProgress()
 
971
        result = bzrlib.tests.TextTestResult(
 
972
            StringIO(),
 
973
            descriptions=0,
 
974
            verbosity=1,
 
975
            pb=pb,
 
976
            )
 
977
        test = self.get_passing_test()
 
978
        feature = Feature()
 
979
        # this seeds the state to handle reporting the test.
 
980
        result.startTest(test)
 
981
        result.report_unsupported(test, feature)
 
982
        # no output on unsupported features
 
983
        self.assertEqual(
 
984
            [('update', '[1 in 0s] passing_test', None, None)
 
985
            ],
 
986
            pb.calls)
 
987
        # the number of missing features should be printed in the progress
 
988
        # summary, so check for that.
 
989
        result.unsupported = {'foo':0, 'bar':0}
 
990
        test.run(result)
 
991
        self.assertEqual(
 
992
            [
 
993
            ('update', '[2 in 0s, 2 missing] passing_test', None, None),
 
994
            ],
 
995
            pb.calls[1:])
952
996
 
953
997
    def test_unavailable_exception(self):
954
998
        """An UnavailableFeature being raised should invoke addNotSupported."""
955
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
956
 
            def stopTestRun(self): pass
957
 
            def report_tests_starting(self): pass
 
999
        class InstrumentedTestResult(ExtendedTestResult):
 
1000
            def done(self): pass
 
1001
            def startTests(self): pass
 
1002
            def report_test_start(self, test): pass
958
1003
            def addNotSupported(self, test, feature):
959
1004
                self._call = test, feature
960
1005
        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")
 
1006
        feature = Feature()
 
1007
        def test_function():
 
1008
            raise UnavailableFeature(feature)
 
1009
        test = unittest.FunctionTestCase(test_function)
966
1010
        test.run(result)
967
1011
        # it should invoke 'addNotSupported'.
968
1012
        self.assertEqual(2, len(result._call))
969
 
        self.assertEqual(test.id(), result._call[0].id())
 
1013
        self.assertEqual(test, result._call[0])
970
1014
        self.assertEqual(feature, result._call[1])
971
1015
        # and not count as an error
972
1016
        self.assertEqual(0, result.error_count)
984
1028
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
985
1029
                                             verbosity=1)
986
1030
        test = self.get_passing_test()
987
 
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
988
 
        result.addExpectedFailure(test, err)
 
1031
        err = (KnownFailure, KnownFailure('foo'), None)
 
1032
        result._addKnownFailure(test, err)
989
1033
        self.assertFalse(result.wasStrictlySuccessful())
990
1034
        self.assertEqual(None, result._extractBenchmarkTime(test))
991
1035
 
999
1043
 
1000
1044
    def test_startTests(self):
1001
1045
        """Starting the first test should trigger startTests."""
1002
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
1046
        class InstrumentedTestResult(ExtendedTestResult):
1003
1047
            calls = 0
1004
1048
            def startTests(self): self.calls += 1
 
1049
            def report_test_start(self, test): pass
1005
1050
        result = InstrumentedTestResult(None, None, None, None)
1006
1051
        def test_function():
1007
1052
            pass
1009
1054
        test.run(result)
1010
1055
        self.assertEquals(1, result.calls)
1011
1056
 
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):
 
1057
 
 
1058
class TestUnicodeFilenameFeature(TestCase):
1027
1059
 
1028
1060
    def test_probe_passes(self):
1029
1061
        """UnicodeFilenameFeature._probe passes."""
1032
1064
        tests.UnicodeFilenameFeature._probe()
1033
1065
 
1034
1066
 
1035
 
class TestRunner(tests.TestCase):
 
1067
class TestRunner(TestCase):
1036
1068
 
1037
1069
    def dummy_test(self):
1038
1070
        pass
1043
1075
        This current saves and restores:
1044
1076
        TestCaseInTempDir.TEST_ROOT
1045
1077
 
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.
 
1078
        There should be no tests in this file that use bzrlib.tests.TextTestRunner
 
1079
        without using this convenience method, because of our use of global state.
1049
1080
        """
1050
 
        old_root = tests.TestCaseInTempDir.TEST_ROOT
 
1081
        old_root = TestCaseInTempDir.TEST_ROOT
1051
1082
        try:
1052
 
            tests.TestCaseInTempDir.TEST_ROOT = None
 
1083
            TestCaseInTempDir.TEST_ROOT = None
1053
1084
            return testrunner.run(test)
1054
1085
        finally:
1055
 
            tests.TestCaseInTempDir.TEST_ROOT = old_root
 
1086
            TestCaseInTempDir.TEST_ROOT = old_root
1056
1087
 
1057
1088
    def test_known_failure_failed_run(self):
1058
1089
        # run a test that generates a known failure which should be printed in
1059
1090
        # 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)
 
1091
        def known_failure_test():
 
1092
            raise KnownFailure('failed')
1063
1093
        test = unittest.TestSuite()
1064
 
        test.addTest(Test("known_failure_test"))
 
1094
        test.addTest(unittest.FunctionTestCase(known_failure_test))
1065
1095
        def failing_test():
1066
 
            self.fail('foo')
 
1096
            raise AssertionError('foo')
1067
1097
        test.addTest(unittest.FunctionTestCase(failing_test))
1068
1098
        stream = StringIO()
1069
 
        runner = tests.TextTestRunner(stream=stream)
 
1099
        runner = TextTestRunner(stream=stream)
1070
1100
        result = self.run_test_runner(runner, test)
1071
1101
        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
 
            )
 
1102
        self.assertEqual([
 
1103
            '',
 
1104
            '======================================================================',
 
1105
            'FAIL: unittest.FunctionTestCase (failing_test)',
 
1106
            '----------------------------------------------------------------------',
 
1107
            'Traceback (most recent call last):',
 
1108
            '    raise AssertionError(\'foo\')',
 
1109
            'AssertionError: foo',
 
1110
            '',
 
1111
            '----------------------------------------------------------------------',
 
1112
            '',
 
1113
            'FAILED (failures=1, known_failure_count=1)'],
 
1114
            lines[3:8] + lines[9:13] + lines[14:])
1086
1115
 
1087
1116
    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")
 
1117
        # run a test that generates a known failure which should be printed in the final output.
 
1118
        def known_failure_test():
 
1119
            raise KnownFailure('failed')
 
1120
        test = unittest.FunctionTestCase(known_failure_test)
1094
1121
        stream = StringIO()
1095
 
        runner = tests.TextTestRunner(stream=stream)
 
1122
        runner = TextTestRunner(stream=stream)
1096
1123
        result = self.run_test_runner(runner, test)
1097
1124
        self.assertContainsRe(stream.getvalue(),
1098
1125
            '\n'
1101
1128
            '\n'
1102
1129
            'OK \\(known_failures=1\\)\n')
1103
1130
 
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
1131
    def test_skipped_test(self):
1119
1132
        # run a test that is skipped, and check the suite as a whole still
1120
1133
        # succeeds.
1121
1134
        # skipping_test must be hidden in here so it's not run as a real test
1122
 
        class SkippingTest(tests.TestCase):
 
1135
        class SkippingTest(TestCase):
1123
1136
            def skipping_test(self):
1124
 
                raise tests.TestSkipped('test intentionally skipped')
1125
 
        runner = tests.TextTestRunner(stream=self._log_file)
 
1137
                raise TestSkipped('test intentionally skipped')
 
1138
        runner = TextTestRunner(stream=self._log_file)
1126
1139
        test = SkippingTest("skipping_test")
1127
1140
        result = self.run_test_runner(runner, test)
1128
1141
        self.assertTrue(result.wasSuccessful())
1129
1142
 
1130
1143
    def test_skipped_from_setup(self):
1131
1144
        calls = []
1132
 
        class SkippedSetupTest(tests.TestCase):
 
1145
        class SkippedSetupTest(TestCase):
1133
1146
 
1134
1147
            def setUp(self):
1135
1148
                calls.append('setUp')
1136
1149
                self.addCleanup(self.cleanup)
1137
 
                raise tests.TestSkipped('skipped setup')
 
1150
                raise TestSkipped('skipped setup')
1138
1151
 
1139
1152
            def test_skip(self):
1140
1153
                self.fail('test reached')
1142
1155
            def cleanup(self):
1143
1156
                calls.append('cleanup')
1144
1157
 
1145
 
        runner = tests.TextTestRunner(stream=self._log_file)
 
1158
        runner = TextTestRunner(stream=self._log_file)
1146
1159
        test = SkippedSetupTest('test_skip')
1147
1160
        result = self.run_test_runner(runner, test)
1148
1161
        self.assertTrue(result.wasSuccessful())
1151
1164
 
1152
1165
    def test_skipped_from_test(self):
1153
1166
        calls = []
1154
 
        class SkippedTest(tests.TestCase):
 
1167
        class SkippedTest(TestCase):
1155
1168
 
1156
1169
            def setUp(self):
1157
 
                tests.TestCase.setUp(self)
 
1170
                TestCase.setUp(self)
1158
1171
                calls.append('setUp')
1159
1172
                self.addCleanup(self.cleanup)
1160
1173
 
1161
1174
            def test_skip(self):
1162
 
                raise tests.TestSkipped('skipped test')
 
1175
                raise TestSkipped('skipped test')
1163
1176
 
1164
1177
            def cleanup(self):
1165
1178
                calls.append('cleanup')
1166
1179
 
1167
 
        runner = tests.TextTestRunner(stream=self._log_file)
 
1180
        runner = TextTestRunner(stream=self._log_file)
1168
1181
        test = SkippedTest('test_skip')
1169
1182
        result = self.run_test_runner(runner, test)
1170
1183
        self.assertTrue(result.wasSuccessful())
1173
1186
 
1174
1187
    def test_not_applicable(self):
1175
1188
        # 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')
 
1189
        def not_applicable_test():
 
1190
            from bzrlib.tests import TestNotApplicable
 
1191
            raise TestNotApplicable('this test never runs')
1179
1192
        out = StringIO()
1180
 
        runner = tests.TextTestRunner(stream=out, verbosity=2)
1181
 
        test = Test("not_applicable_test")
 
1193
        runner = TextTestRunner(stream=out, verbosity=2)
 
1194
        test = unittest.FunctionTestCase(not_applicable_test)
1182
1195
        result = self.run_test_runner(runner, test)
1183
1196
        self._log_file.write(out.getvalue())
1184
1197
        self.assertTrue(result.wasSuccessful())
1188
1201
        self.assertContainsRe(out.getvalue(),
1189
1202
                r'(?m)^    this test never runs')
1190
1203
 
 
1204
    def test_not_applicable_demo(self):
 
1205
        # just so you can see it in the test output
 
1206
        raise TestNotApplicable('this test is just a demonstation')
 
1207
 
1191
1208
    def test_unsupported_features_listed(self):
1192
1209
        """When unsupported features are encountered they are detailed."""
1193
 
        class Feature1(tests.Feature):
 
1210
        class Feature1(Feature):
1194
1211
            def _probe(self): return False
1195
 
        class Feature2(tests.Feature):
 
1212
        class Feature2(Feature):
1196
1213
            def _probe(self): return False
1197
1214
        # create sample tests
1198
1215
        test1 = SampleTestCase('_test_pass')
1203
1220
        test.addTest(test1)
1204
1221
        test.addTest(test2)
1205
1222
        stream = StringIO()
1206
 
        runner = tests.TextTestRunner(stream=stream)
 
1223
        runner = TextTestRunner(stream=stream)
1207
1224
        result = self.run_test_runner(runner, test)
1208
1225
        lines = stream.getvalue().splitlines()
1209
1226
        self.assertEqual([
1213
1230
            ],
1214
1231
            lines[-3:])
1215
1232
 
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
1233
    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()
 
1234
        # tests that the running the benchmark produces a history file
 
1235
        # containing a timestamp and the revision id of the bzrlib source which
 
1236
        # was tested.
 
1237
        workingtree = _get_bzr_source_tree()
1230
1238
        test = TestRunner('dummy_test')
1231
1239
        output = StringIO()
1232
 
        runner = tests.TextTestRunner(stream=self._log_file,
1233
 
                                      bench_history=output)
 
1240
        runner = TextTestRunner(stream=self._log_file, bench_history=output)
1234
1241
        result = self.run_test_runner(runner, test)
1235
1242
        output_string = output.getvalue()
1236
1243
        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):
 
1244
        if workingtree is not None:
 
1245
            revision_id = workingtree.get_parent_ids()[0]
 
1246
            self.assertEndsWith(output_string.rstrip(), revision_id)
 
1247
 
 
1248
    def assertLogDeleted(self, test):
 
1249
        log = test._get_log()
 
1250
        self.assertEqual("DELETED log file to reduce memory footprint", log)
 
1251
        self.assertEqual('', test._log_contents)
 
1252
        self.assertIs(None, test._log_file_name)
 
1253
 
 
1254
    def test_success_log_deleted(self):
 
1255
        """Successful tests have their log deleted"""
 
1256
 
 
1257
        class LogTester(TestCase):
 
1258
 
 
1259
            def test_success(self):
 
1260
                self.log('this will be removed\n')
 
1261
 
 
1262
        sio = cStringIO.StringIO()
 
1263
        runner = TextTestRunner(stream=sio)
 
1264
        test = LogTester('test_success')
 
1265
        result = self.run_test_runner(runner, test)
 
1266
 
 
1267
        self.assertLogDeleted(test)
 
1268
 
 
1269
    def test_skipped_log_deleted(self):
 
1270
        """Skipped tests have their log deleted"""
 
1271
 
 
1272
        class LogTester(TestCase):
 
1273
 
 
1274
            def test_skipped(self):
 
1275
                self.log('this will be removed\n')
 
1276
                raise tests.TestSkipped()
 
1277
 
 
1278
        sio = cStringIO.StringIO()
 
1279
        runner = TextTestRunner(stream=sio)
 
1280
        test = LogTester('test_skipped')
 
1281
        result = self.run_test_runner(runner, test)
 
1282
 
 
1283
        self.assertLogDeleted(test)
 
1284
 
 
1285
    def test_not_aplicable_log_deleted(self):
 
1286
        """Not applicable tests have their log deleted"""
 
1287
 
 
1288
        class LogTester(TestCase):
 
1289
 
 
1290
            def test_not_applicable(self):
 
1291
                self.log('this will be removed\n')
 
1292
                raise tests.TestNotApplicable()
 
1293
 
 
1294
        sio = cStringIO.StringIO()
 
1295
        runner = TextTestRunner(stream=sio)
 
1296
        test = LogTester('test_not_applicable')
 
1297
        result = self.run_test_runner(runner, test)
 
1298
 
 
1299
        self.assertLogDeleted(test)
 
1300
 
 
1301
    def test_known_failure_log_deleted(self):
 
1302
        """Know failure tests have their log deleted"""
 
1303
 
 
1304
        class LogTester(TestCase):
 
1305
 
 
1306
            def test_known_failure(self):
 
1307
                self.log('this will be removed\n')
 
1308
                raise tests.KnownFailure()
 
1309
 
 
1310
        sio = cStringIO.StringIO()
 
1311
        runner = TextTestRunner(stream=sio)
 
1312
        test = LogTester('test_known_failure')
 
1313
        result = self.run_test_runner(runner, test)
 
1314
 
 
1315
        self.assertLogDeleted(test)
 
1316
 
 
1317
    def test_fail_log_kept(self):
 
1318
        """Failed tests have their log kept"""
 
1319
 
 
1320
        class LogTester(TestCase):
 
1321
 
 
1322
            def test_fail(self):
 
1323
                self.log('this will be kept\n')
 
1324
                self.fail('this test fails')
 
1325
 
 
1326
        sio = cStringIO.StringIO()
 
1327
        runner = TextTestRunner(stream=sio)
 
1328
        test = LogTester('test_fail')
 
1329
        result = self.run_test_runner(runner, test)
 
1330
 
 
1331
        text = sio.getvalue()
 
1332
        self.assertContainsRe(text, 'this will be kept')
 
1333
        self.assertContainsRe(text, 'this test fails')
 
1334
 
 
1335
        log = test._get_log()
 
1336
        self.assertContainsRe(log, 'this will be kept')
 
1337
        self.assertEqual(log, test._log_contents)
 
1338
 
 
1339
    def test_error_log_kept(self):
 
1340
        """Tests with errors have their log kept"""
 
1341
 
 
1342
        class LogTester(TestCase):
 
1343
 
 
1344
            def test_error(self):
 
1345
                self.log('this will be kept\n')
 
1346
                raise ValueError('random exception raised')
 
1347
 
 
1348
        sio = cStringIO.StringIO()
 
1349
        runner = TextTestRunner(stream=sio)
 
1350
        test = LogTester('test_error')
 
1351
        result = self.run_test_runner(runner, test)
 
1352
 
 
1353
        text = sio.getvalue()
 
1354
        self.assertContainsRe(text, 'this will be kept')
 
1355
        self.assertContainsRe(text, 'random exception raised')
 
1356
 
 
1357
        log = test._get_log()
 
1358
        self.assertContainsRe(log, 'this will be kept')
 
1359
        self.assertEqual(log, test._log_contents)
 
1360
 
 
1361
 
 
1362
class SampleTestCase(TestCase):
1281
1363
 
1282
1364
    def _test_pass(self):
1283
1365
        pass
1285
1367
class _TestException(Exception):
1286
1368
    pass
1287
1369
 
1288
 
 
1289
 
class TestTestCase(tests.TestCase):
 
1370
class TestTestCase(TestCase):
1290
1371
    """Tests that test the core bzrlib TestCase."""
1291
1372
 
1292
1373
    def test_assertLength_matches_empty(self):
1309
1390
            exception.args[0])
1310
1391
 
1311
1392
    def test_base_setUp_not_called_causes_failure(self):
1312
 
        class TestCaseWithBrokenSetUp(tests.TestCase):
 
1393
        class TestCaseWithBrokenSetUp(TestCase):
1313
1394
            def setUp(self):
1314
1395
                pass # does not call TestCase.setUp
1315
1396
            def test_foo(self):
1321
1402
        self.assertEqual(1, result.testsRun)
1322
1403
 
1323
1404
    def test_base_tearDown_not_called_causes_failure(self):
1324
 
        class TestCaseWithBrokenTearDown(tests.TestCase):
 
1405
        class TestCaseWithBrokenTearDown(TestCase):
1325
1406
            def tearDown(self):
1326
1407
                pass # does not call TestCase.tearDown
1327
1408
            def test_foo(self):
1335
1416
    def test_debug_flags_sanitised(self):
1336
1417
        """The bzrlib debug flags should be sanitised by setUp."""
1337
1418
        if 'allow_debug' in tests.selftest_debug_flags:
1338
 
            raise tests.TestNotApplicable(
 
1419
            raise TestNotApplicable(
1339
1420
                '-Eallow_debug option prevents debug flag sanitisation')
1340
1421
        # we could set something and run a test that will check
1341
1422
        # it gets santised, but this is probably sufficient for now:
1342
1423
        # 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)
 
1424
        self.assertEqual(set(), bzrlib.debug.debug_flags)
1347
1425
 
1348
1426
    def change_selftest_debug_flags(self, new_flags):
1349
 
        self.overrideAttr(tests, 'selftest_debug_flags', set(new_flags))
 
1427
        orig_selftest_flags = tests.selftest_debug_flags
 
1428
        self.addCleanup(self._restore_selftest_debug_flags, orig_selftest_flags)
 
1429
        tests.selftest_debug_flags = set(new_flags)
 
1430
 
 
1431
    def _restore_selftest_debug_flags(self, flags):
 
1432
        tests.selftest_debug_flags = flags
1350
1433
 
1351
1434
    def test_allow_debug_flag(self):
1352
1435
        """The -Eallow_debug flag prevents bzrlib.debug.debug_flags from being
1354
1437
        """
1355
1438
        self.change_selftest_debug_flags(set(['allow_debug']))
1356
1439
        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)
 
1440
        class TestThatRecordsFlags(TestCase):
 
1441
            def test_foo(nested_self):
 
1442
                self.flags = set(bzrlib.debug.debug_flags)
 
1443
        test = TestThatRecordsFlags('test_foo')
 
1444
        test.run(self.make_test_result())
 
1445
        self.assertEqual(set(['a-flag']), self.flags)
1399
1446
 
1400
1447
    def test_debug_flags_restored(self):
1401
1448
        """The bzrlib debug flags should be restored to their original state
1404
1451
        self.change_selftest_debug_flags(set(['allow_debug']))
1405
1452
        # Now run a test that modifies debug.debug_flags.
1406
1453
        bzrlib.debug.debug_flags = set(['original-state'])
1407
 
        class TestThatModifiesFlags(tests.TestCase):
 
1454
        class TestThatModifiesFlags(TestCase):
1408
1455
            def test_foo(self):
1409
1456
                bzrlib.debug.debug_flags = set(['modified'])
1410
1457
        test = TestThatModifiesFlags('test_foo')
1412
1459
        self.assertEqual(set(['original-state']), bzrlib.debug.debug_flags)
1413
1460
 
1414
1461
    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)
 
1462
        return bzrlib.tests.TextTestResult(
 
1463
            self._log_file, descriptions=0, verbosity=1)
1417
1464
 
1418
1465
    def inner_test(self):
1419
1466
        # the inner child test
1426
1473
        result = self.make_test_result()
1427
1474
        self.inner_test.run(result)
1428
1475
        note("outer finish")
1429
 
        self.addCleanup(osutils.delete_any, self._log_file_name)
1430
1476
 
1431
1477
    def test_trace_nesting(self):
1432
1478
        # this tests that each test case nests its trace facility correctly.
1456
1502
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1457
1503
        output_stream = StringIO()
1458
1504
        result = bzrlib.tests.VerboseTestResult(
1459
 
            output_stream,
 
1505
            unittest._WritelnDecorator(output_stream),
1460
1506
            descriptions=0,
1461
 
            verbosity=2)
 
1507
            verbosity=2,
 
1508
            num_tests=sample_test.countTestCases())
1462
1509
        sample_test.run(result)
1463
1510
        self.assertContainsRe(
1464
1511
            output_stream.getvalue(),
1465
 
            r"\d+ms\*\n$")
 
1512
            r"\d+ms/ +\d+ms\n$")
1466
1513
 
1467
1514
    def test_hooks_sanitised(self):
1468
1515
        """The bzrlib hooks should be sanitised by setUp."""
1492
1539
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1493
1540
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1494
1541
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
1495
 
        del self._benchcalls[:]
1496
1542
 
1497
1543
    def test_knownFailure(self):
1498
1544
        """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)
 
1545
        self.assertRaises(KnownFailure, self.knownFailure, "A Failure")
1517
1546
 
1518
1547
    def test_requireFeature_available(self):
1519
1548
        """self.requireFeature(available) is a no-op."""
1520
 
        class Available(tests.Feature):
 
1549
        class Available(Feature):
1521
1550
            def _probe(self):return True
1522
1551
        feature = Available()
1523
1552
        self.requireFeature(feature)
1524
1553
 
1525
1554
    def test_requireFeature_unavailable(self):
1526
1555
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1527
 
        class Unavailable(tests.Feature):
 
1556
        class Unavailable(Feature):
1528
1557
            def _probe(self):return False
1529
1558
        feature = Unavailable()
1530
 
        self.assertRaises(tests.UnavailableFeature,
1531
 
                          self.requireFeature, feature)
 
1559
        self.assertRaises(UnavailableFeature, self.requireFeature, feature)
1532
1560
 
1533
1561
    def test_run_no_parameters(self):
1534
1562
        test = SampleTestCase('_test_pass')
1564
1592
        """Test disabled tests behaviour with support aware results."""
1565
1593
        test = SampleTestCase('_test_pass')
1566
1594
        class DisabledFeature(object):
1567
 
            def __eq__(self, other):
1568
 
                return isinstance(other, DisabledFeature)
1569
1595
            def available(self):
1570
1596
                return False
1571
1597
        the_feature = DisabledFeature()
1582
1608
                self.calls.append(('addNotSupported', test, feature))
1583
1609
        result = InstrumentedTestResult()
1584
1610
        test.run(result)
1585
 
        case = result.calls[0][1]
1586
1611
        self.assertEqual([
1587
 
            ('startTest', case),
1588
 
            ('addNotSupported', case, the_feature),
1589
 
            ('stopTest', case),
 
1612
            ('startTest', test),
 
1613
            ('addNotSupported', test, the_feature),
 
1614
            ('stopTest', test),
1590
1615
            ],
1591
1616
            result.calls)
1592
1617
 
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
1618
    def test_assert_list_raises_on_generator(self):
1602
1619
        def generator_which_will_raise():
1603
1620
            # This will not raise until after the first yield
1653
1670
        self.assertRaises(AssertionError,
1654
1671
            self.assertListRaises, _TestException, success_generator)
1655
1672
 
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
1673
 
1829
1674
# NB: Don't delete this; it's not actually from 0.11!
1830
1675
@deprecated_function(deprecated_in((0, 11, 0)))
1853
1698
        return sample_deprecated_function()
1854
1699
 
1855
1700
 
1856
 
class TestExtraAssertions(tests.TestCase):
 
1701
class TestExtraAssertions(TestCase):
1857
1702
    """Tests for new test assertions in bzrlib test suite"""
1858
1703
 
1859
1704
    def test_assert_isinstance(self):
1860
1705
        self.assertIsInstance(2, int)
1861
1706
        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'>")
 
1707
        self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1865
1708
        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
1709
 
1872
1710
    def test_assertEndsWith(self):
1873
1711
        self.assertEndsWith('foo', 'oo')
1874
1712
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1875
1713
 
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
1714
    def test_applyDeprecated_not_deprecated(self):
1892
1715
        sample_object = ApplyDeprecatedHelper()
1893
1716
        # calling an undeprecated callable raises an assertion
1936
1759
        self.callDeprecated([], testfunc, be_deprecated=False)
1937
1760
 
1938
1761
 
1939
 
class TestWarningTests(tests.TestCase):
 
1762
class TestWarningTests(TestCase):
1940
1763
    """Tests for calling methods that raise warnings."""
1941
1764
 
1942
1765
    def test_callCatchWarnings(self):
1952
1775
        self.assertEquals("this is your last warning", str(w0))
1953
1776
 
1954
1777
 
1955
 
class TestConvenienceMakers(tests.TestCaseWithTransport):
 
1778
class TestConvenienceMakers(TestCaseWithTransport):
1956
1779
    """Test for the make_* convenience functions."""
1957
1780
 
1958
1781
    def test_make_branch_and_tree_with_format(self):
1970
1793
        tree = self.make_branch_and_memory_tree('a')
1971
1794
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1972
1795
 
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://'))
 
1796
 
 
1797
class TestSFTPMakeBranchAndTree(TestCaseWithSFTPServer):
 
1798
 
 
1799
    def test_make_tree_for_sftp_branch(self):
 
1800
        """Transports backed by local directories create local trees."""
 
1801
 
1979
1802
        tree = self.make_branch_and_tree('t1')
1980
1803
        base = tree.bzrdir.root_transport.base
1981
 
        self.assertStartsWith(base, 'file://')
 
1804
        self.failIf(base.startswith('sftp'),
 
1805
                'base %r is on sftp but should be local' % base)
1982
1806
        self.assertEquals(tree.bzrdir.root_transport,
1983
1807
                tree.branch.bzrdir.root_transport)
1984
1808
        self.assertEquals(tree.bzrdir.root_transport,
1985
1809
                tree.branch.repository.bzrdir.root_transport)
1986
1810
 
1987
1811
 
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):
 
1812
class TestSelftest(TestCase):
2006
1813
    """Tests of bzrlib.tests.selftest."""
2007
1814
 
2008
1815
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
2009
1816
        factory_called = []
2010
1817
        def factory():
2011
1818
            factory_called.append(True)
2012
 
            return TestUtil.TestSuite()
 
1819
            return TestSuite()
2013
1820
        out = StringIO()
2014
1821
        err = StringIO()
2015
1822
        self.apply_redirected(out, err, None, bzrlib.tests.selftest,
2016
1823
            test_suite_factory=factory)
2017
1824
        self.assertEqual([True], factory_called)
2018
1825
 
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
 
1826
 
 
1827
class TestKnownFailure(TestCase):
 
1828
 
 
1829
    def test_known_failure(self):
 
1830
        """Check that KnownFailure is defined appropriately."""
 
1831
        # a KnownFailure is an assertion error for compatability with unaware
 
1832
        # runners.
 
1833
        self.assertIsInstance(KnownFailure(""), AssertionError)
 
1834
 
 
1835
    def test_expect_failure(self):
 
1836
        try:
 
1837
            self.expectFailure("Doomed to failure", self.assertTrue, False)
 
1838
        except KnownFailure, e:
 
1839
            self.assertEqual('Doomed to failure', e.args[0])
 
1840
        try:
 
1841
            self.expectFailure("Doomed to failure", self.assertTrue, True)
 
1842
        except AssertionError, e:
 
1843
            self.assertEqual('Unexpected success.  Should have failed:'
 
1844
                             ' Doomed to failure', e.args[0])
2429
1845
        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):
 
1846
            self.fail('Assertion not raised')
 
1847
 
 
1848
 
 
1849
class TestFeature(TestCase):
2614
1850
 
2615
1851
    def test_caching(self):
2616
1852
        """Feature._probe is called by the feature at most once."""
2617
 
        class InstrumentedFeature(tests.Feature):
 
1853
        class InstrumentedFeature(Feature):
2618
1854
            def __init__(self):
2619
 
                super(InstrumentedFeature, self).__init__()
 
1855
                Feature.__init__(self)
2620
1856
                self.calls = []
2621
1857
            def _probe(self):
2622
1858
                self.calls.append('_probe')
2629
1865
 
2630
1866
    def test_named_str(self):
2631
1867
        """Feature.__str__ should thunk to feature_name()."""
2632
 
        class NamedFeature(tests.Feature):
 
1868
        class NamedFeature(Feature):
2633
1869
            def feature_name(self):
2634
1870
                return 'symlinks'
2635
1871
        feature = NamedFeature()
2637
1873
 
2638
1874
    def test_default_str(self):
2639
1875
        """Feature.__str__ should default to __class__.__name__."""
2640
 
        class NamedFeature(tests.Feature):
 
1876
        class NamedFeature(Feature):
2641
1877
            pass
2642
1878
        feature = NamedFeature()
2643
1879
        self.assertEqual('NamedFeature', str(feature))
2644
1880
 
2645
1881
 
2646
 
class TestUnavailableFeature(tests.TestCase):
 
1882
class TestUnavailableFeature(TestCase):
2647
1883
 
2648
1884
    def test_access_feature(self):
2649
 
        feature = tests.Feature()
2650
 
        exception = tests.UnavailableFeature(feature)
 
1885
        feature = Feature()
 
1886
        exception = UnavailableFeature(feature)
2651
1887
        self.assertIs(feature, exception.args[0])
2652
1888
 
2653
1889
 
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):
 
1890
class TestSelftestFiltering(TestCase):
2688
1891
 
2689
1892
    def setUp(self):
2690
 
        tests.TestCase.setUp(self)
 
1893
        TestCase.setUp(self)
2691
1894
        self.suite = TestUtil.TestSuite()
2692
1895
        self.loader = TestUtil.TestLoader()
2693
 
        self.suite.addTest(self.loader.loadTestsFromModule(
2694
 
            sys.modules['bzrlib.tests.test_selftest']))
 
1896
        self.suite.addTest(self.loader.loadTestsFromModuleNames([
 
1897
            'bzrlib.tests.test_selftest']))
2695
1898
        self.all_names = _test_ids(self.suite)
2696
1899
 
2697
1900
    def test_condition_id_re(self):
2698
1901
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2699
1902
            'test_condition_id_re')
2700
 
        filtered_suite = tests.filter_suite_by_condition(
2701
 
            self.suite, tests.condition_id_re('test_condition_id_re'))
 
1903
        filtered_suite = filter_suite_by_condition(self.suite,
 
1904
            condition_id_re('test_condition_id_re'))
2702
1905
        self.assertEqual([test_name], _test_ids(filtered_suite))
2703
1906
 
2704
1907
    def test_condition_id_in_list(self):
2705
1908
        test_names = ['bzrlib.tests.test_selftest.TestSelftestFiltering.'
2706
1909
                      'test_condition_id_in_list']
2707
1910
        id_list = tests.TestIdList(test_names)
2708
 
        filtered_suite = tests.filter_suite_by_condition(
 
1911
        filtered_suite = filter_suite_by_condition(
2709
1912
            self.suite, tests.condition_id_in_list(id_list))
2710
1913
        my_pattern = 'TestSelftestFiltering.*test_condition_id_in_list'
2711
 
        re_filtered = tests.filter_suite_by_re(self.suite, my_pattern)
 
1914
        re_filtered = filter_suite_by_re(self.suite, my_pattern)
2712
1915
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
2713
1916
 
2714
1917
    def test_condition_id_startswith(self):
2718
1921
        test_names = [ klass + 'test_condition_id_in_list',
2719
1922
                      klass + 'test_condition_id_startswith',
2720
1923
                     ]
2721
 
        filtered_suite = tests.filter_suite_by_condition(
 
1924
        filtered_suite = filter_suite_by_condition(
2722
1925
            self.suite, tests.condition_id_startswith([start1, start2]))
2723
1926
        self.assertEqual(test_names, _test_ids(filtered_suite))
2724
1927
 
2725
1928
    def test_condition_isinstance(self):
2726
 
        filtered_suite = tests.filter_suite_by_condition(
2727
 
            self.suite, tests.condition_isinstance(self.__class__))
 
1929
        filtered_suite = filter_suite_by_condition(self.suite,
 
1930
            condition_isinstance(self.__class__))
2728
1931
        class_pattern = 'bzrlib.tests.test_selftest.TestSelftestFiltering.'
2729
 
        re_filtered = tests.filter_suite_by_re(self.suite, class_pattern)
 
1932
        re_filtered = filter_suite_by_re(self.suite, class_pattern)
2730
1933
        self.assertEqual(_test_ids(re_filtered), _test_ids(filtered_suite))
2731
1934
 
2732
1935
    def test_exclude_tests_by_condition(self):
2733
1936
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2734
1937
            'test_exclude_tests_by_condition')
2735
 
        filtered_suite = tests.exclude_tests_by_condition(self.suite,
 
1938
        filtered_suite = exclude_tests_by_condition(self.suite,
2736
1939
            lambda x:x.id() == excluded_name)
2737
1940
        self.assertEqual(len(self.all_names) - 1,
2738
1941
            filtered_suite.countTestCases())
2743
1946
 
2744
1947
    def test_exclude_tests_by_re(self):
2745
1948
        self.all_names = _test_ids(self.suite)
2746
 
        filtered_suite = tests.exclude_tests_by_re(self.suite,
2747
 
                                                   'exclude_tests_by_re')
 
1949
        filtered_suite = exclude_tests_by_re(self.suite, 'exclude_tests_by_re')
2748
1950
        excluded_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2749
1951
            'test_exclude_tests_by_re')
2750
1952
        self.assertEqual(len(self.all_names) - 1,
2757
1959
    def test_filter_suite_by_condition(self):
2758
1960
        test_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2759
1961
            'test_filter_suite_by_condition')
2760
 
        filtered_suite = tests.filter_suite_by_condition(self.suite,
 
1962
        filtered_suite = filter_suite_by_condition(self.suite,
2761
1963
            lambda x:x.id() == test_name)
2762
1964
        self.assertEqual([test_name], _test_ids(filtered_suite))
2763
1965
 
2764
1966
    def test_filter_suite_by_re(self):
2765
 
        filtered_suite = tests.filter_suite_by_re(self.suite,
2766
 
                                                  'test_filter_suite_by_r')
 
1967
        filtered_suite = filter_suite_by_re(self.suite, 'test_filter_suite_by_r')
2767
1968
        filtered_names = _test_ids(filtered_suite)
2768
1969
        self.assertEqual(filtered_names, ['bzrlib.tests.test_selftest.'
2769
1970
            'TestSelftestFiltering.test_filter_suite_by_re'])
2797
1998
 
2798
1999
    def test_preserve_input(self):
2799
2000
        # 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("@#$"))
 
2001
        self.assertTrue(self.suite is preserve_input(self.suite))
 
2002
        self.assertTrue("@#$" is preserve_input("@#$"))
2802
2003
 
2803
2004
    def test_randomize_suite(self):
2804
 
        randomized_suite = tests.randomize_suite(self.suite)
 
2005
        randomized_suite = randomize_suite(self.suite)
2805
2006
        # randomizing should not add or remove test names.
2806
2007
        self.assertEqual(set(_test_ids(self.suite)),
2807
2008
                         set(_test_ids(randomized_suite)))
2817
2018
 
2818
2019
    def test_split_suit_by_condition(self):
2819
2020
        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)
 
2021
        condition = condition_id_re('test_filter_suite_by_r')
 
2022
        split_suite = split_suite_by_condition(self.suite, condition)
2822
2023
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2823
2024
            'test_filter_suite_by_re')
2824
2025
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
2829
2030
 
2830
2031
    def test_split_suit_by_re(self):
2831
2032
        self.all_names = _test_ids(self.suite)
2832
 
        split_suite = tests.split_suite_by_re(self.suite,
2833
 
                                              'test_filter_suite_by_r')
 
2033
        split_suite = split_suite_by_re(self.suite, 'test_filter_suite_by_r')
2834
2034
        filtered_name = ('bzrlib.tests.test_selftest.TestSelftestFiltering.'
2835
2035
            'test_filter_suite_by_re')
2836
2036
        self.assertEqual([filtered_name], _test_ids(split_suite[0]))
2840
2040
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2841
2041
 
2842
2042
 
2843
 
class TestCheckInventoryShape(tests.TestCaseWithTransport):
 
2043
class TestCheckInventoryShape(TestCaseWithTransport):
2844
2044
 
2845
2045
    def test_check_inventory_shape(self):
2846
2046
        files = ['a', 'b/', 'b/c']
2854
2054
            tree.unlock()
2855
2055
 
2856
2056
 
2857
 
class TestBlackboxSupport(tests.TestCase):
 
2057
class TestBlackboxSupport(TestCase):
2858
2058
    """Tests for testsuite blackbox features."""
2859
2059
 
2860
2060
    def test_run_bzr_failure_not_caught(self):
2868
2068
        # the test framework
2869
2069
        self.assertEquals('always fails', str(e))
2870
2070
        # check that there's no traceback in the test log
2871
 
        self.assertNotContainsRe(self.get_log(), r'Traceback')
 
2071
        self.assertNotContainsRe(self._get_log(keep_log_file=True),
 
2072
            r'Traceback')
2872
2073
 
2873
2074
    def test_run_bzr_user_error_caught(self):
2874
2075
        # Running bzr in blackbox mode, normal/expected/user errors should be
2875
2076
        # caught in the regular way and turned into an error message plus exit
2876
2077
        # 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)
 
2078
        out, err = self.run_bzr(["log", "/nonexistantpath"], retcode=3)
2883
2079
        self.assertEqual(out, '')
2884
2080
        self.assertContainsRe(err,
2885
2081
            'bzr: ERROR: Not a branch: ".*nonexistantpath/".\n')
2886
2082
 
2887
2083
 
2888
 
class TestTestLoader(tests.TestCase):
 
2084
class TestTestLoader(TestCase):
2889
2085
    """Tests for the test loader."""
2890
2086
 
2891
2087
    def _get_loader_and_module(self):
2892
2088
        """Gets a TestLoader and a module with one test in it."""
2893
2089
        loader = TestUtil.TestLoader()
2894
2090
        module = {}
2895
 
        class Stub(tests.TestCase):
 
2091
        class Stub(TestCase):
2896
2092
            def test_foo(self):
2897
2093
                pass
2898
2094
        class MyModule(object):
2911
2107
        # load_tests do not need that :)
2912
2108
        def load_tests(self, standard_tests, module, loader):
2913
2109
            result = loader.suiteClass()
2914
 
            for test in tests.iter_suite_tests(standard_tests):
 
2110
            for test in iter_suite_tests(standard_tests):
2915
2111
                result.addTests([test, test])
2916
2112
            return result
2917
2113
        # add a load_tests() method which multiplies the tests from the module.
2936
2132
 
2937
2133
    def _create_suite(self, test_id_list):
2938
2134
 
2939
 
        class Stub(tests.TestCase):
 
2135
        class Stub(TestCase):
2940
2136
            def test_foo(self):
2941
2137
                pass
2942
2138
 
2952
2148
 
2953
2149
    def _test_ids(self, test_suite):
2954
2150
        """Get the ids for the tests in a test suite."""
2955
 
        return [t.id() for t in tests.iter_suite_tests(test_suite)]
 
2151
        return [t.id() for t in iter_suite_tests(test_suite)]
2956
2152
 
2957
2153
    def test_empty_list(self):
2958
2154
        id_list = self._create_id_list([])
2984
2180
        self.assertTrue(id_list.refers_to('mod.class'))
2985
2181
        self.assertTrue(id_list.refers_to('mod.class.meth'))
2986
2182
 
 
2183
    def test_test_suite(self):
 
2184
        # This test is slow, so we do a single test with one test in each
 
2185
        # category
 
2186
        test_list = [
 
2187
            # testmod_names
 
2188
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
 
2189
            'bzrlib.tests.test_selftest.TestTestIdList.test_test_suite',
 
2190
            # transport implementations
 
2191
            'bzrlib.tests.test_transport_implementations.TransportTests'
 
2192
            '.test_abspath(LocalURLServer)',
 
2193
            # modules_to_doctest
 
2194
            'bzrlib.timestamp.format_highres_date',
 
2195
            # plugins can't be tested that way since selftest may be run with
 
2196
            # --no-plugins
 
2197
            ]
 
2198
        suite = tests.test_suite(test_list)
 
2199
        self.assertEquals(test_list, _test_ids(suite))
 
2200
 
2987
2201
    def test_test_suite_matches_id_list_with_unknown(self):
2988
2202
        loader = TestUtil.TestLoader()
2989
2203
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2997
2211
        loader = TestUtil.TestLoader()
2998
2212
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2999
2213
        dupes = loader.suiteClass()
3000
 
        for test in tests.iter_suite_tests(suite):
 
2214
        for test in iter_suite_tests(suite):
3001
2215
            dupes.addTest(test)
3002
2216
            dupes.addTest(test) # Add it again
3003
2217
 
3009
2223
                          duplicates)
3010
2224
 
3011
2225
 
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
2226
class TestLoadTestIdList(tests.TestCaseInTempDir):
3091
2227
 
3092
2228
    def _create_test_list_file(self, file_name, content):
3130
2266
    def test_load_tests(self):
3131
2267
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
3132
2268
        loader = self._create_loader(test_list)
 
2269
 
3133
2270
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3134
2271
        self.assertEquals(test_list, _test_ids(suite))
3135
2272
 
3136
2273
    def test_exclude_tests(self):
3137
2274
        test_list = ['bogus']
3138
2275
        loader = self._create_loader(test_list)
 
2276
 
3139
2277
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
3140
2278
        self.assertEquals([], _test_ids(suite))
3141
2279
 
3186
2324
        tpr.register('bar', 'bbb.aaa.rrr')
3187
2325
        tpr.register('bar', 'bBB.aAA.rRR')
3188
2326
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
3189
 
        self.assertThat(self.get_log(),
3190
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
 
2327
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
2328
                              r'.*bar.*bbb.aaa.rrr.*bBB.aAA.rRR')
3191
2329
 
3192
2330
    def test_get_unknown_prefix(self):
3193
2331
        tpr = self._get_registry()
3213
2351
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3214
2352
 
3215
2353
 
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):
 
2354
class TestRunSuite(TestCase):
3305
2355
 
3306
2356
    def test_runner_class(self):
3307
2357
        """run_suite accepts and uses a runner_class keyword argument."""
3308
 
        class Stub(tests.TestCase):
 
2358
        class Stub(TestCase):
3309
2359
            def test_foo(self):
3310
2360
                pass
3311
2361
        suite = Stub("test_foo")
3312
2362
        calls = []
3313
 
        class MyRunner(tests.TextTestRunner):
 
2363
        class MyRunner(TextTestRunner):
3314
2364
            def run(self, test):
3315
2365
                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)
 
2366
                return ExtendedTestResult(self.stream, self.descriptions,
 
2367
                    self.verbosity)
 
2368
        run_suite(suite, runner_class=MyRunner, stream=StringIO())
 
2369
        self.assertEqual(calls, [suite])
 
2370
 
 
2371
    def test_done(self):
 
2372
        """run_suite should call result.done()"""
 
2373
        self.calls = 0
 
2374
        def one_more_call(): self.calls += 1
 
2375
        def test_function():
 
2376
            pass
 
2377
        test = unittest.FunctionTestCase(test_function)
 
2378
        class InstrumentedTestResult(ExtendedTestResult):
 
2379
            def done(self): one_more_call()
 
2380
        class MyRunner(TextTestRunner):
 
2381
            def run(self, test):
 
2382
                return InstrumentedTestResult(self.stream, self.descriptions,
 
2383
                                              self.verbosity)
 
2384
        run_suite(test, runner_class=MyRunner, stream=StringIO())
 
2385
        self.assertEquals(1, self.calls)