~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Robert Collins
  • Date: 2009-12-16 22:29:31 UTC
  • mto: This revision was merged to the branch mainline in revision 4920.
  • Revision ID: robertc@robertcollins.net-20091216222931-wbbn5ey4mwmpatwd
Review feedback.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007, 2008, 2009 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
17
17
"""Tests for the test framework."""
18
18
 
19
19
from cStringIO import StringIO
20
 
import doctest
 
20
from doctest import ELLIPSIS
21
21
import os
22
22
import signal
23
23
import sys
24
 
import threading
25
24
import time
26
25
import unittest
27
26
import warnings
28
27
 
29
 
from testtools import (
30
 
    ExtendedToOriginalDecorator,
31
 
    MultiTestResult,
32
 
    __version__ as testtools_version,
33
 
    )
34
 
from testtools.content import Content
 
28
from testtools import MultiTestResult
35
29
from testtools.content_type import ContentType
36
30
from testtools.matchers import (
37
31
    DocTestMatches,
38
32
    Equals,
39
33
    )
40
 
import testtools.testresult.doubles
 
34
import testtools.tests.helpers
41
35
 
42
36
import bzrlib
43
37
from bzrlib import (
44
38
    branchbuilder,
45
39
    bzrdir,
 
40
    debug,
46
41
    errors,
47
 
    hooks,
48
42
    lockdir,
49
43
    memorytree,
50
44
    osutils,
 
45
    progress,
51
46
    remote,
52
47
    repository,
53
48
    symbol_versioning,
54
49
    tests,
55
50
    transport,
56
51
    workingtree,
57
 
    workingtree_3,
58
 
    workingtree_4,
59
52
    )
60
53
from bzrlib.repofmt import (
61
54
    groupcompress_repo,
 
55
    pack_repo,
 
56
    weaverepo,
62
57
    )
63
58
from bzrlib.symbol_versioning import (
64
59
    deprecated_function,
66
61
    deprecated_method,
67
62
    )
68
63
from bzrlib.tests import (
69
 
    features,
 
64
    SubUnitFeature,
70
65
    test_lsprof,
71
 
    test_server,
 
66
    test_sftp_transport,
72
67
    TestUtil,
73
68
    )
74
 
from bzrlib.trace import note, mutter
75
 
from bzrlib.transport import memory
 
69
from bzrlib.trace import note
 
70
from bzrlib.transport.memory import MemoryServer, MemoryTransport
 
71
from bzrlib.version import _get_bzr_source_tree
76
72
 
77
73
 
78
74
def _test_ids(test_suite):
80
76
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
81
77
 
82
78
 
 
79
class SelftestTests(tests.TestCase):
 
80
 
 
81
    def test_import_tests(self):
 
82
        mod = TestUtil._load_module_by_name('bzrlib.tests.test_selftest')
 
83
        self.assertEqual(mod.SelftestTests, SelftestTests)
 
84
 
 
85
    def test_import_test_failure(self):
 
86
        self.assertRaises(ImportError,
 
87
                          TestUtil._load_module_by_name,
 
88
                          'bzrlib.no-name-yet')
 
89
 
 
90
 
83
91
class MetaTestLog(tests.TestCase):
84
92
 
85
93
    def test_logging(self):
91
99
            "text", "plain", {"charset": "utf8"})))
92
100
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
93
101
        self.assertThat(self.get_log(),
94
 
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
 
102
            DocTestMatches(u"...a test message\n", ELLIPSIS))
 
103
 
 
104
 
 
105
class TestUnicodeFilename(tests.TestCase):
 
106
 
 
107
    def test_probe_passes(self):
 
108
        """UnicodeFilename._probe passes."""
 
109
        # We can't test much more than that because the behaviour depends
 
110
        # on the platform.
 
111
        tests.UnicodeFilename._probe()
95
112
 
96
113
 
97
114
class TestTreeShape(tests.TestCaseInTempDir):
98
115
 
99
116
    def test_unicode_paths(self):
100
 
        self.requireFeature(features.UnicodeFilenameFeature)
 
117
        self.requireFeature(tests.UnicodeFilename)
101
118
 
102
119
        filename = u'hell\u00d8'
103
120
        self.build_tree_contents([(filename, 'contents of hello')])
104
 
        self.assertPathExists(filename)
105
 
 
106
 
 
107
 
class TestClassesAvailable(tests.TestCase):
108
 
    """As a convenience we expose Test* classes from bzrlib.tests"""
109
 
 
110
 
    def test_test_case(self):
111
 
        from bzrlib.tests import TestCase
112
 
 
113
 
    def test_test_loader(self):
114
 
        from bzrlib.tests import TestLoader
115
 
 
116
 
    def test_test_suite(self):
117
 
        from bzrlib.tests import TestSuite
 
121
        self.failUnlessExists(filename)
118
122
 
119
123
 
120
124
class TestTransportScenarios(tests.TestCase):
203
207
    def test_scenarios(self):
204
208
        # check that constructor parameters are passed through to the adapted
205
209
        # test.
206
 
        from bzrlib.tests.per_controldir import make_scenarios
 
210
        from bzrlib.tests.per_bzrdir import make_scenarios
207
211
        vfs_factory = "v"
208
212
        server1 = "a"
209
213
        server2 = "b"
307
311
        from bzrlib.tests.per_interrepository import make_scenarios
308
312
        server1 = "a"
309
313
        server2 = "b"
310
 
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
 
314
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
311
315
        scenarios = make_scenarios(server1, server2, formats)
312
316
        self.assertEqual([
313
317
            ('C0,str,str',
314
318
             {'repository_format': 'C1',
315
319
              'repository_format_to': 'C2',
316
320
              'transport_readonly_server': 'b',
317
 
              'transport_server': 'a',
318
 
              'extra_setup': 'C3'}),
 
321
              'transport_server': 'a'}),
319
322
            ('D0,str,str',
320
323
             {'repository_format': 'D1',
321
324
              'repository_format_to': 'D2',
322
325
              'transport_readonly_server': 'b',
323
 
              'transport_server': 'a',
324
 
              'extra_setup': 'D3'})],
 
326
              'transport_server': 'a'})],
325
327
            scenarios)
326
328
 
327
329
 
333
335
        from bzrlib.tests.per_workingtree import make_scenarios
334
336
        server1 = "a"
335
337
        server2 = "b"
336
 
        formats = [workingtree_4.WorkingTreeFormat4(),
337
 
                   workingtree_3.WorkingTreeFormat3(),]
 
338
        formats = [workingtree.WorkingTreeFormat2(),
 
339
                   workingtree.WorkingTreeFormat3(),]
338
340
        scenarios = make_scenarios(server1, server2, formats)
339
341
        self.assertEqual([
340
 
            ('WorkingTreeFormat4',
 
342
            ('WorkingTreeFormat2',
341
343
             {'bzrdir_format': formats[0]._matchingbzrdir,
342
344
              'transport_readonly_server': 'b',
343
345
              'transport_server': 'a',
370
372
            )
371
373
        server1 = "a"
372
374
        server2 = "b"
373
 
        formats = [workingtree_4.WorkingTreeFormat4(),
374
 
                   workingtree_3.WorkingTreeFormat3(),]
 
375
        formats = [workingtree.WorkingTreeFormat2(),
 
376
                   workingtree.WorkingTreeFormat3(),]
375
377
        scenarios = make_scenarios(server1, server2, formats)
376
378
        self.assertEqual(7, len(scenarios))
377
 
        default_wt_format = workingtree.format_registry.get_default()
378
 
        wt4_format = workingtree_4.WorkingTreeFormat4()
379
 
        wt5_format = workingtree_4.WorkingTreeFormat5()
 
379
        default_wt_format = workingtree.WorkingTreeFormat4._default_format
 
380
        wt4_format = workingtree.WorkingTreeFormat4()
 
381
        wt5_format = workingtree.WorkingTreeFormat5()
380
382
        expected_scenarios = [
381
 
            ('WorkingTreeFormat4',
 
383
            ('WorkingTreeFormat2',
382
384
             {'bzrdir_format': formats[0]._matchingbzrdir,
383
385
              'transport_readonly_server': 'b',
384
386
              'transport_server': 'a',
444
446
        # ones to add.
445
447
        from bzrlib.tests.per_tree import (
446
448
            return_parameter,
 
449
            revision_tree_from_workingtree
447
450
            )
448
451
        from bzrlib.tests.per_intertree import (
449
452
            make_scenarios,
450
453
            )
451
 
        from bzrlib.workingtree_3 import WorkingTreeFormat3
452
 
        from bzrlib.workingtree_4 import WorkingTreeFormat4
 
454
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
453
455
        input_test = TestInterTreeScenarios(
454
456
            "test_scenarios")
455
457
        server1 = "a"
456
458
        server2 = "b"
457
 
        format1 = WorkingTreeFormat4()
 
459
        format1 = WorkingTreeFormat2()
458
460
        format2 = WorkingTreeFormat3()
459
461
        formats = [("1", str, format1, format2, "converter1"),
460
462
            ("2", int, format2, format1, "converter2")]
506
508
        self.assertRaises(AssertionError, self.assertEqualStat,
507
509
            os.lstat("foo"), os.lstat("longname"))
508
510
 
509
 
    def test_failUnlessExists(self):
510
 
        """Deprecated failUnlessExists and failIfExists"""
511
 
        self.applyDeprecated(
512
 
            deprecated_in((2, 4)),
513
 
            self.failUnlessExists, '.')
514
 
        self.build_tree(['foo/', 'foo/bar'])
515
 
        self.applyDeprecated(
516
 
            deprecated_in((2, 4)),
517
 
            self.failUnlessExists, 'foo/bar')
518
 
        self.applyDeprecated(
519
 
            deprecated_in((2, 4)),
520
 
            self.failIfExists, 'foo/foo')
521
 
 
522
 
    def test_assertPathExists(self):
523
 
        self.assertPathExists('.')
524
 
        self.build_tree(['foo/', 'foo/bar'])
525
 
        self.assertPathExists('foo/bar')
526
 
        self.assertPathDoesNotExist('foo/foo')
527
 
 
528
511
 
529
512
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
530
513
 
564
547
        tree = self.make_branch_and_memory_tree('dir')
565
548
        # Guard against regression into MemoryTransport leaking
566
549
        # files to disk instead of keeping them in memory.
567
 
        self.assertFalse(osutils.lexists('dir'))
 
550
        self.failIf(osutils.lexists('dir'))
568
551
        self.assertIsInstance(tree, memorytree.MemoryTree)
569
552
 
570
553
    def test_make_branch_and_memory_tree_with_format(self):
571
554
        """make_branch_and_memory_tree should accept a format option."""
572
555
        format = bzrdir.BzrDirMetaFormat1()
573
 
        format.repository_format = repository.format_registry.get_default()
 
556
        format.repository_format = weaverepo.RepositoryFormat7()
574
557
        tree = self.make_branch_and_memory_tree('dir', format=format)
575
558
        # Guard against regression into MemoryTransport leaking
576
559
        # files to disk instead of keeping them in memory.
577
 
        self.assertFalse(osutils.lexists('dir'))
 
560
        self.failIf(osutils.lexists('dir'))
578
561
        self.assertIsInstance(tree, memorytree.MemoryTree)
579
562
        self.assertEqual(format.repository_format.__class__,
580
563
            tree.branch.repository._format.__class__)
584
567
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
585
568
        # Guard against regression into MemoryTransport leaking
586
569
        # files to disk instead of keeping them in memory.
587
 
        self.assertFalse(osutils.lexists('dir'))
 
570
        self.failIf(osutils.lexists('dir'))
588
571
 
589
572
    def test_make_branch_builder_with_format(self):
590
573
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
591
574
        # that the format objects are used.
592
575
        format = bzrdir.BzrDirMetaFormat1()
593
 
        repo_format = repository.format_registry.get_default()
 
576
        repo_format = weaverepo.RepositoryFormat7()
594
577
        format.repository_format = repo_format
595
578
        builder = self.make_branch_builder('dir', format=format)
596
579
        the_branch = builder.get_branch()
597
580
        # Guard against regression into MemoryTransport leaking
598
581
        # files to disk instead of keeping them in memory.
599
 
        self.assertFalse(osutils.lexists('dir'))
 
582
        self.failIf(osutils.lexists('dir'))
600
583
        self.assertEqual(format.repository_format.__class__,
601
584
                         the_branch.repository._format.__class__)
602
585
        self.assertEqual(repo_format.get_format_string(),
608
591
        the_branch = builder.get_branch()
609
592
        # Guard against regression into MemoryTransport leaking
610
593
        # files to disk instead of keeping them in memory.
611
 
        self.assertFalse(osutils.lexists('dir'))
 
594
        self.failIf(osutils.lexists('dir'))
612
595
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
613
596
        self.assertEqual(dir_format.repository_format.__class__,
614
597
                         the_branch.repository._format.__class__)
625
608
                l.attempt_lock()
626
609
        test = TestDanglingLock('test_function')
627
610
        result = test.run()
628
 
        total_failures = result.errors + result.failures
629
611
        if self._lock_check_thorough:
630
 
            self.assertEqual(1, len(total_failures))
 
612
            self.assertEqual(1, len(result.errors))
631
613
        else:
632
614
            # When _lock_check_thorough is disabled, then we don't trigger a
633
615
            # failure
634
 
            self.assertEqual(0, len(total_failures))
 
616
            self.assertEqual(0, len(result.errors))
635
617
 
636
618
 
637
619
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
638
620
    """Tests for the convenience functions TestCaseWithTransport introduces."""
639
621
 
640
622
    def test_get_readonly_url_none(self):
 
623
        from bzrlib.transport import get_transport
 
624
        from bzrlib.transport.memory import MemoryServer
641
625
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
642
 
        self.vfs_transport_factory = memory.MemoryServer
 
626
        self.vfs_transport_factory = MemoryServer
643
627
        self.transport_readonly_server = None
644
628
        # calling get_readonly_transport() constructs a decorator on the url
645
629
        # for the server
646
630
        url = self.get_readonly_url()
647
631
        url2 = self.get_readonly_url('foo/bar')
648
 
        t = transport.get_transport(url)
649
 
        t2 = transport.get_transport(url2)
650
 
        self.assertIsInstance(t, ReadonlyTransportDecorator)
651
 
        self.assertIsInstance(t2, ReadonlyTransportDecorator)
 
632
        t = get_transport(url)
 
633
        t2 = get_transport(url2)
 
634
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
 
635
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
652
636
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
653
637
 
654
638
    def test_get_readonly_url_http(self):
655
639
        from bzrlib.tests.http_server import HttpServer
 
640
        from bzrlib.transport import get_transport
 
641
        from bzrlib.transport.local import LocalURLServer
656
642
        from bzrlib.transport.http import HttpTransportBase
657
 
        self.transport_server = test_server.LocalURLServer
 
643
        self.transport_server = LocalURLServer
658
644
        self.transport_readonly_server = HttpServer
659
645
        # calling get_readonly_transport() gives us a HTTP server instance.
660
646
        url = self.get_readonly_url()
661
647
        url2 = self.get_readonly_url('foo/bar')
662
648
        # the transport returned may be any HttpTransportBase subclass
663
 
        t = transport.get_transport(url)
664
 
        t2 = transport.get_transport(url2)
665
 
        self.assertIsInstance(t, HttpTransportBase)
666
 
        self.assertIsInstance(t2, HttpTransportBase)
 
649
        t = get_transport(url)
 
650
        t2 = get_transport(url2)
 
651
        self.failUnless(isinstance(t, HttpTransportBase))
 
652
        self.failUnless(isinstance(t2, HttpTransportBase))
667
653
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
668
654
 
669
655
    def test_is_directory(self):
677
663
    def test_make_branch_builder(self):
678
664
        builder = self.make_branch_builder('dir')
679
665
        rev_id = builder.build_commit()
680
 
        self.assertPathExists('dir')
 
666
        self.failUnlessExists('dir')
681
667
        a_dir = bzrdir.BzrDir.open('dir')
682
668
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
683
669
        a_branch = a_dir.open_branch()
691
677
 
692
678
    def setUp(self):
693
679
        super(TestTestCaseTransports, self).setUp()
694
 
        self.vfs_transport_factory = memory.MemoryServer
 
680
        self.vfs_transport_factory = MemoryServer
695
681
 
696
682
    def test_make_bzrdir_preserves_transport(self):
697
683
        t = self.get_transport()
698
684
        result_bzrdir = self.make_bzrdir('subdir')
699
685
        self.assertIsInstance(result_bzrdir.transport,
700
 
                              memory.MemoryTransport)
 
686
                              MemoryTransport)
701
687
        # should not be on disk, should only be in memory
702
 
        self.assertPathDoesNotExist('subdir')
 
688
        self.failIfExists('subdir')
703
689
 
704
690
 
705
691
class TestChrootedTest(tests.ChrootedTestCase):
706
692
 
707
693
    def test_root_is_root(self):
708
 
        t = transport.get_transport(self.get_readonly_url())
 
694
        from bzrlib.transport import get_transport
 
695
        t = get_transport(self.get_readonly_url())
709
696
        url = t.base
710
697
        self.assertEqual(url, t.clone('..').base)
711
698
 
713
700
class TestProfileResult(tests.TestCase):
714
701
 
715
702
    def test_profiles_tests(self):
716
 
        self.requireFeature(features.lsprof_feature)
717
 
        terminal = testtools.testresult.doubles.ExtendedTestResult()
 
703
        self.requireFeature(test_lsprof.LSProfFeature)
 
704
        terminal = testtools.tests.helpers.ExtendedTestResult()
718
705
        result = tests.ProfileResult(terminal)
719
706
        class Sample(tests.TestCase):
720
707
            def a(self):
737
724
                descriptions=0,
738
725
                verbosity=1,
739
726
                )
740
 
        capture = testtools.testresult.doubles.ExtendedTestResult()
 
727
        capture = testtools.tests.helpers.ExtendedTestResult()
741
728
        test_case.run(MultiTestResult(result, capture))
742
729
        run_case = capture._events[0][1]
743
730
        timed_string = result._testTimeString(run_case)
764
751
        self.check_timing(ShortDelayTestCase('test_short_delay'),
765
752
                          r"^ +[0-9]+ms$")
766
753
 
 
754
    def _patch_get_bzr_source_tree(self):
 
755
        # Reading from the actual source tree breaks isolation, but we don't
 
756
        # want to assume that thats *all* that would happen.
 
757
        def _get_bzr_source_tree():
 
758
            return None
 
759
        orig_get_bzr_source_tree = bzrlib.version._get_bzr_source_tree
 
760
        bzrlib.version._get_bzr_source_tree = _get_bzr_source_tree
 
761
        def restore():
 
762
            bzrlib.version._get_bzr_source_tree = orig_get_bzr_source_tree
 
763
        self.addCleanup(restore)
 
764
 
 
765
    def test_assigned_benchmark_file_stores_date(self):
 
766
        self._patch_get_bzr_source_tree()
 
767
        output = StringIO()
 
768
        result = bzrlib.tests.TextTestResult(self._log_file,
 
769
                                        descriptions=0,
 
770
                                        verbosity=1,
 
771
                                        bench_history=output
 
772
                                        )
 
773
        output_string = output.getvalue()
 
774
        # if you are wondering about the regexp please read the comment in
 
775
        # test_bench_history (bzrlib.tests.test_selftest.TestRunner)
 
776
        # XXX: what comment?  -- Andrew Bennetts
 
777
        self.assertContainsRe(output_string, "--date [0-9.]+")
 
778
 
 
779
    def test_benchhistory_records_test_times(self):
 
780
        self._patch_get_bzr_source_tree()
 
781
        result_stream = StringIO()
 
782
        result = bzrlib.tests.TextTestResult(
 
783
            self._log_file,
 
784
            descriptions=0,
 
785
            verbosity=1,
 
786
            bench_history=result_stream
 
787
            )
 
788
 
 
789
        # we want profile a call and check that its test duration is recorded
 
790
        # make a new test instance that when run will generate a benchmark
 
791
        example_test_case = TestTestResult("_time_hello_world_encoding")
 
792
        # execute the test, which should succeed and record times
 
793
        example_test_case.run(result)
 
794
        lines = result_stream.getvalue().splitlines()
 
795
        self.assertEqual(2, len(lines))
 
796
        self.assertContainsRe(lines[1],
 
797
            " *[0-9]+ms bzrlib.tests.test_selftest.TestTestResult"
 
798
            "._time_hello_world_encoding")
 
799
 
767
800
    def _time_hello_world_encoding(self):
768
801
        """Profile two sleep calls
769
802
 
774
807
 
775
808
    def test_lsprofiling(self):
776
809
        """Verbose test result prints lsprof statistics from test cases."""
777
 
        self.requireFeature(features.lsprof_feature)
 
810
        self.requireFeature(test_lsprof.LSProfFeature)
778
811
        result_stream = StringIO()
779
812
        result = bzrlib.tests.VerboseTestResult(
780
 
            result_stream,
 
813
            unittest._WritelnDecorator(result_stream),
781
814
            descriptions=0,
782
815
            verbosity=2,
783
816
            )
809
842
        self.assertContainsRe(output,
810
843
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
811
844
 
812
 
    def test_uses_time_from_testtools(self):
813
 
        """Test case timings in verbose results should use testtools times"""
814
 
        import datetime
815
 
        class TimeAddedVerboseTestResult(tests.VerboseTestResult):
816
 
            def startTest(self, test):
817
 
                self.time(datetime.datetime.utcfromtimestamp(1.145))
818
 
                super(TimeAddedVerboseTestResult, self).startTest(test)
819
 
            def addSuccess(self, test):
820
 
                self.time(datetime.datetime.utcfromtimestamp(51.147))
821
 
                super(TimeAddedVerboseTestResult, self).addSuccess(test)
822
 
            def report_tests_starting(self): pass
823
 
        sio = StringIO()
824
 
        self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
825
 
        self.assertEndsWith(sio.getvalue(), "OK    50002ms\n")
826
 
 
827
845
    def test_known_failure(self):
828
 
        """Using knownFailure should trigger several result actions."""
 
846
        """A KnownFailure being raised should trigger several result actions."""
829
847
        class InstrumentedTestResult(tests.ExtendedTestResult):
830
848
            def stopTestRun(self): pass
831
 
            def report_tests_starting(self): pass
 
849
            def startTests(self): pass
 
850
            def report_test_start(self, test): pass
832
851
            def report_known_failure(self, test, err=None, details=None):
833
852
                self._call = test, 'known failure'
834
853
        result = InstrumentedTestResult(None, None, None, None)
835
854
        class Test(tests.TestCase):
836
855
            def test_function(self):
837
 
                self.knownFailure('failed!')
 
856
                raise tests.KnownFailure('failed!')
838
857
        test = Test("test_function")
839
858
        test.run(result)
840
859
        # it should invoke 'report_known_failure'.
852
871
        # verbose test output formatting
853
872
        result_stream = StringIO()
854
873
        result = bzrlib.tests.VerboseTestResult(
855
 
            result_stream,
 
874
            unittest._WritelnDecorator(result_stream),
856
875
            descriptions=0,
857
876
            verbosity=2,
858
877
            )
859
 
        _get_test("test_xfail").run(result)
860
 
        self.assertContainsRe(result_stream.getvalue(),
861
 
            "\n\\S+\\.test_xfail\\s+XFAIL\\s+\\d+ms\n"
862
 
            "\\s*(?:Text attachment: )?reason"
863
 
            "(?:\n-+\n|: {{{)"
864
 
            "this_fails"
865
 
            "(?:\n-+\n|}}}\n)")
 
878
        test = self.get_passing_test()
 
879
        result.startTest(test)
 
880
        prefix = len(result_stream.getvalue())
 
881
        # the err parameter has the shape:
 
882
        # (class, exception object, traceback)
 
883
        # KnownFailures dont get their tracebacks shown though, so we
 
884
        # can skip that.
 
885
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
 
886
        result.report_known_failure(test, err)
 
887
        output = result_stream.getvalue()[prefix:]
 
888
        lines = output.splitlines()
 
889
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
 
890
        self.assertEqual(lines[1], '    foo')
 
891
        self.assertEqual(2, len(lines))
866
892
 
867
893
    def get_passing_test(self):
868
894
        """Return a test object that can't be run usefully."""
874
900
        """Test the behaviour of invoking addNotSupported."""
875
901
        class InstrumentedTestResult(tests.ExtendedTestResult):
876
902
            def stopTestRun(self): pass
877
 
            def report_tests_starting(self): pass
 
903
            def startTests(self): pass
 
904
            def report_test_start(self, test): pass
878
905
            def report_unsupported(self, test, feature):
879
906
                self._call = test, feature
880
907
        result = InstrumentedTestResult(None, None, None, None)
881
908
        test = SampleTestCase('_test_pass')
882
 
        feature = features.Feature()
 
909
        feature = tests.Feature()
883
910
        result.startTest(test)
884
911
        result.addNotSupported(test, feature)
885
912
        # it should invoke 'report_unsupported'.
899
926
        # verbose test output formatting
900
927
        result_stream = StringIO()
901
928
        result = bzrlib.tests.VerboseTestResult(
902
 
            result_stream,
 
929
            unittest._WritelnDecorator(result_stream),
903
930
            descriptions=0,
904
931
            verbosity=2,
905
932
            )
906
933
        test = self.get_passing_test()
907
 
        feature = features.Feature()
 
934
        feature = tests.Feature()
908
935
        result.startTest(test)
909
936
        prefix = len(result_stream.getvalue())
910
937
        result.report_unsupported(test, feature)
919
946
        """An UnavailableFeature being raised should invoke addNotSupported."""
920
947
        class InstrumentedTestResult(tests.ExtendedTestResult):
921
948
            def stopTestRun(self): pass
922
 
            def report_tests_starting(self): pass
 
949
            def startTests(self): pass
 
950
            def report_test_start(self, test): pass
923
951
            def addNotSupported(self, test, feature):
924
952
                self._call = test, feature
925
953
        result = InstrumentedTestResult(None, None, None, None)
926
 
        feature = features.Feature()
 
954
        feature = tests.Feature()
927
955
        class Test(tests.TestCase):
928
956
            def test_function(self):
929
957
                raise tests.UnavailableFeature(feature)
948
976
    def test_strict_with_known_failure(self):
949
977
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
950
978
                                             verbosity=1)
951
 
        test = _get_test("test_xfail")
952
 
        test.run(result)
 
979
        test = self.get_passing_test()
 
980
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
 
981
        result.addExpectedFailure(test, err)
953
982
        self.assertFalse(result.wasStrictlySuccessful())
954
983
        self.assertEqual(None, result._extractBenchmarkTime(test))
955
984
 
966
995
        class InstrumentedTestResult(tests.ExtendedTestResult):
967
996
            calls = 0
968
997
            def startTests(self): self.calls += 1
 
998
            def report_test_start(self, test): pass
969
999
        result = InstrumentedTestResult(None, None, None, None)
970
1000
        def test_function():
971
1001
            pass
973
1003
        test.run(result)
974
1004
        self.assertEquals(1, result.calls)
975
1005
 
976
 
    def test_startTests_only_once(self):
977
 
        """With multiple tests startTests should still only be called once"""
978
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
979
 
            calls = 0
980
 
            def startTests(self): self.calls += 1
981
 
        result = InstrumentedTestResult(None, None, None, None)
982
 
        suite = unittest.TestSuite([
983
 
            unittest.FunctionTestCase(lambda: None),
984
 
            unittest.FunctionTestCase(lambda: None)])
985
 
        suite.run(result)
986
 
        self.assertEquals(1, result.calls)
987
 
        self.assertEquals(2, result.count)
 
1006
 
 
1007
class TestUnicodeFilenameFeature(tests.TestCase):
 
1008
 
 
1009
    def test_probe_passes(self):
 
1010
        """UnicodeFilenameFeature._probe passes."""
 
1011
        # We can't test much more than that because the behaviour depends
 
1012
        # on the platform.
 
1013
        tests.UnicodeFilenameFeature._probe()
988
1014
 
989
1015
 
990
1016
class TestRunner(tests.TestCase):
1003
1029
        because of our use of global state.
1004
1030
        """
1005
1031
        old_root = tests.TestCaseInTempDir.TEST_ROOT
 
1032
        old_leak = tests.TestCase._first_thread_leaker_id
1006
1033
        try:
1007
1034
            tests.TestCaseInTempDir.TEST_ROOT = None
 
1035
            tests.TestCase._first_thread_leaker_id = None
1008
1036
            return testrunner.run(test)
1009
1037
        finally:
1010
1038
            tests.TestCaseInTempDir.TEST_ROOT = old_root
 
1039
            tests.TestCase._first_thread_leaker_id = old_leak
1011
1040
 
1012
1041
    def test_known_failure_failed_run(self):
1013
1042
        # run a test that generates a known failure which should be printed in
1018
1047
        test = unittest.TestSuite()
1019
1048
        test.addTest(Test("known_failure_test"))
1020
1049
        def failing_test():
1021
 
            raise AssertionError('foo')
 
1050
            self.fail('foo')
1022
1051
        test.addTest(unittest.FunctionTestCase(failing_test))
1023
1052
        stream = StringIO()
1024
1053
        runner = tests.TextTestRunner(stream=stream)
1028
1057
            '(?sm)^bzr selftest.*$'
1029
1058
            '.*'
1030
1059
            '^======================================================================\n'
1031
 
            '^FAIL: failing_test\n'
 
1060
            '^FAIL: unittest.FunctionTestCase \\(failing_test\\)\n'
1032
1061
            '^----------------------------------------------------------------------\n'
1033
1062
            'Traceback \\(most recent call last\\):\n'
1034
1063
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1035
 
            '    raise AssertionError\\(\'foo\'\\)\n'
 
1064
            '    self.fail\\(\'foo\'\\)\n'
1036
1065
            '.*'
1037
1066
            '^----------------------------------------------------------------------\n'
1038
1067
            '.*'
1044
1073
        # the final output.
1045
1074
        class Test(tests.TestCase):
1046
1075
            def known_failure_test(self):
1047
 
                self.knownFailure("Never works...")
 
1076
                self.expectFailure('failed', self.assertTrue, False)
1048
1077
        test = Test("known_failure_test")
1049
1078
        stream = StringIO()
1050
1079
        runner = tests.TextTestRunner(stream=stream)
1056
1085
            '\n'
1057
1086
            'OK \\(known_failures=1\\)\n')
1058
1087
 
1059
 
    def test_unexpected_success_bad(self):
1060
 
        class Test(tests.TestCase):
1061
 
            def test_truth(self):
1062
 
                self.expectFailure("No absolute truth", self.assertTrue, True)
1063
 
        runner = tests.TextTestRunner(stream=StringIO())
1064
 
        result = self.run_test_runner(runner, Test("test_truth"))
1065
 
        if testtools_version[:3] <= (0, 9, 11):
1066
 
            self.assertContainsRe(runner.stream.getvalue(),
1067
 
                "=+\n"
1068
 
                "FAIL: \\S+\.test_truth\n"
1069
 
                "-+\n"
1070
 
                "(?:.*\n)*"
1071
 
                "No absolute truth\n"
1072
 
                "(?:.*\n)*"
1073
 
                "-+\n"
1074
 
                "Ran 1 test in .*\n"
1075
 
                "\n"
1076
 
                "FAILED \\(failures=1\\)\n\\Z")
1077
 
        else:
1078
 
            self.assertContainsRe(runner.stream.getvalue(),
1079
 
                "=+\n"
1080
 
                "FAIL: \\S+\.test_truth\n"
1081
 
                "-+\n"
1082
 
                "Empty attachments:\n"
1083
 
                "  log\n"
1084
 
                "\n"
1085
 
                "reason: {{{No absolute truth}}}\n"
1086
 
                "-+\n"
1087
 
                "Ran 1 test in .*\n"
1088
 
                "\n"
1089
 
                "FAILED \\(failures=1\\)\n\\Z")
1090
 
 
1091
1088
    def test_result_decorator(self):
1092
1089
        # decorate results
1093
1090
        calls = []
1094
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
 
1091
        class LoggingDecorator(tests.ForwardingResult):
1095
1092
            def startTest(self, test):
1096
 
                ExtendedToOriginalDecorator.startTest(self, test)
 
1093
                tests.ForwardingResult.startTest(self, test)
1097
1094
                calls.append('start')
1098
1095
        test = unittest.FunctionTestCase(lambda:None)
1099
1096
        stream = StringIO()
1177
1174
 
1178
1175
    def test_unsupported_features_listed(self):
1179
1176
        """When unsupported features are encountered they are detailed."""
1180
 
        class Feature1(features.Feature):
 
1177
        class Feature1(tests.Feature):
1181
1178
            def _probe(self): return False
1182
 
        class Feature2(features.Feature):
 
1179
        class Feature2(tests.Feature):
1183
1180
            def _probe(self): return False
1184
1181
        # create sample tests
1185
1182
        test1 = SampleTestCase('_test_pass')
1200
1197
            ],
1201
1198
            lines[-3:])
1202
1199
 
1203
 
    def test_verbose_test_count(self):
1204
 
        """A verbose test run reports the right test count at the start"""
1205
 
        suite = TestUtil.TestSuite([
1206
 
            unittest.FunctionTestCase(lambda:None),
1207
 
            unittest.FunctionTestCase(lambda:None)])
1208
 
        self.assertEqual(suite.countTestCases(), 2)
1209
 
        stream = StringIO()
1210
 
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
1211
 
        # Need to use the CountingDecorator as that's what sets num_tests
1212
 
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
1213
 
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
 
1200
    def _patch_get_bzr_source_tree(self):
 
1201
        # Reading from the actual source tree breaks isolation, but we don't
 
1202
        # want to assume that thats *all* that would happen.
 
1203
        self._get_source_tree_calls = []
 
1204
        def _get_bzr_source_tree():
 
1205
            self._get_source_tree_calls.append("called")
 
1206
            return None
 
1207
        orig_get_bzr_source_tree = bzrlib.version._get_bzr_source_tree
 
1208
        bzrlib.version._get_bzr_source_tree = _get_bzr_source_tree
 
1209
        def restore():
 
1210
            bzrlib.version._get_bzr_source_tree = orig_get_bzr_source_tree
 
1211
        self.addCleanup(restore)
 
1212
 
 
1213
    def test_bench_history(self):
 
1214
        # tests that the running the benchmark passes bench_history into
 
1215
        # the test result object. We can tell that happens if
 
1216
        # _get_bzr_source_tree is called.
 
1217
        self._patch_get_bzr_source_tree()
 
1218
        test = TestRunner('dummy_test')
 
1219
        output = StringIO()
 
1220
        runner = tests.TextTestRunner(stream=self._log_file,
 
1221
                                      bench_history=output)
 
1222
        result = self.run_test_runner(runner, test)
 
1223
        output_string = output.getvalue()
 
1224
        self.assertContainsRe(output_string, "--date [0-9.]+")
 
1225
        self.assertLength(1, self._get_source_tree_calls)
1214
1226
 
1215
1227
    def test_startTestRun(self):
1216
1228
        """run should call result.startTestRun()"""
1217
1229
        calls = []
1218
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
 
1230
        class LoggingDecorator(tests.ForwardingResult):
1219
1231
            def startTestRun(self):
1220
 
                ExtendedToOriginalDecorator.startTestRun(self)
 
1232
                tests.ForwardingResult.startTestRun(self)
1221
1233
                calls.append('startTestRun')
1222
1234
        test = unittest.FunctionTestCase(lambda:None)
1223
1235
        stream = StringIO()
1229
1241
    def test_stopTestRun(self):
1230
1242
        """run should call result.stopTestRun()"""
1231
1243
        calls = []
1232
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
 
1244
        class LoggingDecorator(tests.ForwardingResult):
1233
1245
            def stopTestRun(self):
1234
 
                ExtendedToOriginalDecorator.stopTestRun(self)
 
1246
                tests.ForwardingResult.stopTestRun(self)
1235
1247
                calls.append('stopTestRun')
1236
1248
        test = unittest.FunctionTestCase(lambda:None)
1237
1249
        stream = StringIO()
1240
1252
        result = self.run_test_runner(runner, test)
1241
1253
        self.assertLength(1, calls)
1242
1254
 
1243
 
    def test_unicode_test_output_on_ascii_stream(self):
1244
 
        """Showing results should always succeed even on an ascii console"""
1245
 
        class FailureWithUnicode(tests.TestCase):
1246
 
            def test_log_unicode(self):
1247
 
                self.log(u"\u2606")
1248
 
                self.fail("Now print that log!")
1249
 
        out = StringIO()
1250
 
        self.overrideAttr(osutils, "get_terminal_encoding",
1251
 
            lambda trace=False: "ascii")
1252
 
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
1253
 
            FailureWithUnicode("test_log_unicode"))
1254
 
        if testtools_version[:3] > (0, 9, 11):
1255
 
            self.assertContainsRe(out.getvalue(), "log: {{{\d+\.\d+  \\\\u2606}}}")
1256
 
        else:
1257
 
            self.assertContainsRe(out.getvalue(),
1258
 
                "Text attachment: log\n"
1259
 
                "-+\n"
1260
 
                "\d+\.\d+  \\\\u2606\n"
1261
 
                "-+\n")
1262
 
 
1263
1255
 
1264
1256
class SampleTestCase(tests.TestCase):
1265
1257
 
1330
1322
        self.assertEqual(flags, bzrlib.debug.debug_flags)
1331
1323
 
1332
1324
    def change_selftest_debug_flags(self, new_flags):
1333
 
        self.overrideAttr(tests, 'selftest_debug_flags', set(new_flags))
 
1325
        orig_selftest_flags = tests.selftest_debug_flags
 
1326
        self.addCleanup(self._restore_selftest_debug_flags, orig_selftest_flags)
 
1327
        tests.selftest_debug_flags = set(new_flags)
 
1328
 
 
1329
    def _restore_selftest_debug_flags(self, flags):
 
1330
        tests.selftest_debug_flags = flags
1334
1331
 
1335
1332
    def test_allow_debug_flag(self):
1336
1333
        """The -Eallow_debug flag prevents bzrlib.debug.debug_flags from being
1440
1437
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1441
1438
        output_stream = StringIO()
1442
1439
        result = bzrlib.tests.VerboseTestResult(
1443
 
            output_stream,
 
1440
            unittest._WritelnDecorator(output_stream),
1444
1441
            descriptions=0,
1445
1442
            verbosity=2)
1446
1443
        sample_test.run(result)
1453
1450
        # Note this test won't fail with hooks that the core library doesn't
1454
1451
        # use - but it trigger with a plugin that adds hooks, so its still a
1455
1452
        # useful warning in that case.
1456
 
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
1457
 
        self.assertEqual(
1458
 
            bzrlib.smart.server.SmartServerHooks(),
 
1453
        self.assertEqual(bzrlib.branch.BranchHooks(),
 
1454
            bzrlib.branch.Branch.hooks)
 
1455
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
1459
1456
            bzrlib.smart.server.SmartTCPServer.hooks)
1460
 
        self.assertEqual(
1461
 
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
 
1457
        self.assertEqual(bzrlib.commands.CommandHooks(),
 
1458
            bzrlib.commands.Command.hooks)
1462
1459
 
1463
1460
    def test__gather_lsprof_in_benchmarks(self):
1464
1461
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1465
1462
 
1466
1463
        Each self.time() call is individually and separately profiled.
1467
1464
        """
1468
 
        self.requireFeature(features.lsprof_feature)
 
1465
        self.requireFeature(test_lsprof.LSProfFeature)
1469
1466
        # overrides the class member with an instance member so no cleanup
1470
1467
        # needed.
1471
1468
        self._gather_lsprof_in_benchmarks = True
1487
1484
        # permitted.
1488
1485
        # Manually set one up (TestCase doesn't and shouldn't provide magic
1489
1486
        # machinery)
1490
 
        transport_server = memory.MemoryServer()
1491
 
        transport_server.start_server()
1492
 
        self.addCleanup(transport_server.stop_server)
 
1487
        transport_server = MemoryServer()
 
1488
        transport_server.setUp()
 
1489
        self.addCleanup(transport_server.tearDown)
1493
1490
        t = transport.get_transport(transport_server.get_url())
1494
1491
        bzrdir.BzrDir.create(t.base)
1495
1492
        self.assertRaises(errors.BzrError,
1501
1498
 
1502
1499
    def test_requireFeature_available(self):
1503
1500
        """self.requireFeature(available) is a no-op."""
1504
 
        class Available(features.Feature):
 
1501
        class Available(tests.Feature):
1505
1502
            def _probe(self):return True
1506
1503
        feature = Available()
1507
1504
        self.requireFeature(feature)
1508
1505
 
1509
1506
    def test_requireFeature_unavailable(self):
1510
1507
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1511
 
        class Unavailable(features.Feature):
 
1508
        class Unavailable(tests.Feature):
1512
1509
            def _probe(self):return False
1513
1510
        feature = Unavailable()
1514
1511
        self.assertRaises(tests.UnavailableFeature,
1575
1572
            result.calls)
1576
1573
 
1577
1574
    def test_start_server_registers_url(self):
1578
 
        transport_server = memory.MemoryServer()
 
1575
        transport_server = MemoryServer()
1579
1576
        # A little strict, but unlikely to be changed soon.
1580
1577
        self.assertEqual([], self._bzr_selftest_roots)
1581
1578
        self.start_server(transport_server)
1637
1634
        self.assertRaises(AssertionError,
1638
1635
            self.assertListRaises, _TestException, success_generator)
1639
1636
 
1640
 
    def test_overrideAttr_without_value(self):
1641
 
        self.test_attr = 'original' # Define a test attribute
1642
 
        obj = self # Make 'obj' visible to the embedded test
1643
 
        class Test(tests.TestCase):
1644
 
 
1645
 
            def setUp(self):
1646
 
                tests.TestCase.setUp(self)
1647
 
                self.orig = self.overrideAttr(obj, 'test_attr')
1648
 
 
1649
 
            def test_value(self):
1650
 
                self.assertEqual('original', self.orig)
1651
 
                self.assertEqual('original', obj.test_attr)
1652
 
                obj.test_attr = 'modified'
1653
 
                self.assertEqual('modified', obj.test_attr)
1654
 
 
1655
 
        test = Test('test_value')
1656
 
        test.run(unittest.TestResult())
1657
 
        self.assertEqual('original', obj.test_attr)
1658
 
 
1659
 
    def test_overrideAttr_with_value(self):
1660
 
        self.test_attr = 'original' # Define a test attribute
1661
 
        obj = self # Make 'obj' visible to the embedded test
1662
 
        class Test(tests.TestCase):
1663
 
 
1664
 
            def setUp(self):
1665
 
                tests.TestCase.setUp(self)
1666
 
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1667
 
 
1668
 
            def test_value(self):
1669
 
                self.assertEqual('original', self.orig)
1670
 
                self.assertEqual('modified', obj.test_attr)
1671
 
 
1672
 
        test = Test('test_value')
1673
 
        test.run(unittest.TestResult())
1674
 
        self.assertEqual('original', obj.test_attr)
1675
 
 
1676
 
    def test_recordCalls(self):
1677
 
        from bzrlib.tests import test_selftest
1678
 
        calls = self.recordCalls(
1679
 
            test_selftest, '_add_numbers')
1680
 
        self.assertEqual(test_selftest._add_numbers(2, 10),
1681
 
            12)
1682
 
        self.assertEquals(calls, [((2, 10), {})])
1683
 
 
1684
 
 
1685
 
def _add_numbers(a, b):
1686
 
    return a + b
1687
 
 
1688
 
 
1689
 
class _MissingFeature(features.Feature):
1690
 
    def _probe(self):
1691
 
        return False
1692
 
missing_feature = _MissingFeature()
1693
 
 
1694
 
 
1695
 
def _get_test(name):
1696
 
    """Get an instance of a specific example test.
1697
 
 
1698
 
    We protect this in a function so that they don't auto-run in the test
1699
 
    suite.
1700
 
    """
1701
 
 
1702
 
    class ExampleTests(tests.TestCase):
1703
 
 
1704
 
        def test_fail(self):
1705
 
            mutter('this was a failing test')
1706
 
            self.fail('this test will fail')
1707
 
 
1708
 
        def test_error(self):
1709
 
            mutter('this test errored')
1710
 
            raise RuntimeError('gotcha')
1711
 
 
1712
 
        def test_missing_feature(self):
1713
 
            mutter('missing the feature')
1714
 
            self.requireFeature(missing_feature)
1715
 
 
1716
 
        def test_skip(self):
1717
 
            mutter('this test will be skipped')
1718
 
            raise tests.TestSkipped('reason')
1719
 
 
1720
 
        def test_success(self):
1721
 
            mutter('this test succeeds')
1722
 
 
1723
 
        def test_xfail(self):
1724
 
            mutter('test with expected failure')
1725
 
            self.knownFailure('this_fails')
1726
 
 
1727
 
        def test_unexpected_success(self):
1728
 
            mutter('test with unexpected success')
1729
 
            self.expectFailure('should_fail', lambda: None)
1730
 
 
1731
 
    return ExampleTests(name)
1732
 
 
1733
 
 
1734
 
class TestTestCaseLogDetails(tests.TestCase):
1735
 
 
1736
 
    def _run_test(self, test_name):
1737
 
        test = _get_test(test_name)
1738
 
        result = testtools.TestResult()
1739
 
        test.run(result)
1740
 
        return result
1741
 
 
1742
 
    def test_fail_has_log(self):
1743
 
        result = self._run_test('test_fail')
1744
 
        self.assertEqual(1, len(result.failures))
1745
 
        result_content = result.failures[0][1]
1746
 
        if testtools_version < (0, 9, 12):
1747
 
            self.assertContainsRe(result_content, 'Text attachment: log')
1748
 
        self.assertContainsRe(result_content, 'this was a failing test')
1749
 
 
1750
 
    def test_error_has_log(self):
1751
 
        result = self._run_test('test_error')
1752
 
        self.assertEqual(1, len(result.errors))
1753
 
        result_content = result.errors[0][1]
1754
 
        if testtools_version < (0, 9, 12):
1755
 
            self.assertContainsRe(result_content, 'Text attachment: log')
1756
 
        self.assertContainsRe(result_content, 'this test errored')
1757
 
 
1758
 
    def test_skip_has_no_log(self):
1759
 
        result = self._run_test('test_skip')
1760
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
1761
 
        skips = result.skip_reasons['reason']
1762
 
        self.assertEqual(1, len(skips))
1763
 
        test = skips[0]
1764
 
        self.assertFalse('log' in test.getDetails())
1765
 
 
1766
 
    def test_missing_feature_has_no_log(self):
1767
 
        # testtools doesn't know about addNotSupported, so it just gets
1768
 
        # considered as a skip
1769
 
        result = self._run_test('test_missing_feature')
1770
 
        self.assertEqual([missing_feature], result.skip_reasons.keys())
1771
 
        skips = result.skip_reasons[missing_feature]
1772
 
        self.assertEqual(1, len(skips))
1773
 
        test = skips[0]
1774
 
        self.assertFalse('log' in test.getDetails())
1775
 
 
1776
 
    def test_xfail_has_no_log(self):
1777
 
        result = self._run_test('test_xfail')
1778
 
        self.assertEqual(1, len(result.expectedFailures))
1779
 
        result_content = result.expectedFailures[0][1]
1780
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
1781
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
1782
 
 
1783
 
    def test_unexpected_success_has_log(self):
1784
 
        result = self._run_test('test_unexpected_success')
1785
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
1786
 
        # Inconsistency, unexpectedSuccesses is a list of tests,
1787
 
        # expectedFailures is a list of reasons?
1788
 
        test = result.unexpectedSuccesses[0]
1789
 
        details = test.getDetails()
1790
 
        self.assertTrue('log' in details)
1791
 
 
1792
 
 
1793
 
class TestTestCloning(tests.TestCase):
1794
 
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
1795
 
 
1796
 
    def test_cloned_testcase_does_not_share_details(self):
1797
 
        """A TestCase cloned with clone_test does not share mutable attributes
1798
 
        such as details or cleanups.
1799
 
        """
1800
 
        class Test(tests.TestCase):
1801
 
            def test_foo(self):
1802
 
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1803
 
        orig_test = Test('test_foo')
1804
 
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1805
 
        orig_test.run(unittest.TestResult())
1806
 
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1807
 
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
1808
 
 
1809
 
    def test_double_apply_scenario_preserves_first_scenario(self):
1810
 
        """Applying two levels of scenarios to a test preserves the attributes
1811
 
        added by both scenarios.
1812
 
        """
1813
 
        class Test(tests.TestCase):
1814
 
            def test_foo(self):
1815
 
                pass
1816
 
        test = Test('test_foo')
1817
 
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1818
 
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1819
 
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1820
 
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1821
 
        all_tests = list(tests.iter_suite_tests(suite))
1822
 
        self.assertLength(4, all_tests)
1823
 
        all_xys = sorted((t.x, t.y) for t in all_tests)
1824
 
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1825
 
 
1826
1637
 
1827
1638
# NB: Don't delete this; it's not actually from 0.11!
1828
1639
@deprecated_function(deprecated_in((0, 11, 0)))
1956
1767
    def test_make_branch_and_tree_with_format(self):
1957
1768
        # we should be able to supply a format to make_branch_and_tree
1958
1769
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
 
1770
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
1959
1771
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
1960
1772
                              bzrlib.bzrdir.BzrDirMetaFormat1)
 
1773
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
 
1774
                              bzrlib.bzrdir.BzrDirFormat6)
1961
1775
 
1962
1776
    def test_make_branch_and_memory_tree(self):
1963
1777
        # we should be able to get a new branch and a mutable tree from
1969
1783
        # make_branch_and_tree has to use local branch and repositories
1970
1784
        # when the vfs transport and local disk are colocated, even if
1971
1785
        # a different transport is in use for url generation.
1972
 
        self.transport_server = test_server.FakeVFATServer
 
1786
        from bzrlib.transport.fakevfat import FakeVFATServer
 
1787
        self.transport_server = FakeVFATServer
1973
1788
        self.assertFalse(self.get_url('t1').startswith('file://'))
1974
1789
        tree = self.make_branch_and_tree('t1')
1975
1790
        base = tree.bzrdir.root_transport.base
1980
1795
                tree.branch.repository.bzrdir.root_transport)
1981
1796
 
1982
1797
 
1983
 
class SelfTestHelper(object):
 
1798
class SelfTestHelper:
1984
1799
 
1985
1800
    def run_selftest(self, **kwargs):
1986
1801
        """Run selftest returning its output."""
2040
1855
        self.assertLength(2, output.readlines())
2041
1856
 
2042
1857
    def test_lsprof_tests(self):
2043
 
        self.requireFeature(features.lsprof_feature)
2044
 
        results = []
 
1858
        self.requireFeature(test_lsprof.LSProfFeature)
 
1859
        calls = []
2045
1860
        class Test(object):
2046
1861
            def __call__(test, result):
2047
1862
                test.run(result)
2048
1863
            def run(test, result):
2049
 
                results.append(result)
 
1864
                self.assertIsInstance(result, tests.ForwardingResult)
 
1865
                calls.append("called")
2050
1866
            def countTestCases(self):
2051
1867
                return 1
2052
1868
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
2053
 
        self.assertLength(1, results)
2054
 
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
 
1869
        self.assertLength(1, calls)
2055
1870
 
2056
1871
    def test_random(self):
2057
1872
        # test randomising by listing a number of tests.
2073
1888
        self.assertEqual(expected.getvalue(), repeated.getvalue())
2074
1889
 
2075
1890
    def test_runner_class(self):
2076
 
        self.requireFeature(features.subunit)
 
1891
        self.requireFeature(SubUnitFeature)
2077
1892
        from subunit import ProtocolTestCase
2078
1893
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2079
1894
            test_suite_factory=self.factory)
2111
1926
        self.assertEqual(transport_server, captured_transport[0])
2112
1927
 
2113
1928
    def test_transport_sftp(self):
2114
 
        self.requireFeature(features.paramiko)
2115
 
        from bzrlib.tests import stub_sftp
2116
 
        self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
 
1929
        try:
 
1930
            import bzrlib.transport.sftp
 
1931
        except errors.ParamikoNotPresent:
 
1932
            raise tests.TestSkipped("Paramiko not present")
 
1933
        self.check_transport_set(bzrlib.transport.sftp.SFTPAbsoluteServer)
2117
1934
 
2118
1935
    def test_transport_memory(self):
2119
 
        self.check_transport_set(memory.MemoryServer)
 
1936
        self.check_transport_set(bzrlib.transport.memory.MemoryServer)
2120
1937
 
2121
1938
 
2122
1939
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
2137
1954
            load_list='missing file name', list_only=True)
2138
1955
 
2139
1956
 
2140
 
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2141
 
 
2142
 
    _test_needs_features = [features.subunit]
2143
 
 
2144
 
    def run_subunit_stream(self, test_name):
2145
 
        from subunit import ProtocolTestCase
2146
 
        def factory():
2147
 
            return TestUtil.TestSuite([_get_test(test_name)])
2148
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2149
 
            test_suite_factory=factory)
2150
 
        test = ProtocolTestCase(stream)
2151
 
        result = testtools.TestResult()
2152
 
        test.run(result)
2153
 
        content = stream.getvalue()
2154
 
        return content, result
2155
 
 
2156
 
    def test_fail_has_log(self):
2157
 
        content, result = self.run_subunit_stream('test_fail')
2158
 
        self.assertEqual(1, len(result.failures))
2159
 
        self.assertContainsRe(content, '(?m)^log$')
2160
 
        self.assertContainsRe(content, 'this test will fail')
2161
 
 
2162
 
    def test_error_has_log(self):
2163
 
        content, result = self.run_subunit_stream('test_error')
2164
 
        self.assertContainsRe(content, '(?m)^log$')
2165
 
        self.assertContainsRe(content, 'this test errored')
2166
 
 
2167
 
    def test_skip_has_no_log(self):
2168
 
        content, result = self.run_subunit_stream('test_skip')
2169
 
        self.assertNotContainsRe(content, '(?m)^log$')
2170
 
        self.assertNotContainsRe(content, 'this test will be skipped')
2171
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
2172
 
        skips = result.skip_reasons['reason']
2173
 
        self.assertEqual(1, len(skips))
2174
 
        test = skips[0]
2175
 
        # RemotedTestCase doesn't preserve the "details"
2176
 
        ## self.assertFalse('log' in test.getDetails())
2177
 
 
2178
 
    def test_missing_feature_has_no_log(self):
2179
 
        content, result = self.run_subunit_stream('test_missing_feature')
2180
 
        self.assertNotContainsRe(content, '(?m)^log$')
2181
 
        self.assertNotContainsRe(content, 'missing the feature')
2182
 
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
2183
 
        skips = result.skip_reasons['_MissingFeature\n']
2184
 
        self.assertEqual(1, len(skips))
2185
 
        test = skips[0]
2186
 
        # RemotedTestCase doesn't preserve the "details"
2187
 
        ## self.assertFalse('log' in test.getDetails())
2188
 
 
2189
 
    def test_xfail_has_no_log(self):
2190
 
        content, result = self.run_subunit_stream('test_xfail')
2191
 
        self.assertNotContainsRe(content, '(?m)^log$')
2192
 
        self.assertNotContainsRe(content, 'test with expected failure')
2193
 
        self.assertEqual(1, len(result.expectedFailures))
2194
 
        result_content = result.expectedFailures[0][1]
2195
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
2196
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
2197
 
 
2198
 
    def test_unexpected_success_has_log(self):
2199
 
        content, result = self.run_subunit_stream('test_unexpected_success')
2200
 
        self.assertContainsRe(content, '(?m)^log$')
2201
 
        self.assertContainsRe(content, 'test with unexpected success')
2202
 
        # GZ 2011-05-18: Old versions of subunit treat unexpected success as a
2203
 
        #                success, if a min version check is added remove this
2204
 
        from subunit import TestProtocolClient as _Client
2205
 
        if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
2206
 
            self.expectFailure('subunit treats "unexpectedSuccess"'
2207
 
                               ' as a plain success',
2208
 
                self.assertEqual, 1, len(result.unexpectedSuccesses))
2209
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
2210
 
        test = result.unexpectedSuccesses[0]
2211
 
        # RemotedTestCase doesn't preserve the "details"
2212
 
        ## self.assertTrue('log' in test.getDetails())
2213
 
 
2214
 
    def test_success_has_no_log(self):
2215
 
        content, result = self.run_subunit_stream('test_success')
2216
 
        self.assertEqual(1, result.testsRun)
2217
 
        self.assertNotContainsRe(content, '(?m)^log$')
2218
 
        self.assertNotContainsRe(content, 'this test succeeds')
2219
 
 
2220
 
 
2221
1957
class TestRunBzr(tests.TestCase):
2222
1958
 
2223
1959
    out = ''
2346
2082
        # stdout and stderr of the invoked run_bzr
2347
2083
        current_factory = bzrlib.ui.ui_factory
2348
2084
        self.run_bzr(['foo'])
2349
 
        self.assertFalse(current_factory is self.factory)
 
2085
        self.failIf(current_factory is self.factory)
2350
2086
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2351
2087
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2352
2088
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2509
2245
 
2510
2246
 
2511
2247
class TestStartBzrSubProcess(tests.TestCase):
2512
 
    """Stub test start_bzr_subprocess."""
2513
2248
 
2514
 
    def _subprocess_log_cleanup(self):
2515
 
        """Inhibits the base version as we don't produce a log file."""
 
2249
    def check_popen_state(self):
 
2250
        """Replace to make assertions when popen is called."""
2516
2251
 
2517
2252
    def _popen(self, *args, **kwargs):
2518
 
        """Override the base version to record the command that is run.
2519
 
 
2520
 
        From there we can ensure it is correct without spawning a real process.
2521
 
        """
 
2253
        """Record the command that is run, so that we can ensure it is correct"""
2522
2254
        self.check_popen_state()
2523
2255
        self._popen_args = args
2524
2256
        self._popen_kwargs = kwargs
2525
2257
        raise _DontSpawnProcess()
2526
2258
 
2527
 
    def check_popen_state(self):
2528
 
        """Replace to make assertions when popen is called."""
2529
 
 
2530
2259
    def test_run_bzr_subprocess_no_plugins(self):
2531
2260
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2532
2261
        command = self._popen_args[0]
2536
2265
 
2537
2266
    def test_allow_plugins(self):
2538
2267
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2539
 
                          allow_plugins=True)
 
2268
            allow_plugins=True)
2540
2269
        command = self._popen_args[0]
2541
2270
        self.assertEqual([], command[2:])
2542
2271
 
2543
2272
    def test_set_env(self):
2544
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2273
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2545
2274
        # set in the child
2546
2275
        def check_environment():
2547
2276
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2548
2277
        self.check_popen_state = check_environment
2549
2278
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2550
 
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
 
2279
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
2551
2280
        # not set in theparent
2552
2281
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2553
2282
 
2554
2283
    def test_run_bzr_subprocess_env_del(self):
2555
2284
        """run_bzr_subprocess can remove environment variables too."""
2556
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2285
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2557
2286
        def check_environment():
2558
2287
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2559
2288
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2560
2289
        self.check_popen_state = check_environment
2561
2290
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2562
 
                          env_changes={'EXISTANT_ENV_VAR':None})
 
2291
            env_changes={'EXISTANT_ENV_VAR':None})
2563
2292
        # Still set in parent
2564
2293
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2565
2294
        del os.environ['EXISTANT_ENV_VAR']
2566
2295
 
2567
2296
    def test_env_del_missing(self):
2568
 
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
 
2297
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2569
2298
        def check_environment():
2570
2299
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2571
2300
        self.check_popen_state = check_environment
2572
2301
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2573
 
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
 
2302
            env_changes={'NON_EXISTANT_ENV_VAR':None})
2574
2303
 
2575
2304
    def test_working_dir(self):
2576
2305
        """Test that we can specify the working dir for the child"""
2579
2308
        chdirs = []
2580
2309
        def chdir(path):
2581
2310
            chdirs.append(path)
2582
 
        self.overrideAttr(os, 'chdir', chdir)
2583
 
        def getcwd():
2584
 
            return 'current'
2585
 
        self.overrideAttr(osutils, 'getcwd', getcwd)
2586
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2587
 
                          working_dir='foo')
 
2311
        os.chdir = chdir
 
2312
        try:
 
2313
            def getcwd():
 
2314
                return 'current'
 
2315
            osutils.getcwd = getcwd
 
2316
            try:
 
2317
                self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2318
                    working_dir='foo')
 
2319
            finally:
 
2320
                osutils.getcwd = orig_getcwd
 
2321
        finally:
 
2322
            os.chdir = orig_chdir
2588
2323
        self.assertEqual(['foo', 'current'], chdirs)
2589
2324
 
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
2325
 
2596
2326
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2597
2327
    """Tests that really need to do things with an external bzr."""
2610
2340
        self.assertEqual('bzr: interrupted\n', result[1])
2611
2341
 
2612
2342
 
 
2343
class TestFeature(tests.TestCase):
 
2344
 
 
2345
    def test_caching(self):
 
2346
        """Feature._probe is called by the feature at most once."""
 
2347
        class InstrumentedFeature(tests.Feature):
 
2348
            def __init__(self):
 
2349
                super(InstrumentedFeature, self).__init__()
 
2350
                self.calls = []
 
2351
            def _probe(self):
 
2352
                self.calls.append('_probe')
 
2353
                return False
 
2354
        feature = InstrumentedFeature()
 
2355
        feature.available()
 
2356
        self.assertEqual(['_probe'], feature.calls)
 
2357
        feature.available()
 
2358
        self.assertEqual(['_probe'], feature.calls)
 
2359
 
 
2360
    def test_named_str(self):
 
2361
        """Feature.__str__ should thunk to feature_name()."""
 
2362
        class NamedFeature(tests.Feature):
 
2363
            def feature_name(self):
 
2364
                return 'symlinks'
 
2365
        feature = NamedFeature()
 
2366
        self.assertEqual('symlinks', str(feature))
 
2367
 
 
2368
    def test_default_str(self):
 
2369
        """Feature.__str__ should default to __class__.__name__."""
 
2370
        class NamedFeature(tests.Feature):
 
2371
            pass
 
2372
        feature = NamedFeature()
 
2373
        self.assertEqual('NamedFeature', str(feature))
 
2374
 
 
2375
 
 
2376
class TestUnavailableFeature(tests.TestCase):
 
2377
 
 
2378
    def test_access_feature(self):
 
2379
        feature = tests.Feature()
 
2380
        exception = tests.UnavailableFeature(feature)
 
2381
        self.assertIs(feature, exception.args[0])
 
2382
 
 
2383
 
 
2384
class TestModuleAvailableFeature(tests.TestCase):
 
2385
 
 
2386
    def test_available_module(self):
 
2387
        feature = tests.ModuleAvailableFeature('bzrlib.tests')
 
2388
        self.assertEqual('bzrlib.tests', feature.module_name)
 
2389
        self.assertEqual('bzrlib.tests', str(feature))
 
2390
        self.assertTrue(feature.available())
 
2391
        self.assertIs(tests, feature.module)
 
2392
 
 
2393
    def test_unavailable_module(self):
 
2394
        feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
 
2395
        self.assertEqual('bzrlib.no_such_module_exists', str(feature))
 
2396
        self.assertFalse(feature.available())
 
2397
        self.assertIs(None, feature.module)
 
2398
 
 
2399
 
2613
2400
class TestSelftestFiltering(tests.TestCase):
2614
2401
 
2615
2402
    def setUp(self):
2766
2553
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2767
2554
 
2768
2555
 
2769
 
class TestCheckTreeShape(tests.TestCaseWithTransport):
 
2556
class TestCheckInventoryShape(tests.TestCaseWithTransport):
2770
2557
 
2771
 
    def test_check_tree_shape(self):
 
2558
    def test_check_inventory_shape(self):
2772
2559
        files = ['a', 'b/', 'b/c']
2773
2560
        tree = self.make_branch_and_tree('.')
2774
2561
        self.build_tree(files)
2775
2562
        tree.add(files)
2776
2563
        tree.lock_read()
2777
2564
        try:
2778
 
            self.check_tree_shape(tree, files)
 
2565
            self.check_inventory_shape(tree.inventory, files)
2779
2566
        finally:
2780
2567
            tree.unlock()
2781
2568
 
2800
2587
        # Running bzr in blackbox mode, normal/expected/user errors should be
2801
2588
        # caught in the regular way and turned into an error message plus exit
2802
2589
        # code.
2803
 
        transport_server = memory.MemoryServer()
2804
 
        transport_server.start_server()
2805
 
        self.addCleanup(transport_server.stop_server)
 
2590
        transport_server = MemoryServer()
 
2591
        transport_server.setUp()
 
2592
        self.addCleanup(transport_server.tearDown)
2806
2593
        url = transport_server.get_url()
2807
2594
        self.permit_url(url)
2808
2595
        out, err = self.run_bzr(["log", "%s/nonexistantpath" % url], retcode=3)
2952
2739
        # Test that a plausible list of modules to doctest is returned
2953
2740
        # by _test_suite_modules_to_doctest.
2954
2741
        test_list = tests._test_suite_modules_to_doctest()
2955
 
        if __doc__ is None:
2956
 
            # When docstrings are stripped, there are no modules to doctest
2957
 
            self.assertEqual([], test_list)
2958
 
            return
2959
2742
        self.assertSubset([
2960
2743
            'bzrlib.timestamp',
2961
2744
            ],
2968
2751
        # test doubles that supply a few sample tests to load, and check they
2969
2752
        # are loaded.
2970
2753
        calls = []
2971
 
        def testmod_names():
 
2754
        def _test_suite_testmod_names():
2972
2755
            calls.append("testmod_names")
2973
2756
            return [
2974
2757
                'bzrlib.tests.blackbox.test_branch',
2975
2758
                'bzrlib.tests.per_transport',
2976
2759
                'bzrlib.tests.test_selftest',
2977
2760
                ]
2978
 
        self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
2979
 
        def doctests():
 
2761
        original_testmod_names = tests._test_suite_testmod_names
 
2762
        def _test_suite_modules_to_doctest():
2980
2763
            calls.append("modules_to_doctest")
2981
 
            if __doc__ is None:
2982
 
                return []
2983
2764
            return ['bzrlib.timestamp']
2984
 
        self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
 
2765
        orig_modules_to_doctest = tests._test_suite_modules_to_doctest
 
2766
        def restore_names():
 
2767
            tests._test_suite_testmod_names = original_testmod_names
 
2768
            tests._test_suite_modules_to_doctest = orig_modules_to_doctest
 
2769
        self.addCleanup(restore_names)
 
2770
        tests._test_suite_testmod_names = _test_suite_testmod_names
 
2771
        tests._test_suite_modules_to_doctest = _test_suite_modules_to_doctest
2985
2772
        expected_test_list = [
2986
2773
            # testmod_names
2987
2774
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
2988
2775
            ('bzrlib.tests.per_transport.TransportTests'
2989
2776
             '.test_abspath(LocalTransport,LocalURLServer)'),
2990
2777
            'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
 
2778
            # modules_to_doctest
 
2779
            'bzrlib.timestamp.format_highres_date',
2991
2780
            # plugins can't be tested that way since selftest may be run with
2992
2781
            # --no-plugins
2993
2782
            ]
2994
 
        if __doc__ is not None:
2995
 
            expected_test_list.extend([
2996
 
                # modules_to_doctest
2997
 
                'bzrlib.timestamp.format_highres_date',
2998
 
                ])
2999
2783
        suite = tests.test_suite()
3000
2784
        self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
3001
2785
            set(calls))
3113
2897
        tpr.register('bar', 'bBB.aAA.rRR')
3114
2898
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
3115
2899
        self.assertThat(self.get_log(),
3116
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
3117
 
                           doctest.ELLIPSIS))
 
2900
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
3118
2901
 
3119
2902
    def test_get_unknown_prefix(self):
3120
2903
        tpr = self._get_registry()
3140
2923
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3141
2924
 
3142
2925
 
3143
 
class TestThreadLeakDetection(tests.TestCase):
3144
 
    """Ensure when tests leak threads we detect and report it"""
3145
 
 
3146
 
    class LeakRecordingResult(tests.ExtendedTestResult):
3147
 
        def __init__(self):
3148
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3149
 
            self.leaks = []
3150
 
        def _report_thread_leak(self, test, leaks, alive):
3151
 
            self.leaks.append((test, leaks))
3152
 
 
3153
 
    def test_testcase_without_addCleanups(self):
3154
 
        """Check old TestCase instances don't break with leak detection"""
3155
 
        class Test(unittest.TestCase):
3156
 
            def runTest(self):
3157
 
                pass
3158
 
        result = self.LeakRecordingResult()
3159
 
        test = Test()
3160
 
        result.startTestRun()
3161
 
        test.run(result)
3162
 
        result.stopTestRun()
3163
 
        self.assertEqual(result._tests_leaking_threads_count, 0)
3164
 
        self.assertEqual(result.leaks, [])
3165
 
        
3166
 
    def test_thread_leak(self):
3167
 
        """Ensure a thread that outlives the running of a test is reported
3168
 
 
3169
 
        Uses a thread that blocks on an event, and is started by the inner
3170
 
        test case. As the thread outlives the inner case's run, it should be
3171
 
        detected as a leak, but the event is then set so that the thread can
3172
 
        be safely joined in cleanup so it's not leaked for real.
3173
 
        """
3174
 
        event = threading.Event()
3175
 
        thread = threading.Thread(name="Leaker", target=event.wait)
3176
 
        class Test(tests.TestCase):
3177
 
            def test_leak(self):
3178
 
                thread.start()
3179
 
        result = self.LeakRecordingResult()
3180
 
        test = Test("test_leak")
3181
 
        self.addCleanup(thread.join)
3182
 
        self.addCleanup(event.set)
3183
 
        result.startTestRun()
3184
 
        test.run(result)
3185
 
        result.stopTestRun()
3186
 
        self.assertEqual(result._tests_leaking_threads_count, 1)
3187
 
        self.assertEqual(result._first_thread_leaker_id, test.id())
3188
 
        self.assertEqual(result.leaks, [(test, set([thread]))])
3189
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3190
 
 
3191
 
    def test_multiple_leaks(self):
3192
 
        """Check multiple leaks are blamed on the test cases at fault
3193
 
 
3194
 
        Same concept as the previous test, but has one inner test method that
3195
 
        leaks two threads, and one that doesn't leak at all.
3196
 
        """
3197
 
        event = threading.Event()
3198
 
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
3199
 
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
3200
 
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
3201
 
        class Test(tests.TestCase):
3202
 
            def test_first_leak(self):
3203
 
                thread_b.start()
3204
 
            def test_second_no_leak(self):
3205
 
                pass
3206
 
            def test_third_leak(self):
3207
 
                thread_c.start()
3208
 
                thread_a.start()
3209
 
        result = self.LeakRecordingResult()
3210
 
        first_test = Test("test_first_leak")
3211
 
        third_test = Test("test_third_leak")
3212
 
        self.addCleanup(thread_a.join)
3213
 
        self.addCleanup(thread_b.join)
3214
 
        self.addCleanup(thread_c.join)
3215
 
        self.addCleanup(event.set)
3216
 
        result.startTestRun()
3217
 
        unittest.TestSuite(
3218
 
            [first_test, Test("test_second_no_leak"), third_test]
3219
 
            ).run(result)
3220
 
        result.stopTestRun()
3221
 
        self.assertEqual(result._tests_leaking_threads_count, 2)
3222
 
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
3223
 
        self.assertEqual(result.leaks, [
3224
 
            (first_test, set([thread_b])),
3225
 
            (third_test, set([thread_a, thread_c]))])
3226
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3227
 
 
3228
 
 
3229
 
class TestPostMortemDebugging(tests.TestCase):
3230
 
    """Check post mortem debugging works when tests fail or error"""
3231
 
 
3232
 
    class TracebackRecordingResult(tests.ExtendedTestResult):
3233
 
        def __init__(self):
3234
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3235
 
            self.postcode = None
3236
 
        def _post_mortem(self, tb=None):
3237
 
            """Record the code object at the end of the current traceback"""
3238
 
            tb = tb or sys.exc_info()[2]
3239
 
            if tb is not None:
3240
 
                next = tb.tb_next
3241
 
                while next is not None:
3242
 
                    tb = next
3243
 
                    next = next.tb_next
3244
 
                self.postcode = tb.tb_frame.f_code
3245
 
        def report_error(self, test, err):
3246
 
            pass
3247
 
        def report_failure(self, test, err):
3248
 
            pass
3249
 
 
3250
 
    def test_location_unittest_error(self):
3251
 
        """Needs right post mortem traceback with erroring unittest case"""
3252
 
        class Test(unittest.TestCase):
3253
 
            def runTest(self):
3254
 
                raise RuntimeError
3255
 
        result = self.TracebackRecordingResult()
3256
 
        Test().run(result)
3257
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3258
 
 
3259
 
    def test_location_unittest_failure(self):
3260
 
        """Needs right post mortem traceback with failing unittest case"""
3261
 
        class Test(unittest.TestCase):
3262
 
            def runTest(self):
3263
 
                raise self.failureException
3264
 
        result = self.TracebackRecordingResult()
3265
 
        Test().run(result)
3266
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3267
 
 
3268
 
    def test_location_bt_error(self):
3269
 
        """Needs right post mortem traceback with erroring bzrlib.tests case"""
3270
 
        class Test(tests.TestCase):
3271
 
            def test_error(self):
3272
 
                raise RuntimeError
3273
 
        result = self.TracebackRecordingResult()
3274
 
        Test("test_error").run(result)
3275
 
        self.assertEqual(result.postcode, Test.test_error.func_code)
3276
 
 
3277
 
    def test_location_bt_failure(self):
3278
 
        """Needs right post mortem traceback with failing bzrlib.tests case"""
3279
 
        class Test(tests.TestCase):
3280
 
            def test_failure(self):
3281
 
                raise self.failureException
3282
 
        result = self.TracebackRecordingResult()
3283
 
        Test("test_failure").run(result)
3284
 
        self.assertEqual(result.postcode, Test.test_failure.func_code)
3285
 
 
3286
 
    def test_env_var_triggers_post_mortem(self):
3287
 
        """Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
3288
 
        import pdb
3289
 
        result = tests.ExtendedTestResult(StringIO(), 0, 1)
3290
 
        post_mortem_calls = []
3291
 
        self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3292
 
        self.overrideEnv('BZR_TEST_PDB', None)
3293
 
        result._post_mortem(1)
3294
 
        self.overrideEnv('BZR_TEST_PDB', 'on')
3295
 
        result._post_mortem(2)
3296
 
        self.assertEqual([2], post_mortem_calls)
3297
 
 
3298
 
 
3299
2926
class TestRunSuite(tests.TestCase):
3300
2927
 
3301
2928
    def test_runner_class(self):
3312
2939
                                                self.verbosity)
3313
2940
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
3314
2941
        self.assertLength(1, calls)
3315
 
 
3316
 
 
3317
 
class TestEnvironHandling(tests.TestCase):
3318
 
 
3319
 
    def test_overrideEnv_None_called_twice_doesnt_leak(self):
3320
 
        self.assertFalse('MYVAR' in os.environ)
3321
 
        self.overrideEnv('MYVAR', '42')
3322
 
        # We use an embedded test to make sure we fix the _captureVar bug
3323
 
        class Test(tests.TestCase):
3324
 
            def test_me(self):
3325
 
                # The first call save the 42 value
3326
 
                self.overrideEnv('MYVAR', None)
3327
 
                self.assertEquals(None, os.environ.get('MYVAR'))
3328
 
                # Make sure we can call it twice
3329
 
                self.overrideEnv('MYVAR', None)
3330
 
                self.assertEquals(None, os.environ.get('MYVAR'))
3331
 
        output = StringIO()
3332
 
        result = tests.TextTestResult(output, 0, 1)
3333
 
        Test('test_me').run(result)
3334
 
        if not result.wasStrictlySuccessful():
3335
 
            self.fail(output.getvalue())
3336
 
        # We get our value back
3337
 
        self.assertEquals('42', os.environ.get('MYVAR'))
3338
 
 
3339
 
 
3340
 
class TestIsolatedEnv(tests.TestCase):
3341
 
    """Test isolating tests from os.environ.
3342
 
 
3343
 
    Since we use tests that are already isolated from os.environ a bit of care
3344
 
    should be taken when designing the tests to avoid bootstrap side-effects.
3345
 
    The tests start an already clean os.environ which allow doing valid
3346
 
    assertions about which variables are present or not and design tests around
3347
 
    these assertions.
3348
 
    """
3349
 
 
3350
 
    class ScratchMonkey(tests.TestCase):
3351
 
 
3352
 
        def test_me(self):
3353
 
            pass
3354
 
 
3355
 
    def test_basics(self):
3356
 
        # Make sure we know the definition of BZR_HOME: not part of os.environ
3357
 
        # for tests.TestCase.
3358
 
        self.assertTrue('BZR_HOME' in tests.isolated_environ)
3359
 
        self.assertEquals(None, tests.isolated_environ['BZR_HOME'])
3360
 
        # Being part of isolated_environ, BZR_HOME should not appear here
3361
 
        self.assertFalse('BZR_HOME' in os.environ)
3362
 
        # Make sure we know the definition of LINES: part of os.environ for
3363
 
        # tests.TestCase
3364
 
        self.assertTrue('LINES' in tests.isolated_environ)
3365
 
        self.assertEquals('25', tests.isolated_environ['LINES'])
3366
 
        self.assertEquals('25', os.environ['LINES'])
3367
 
 
3368
 
    def test_injecting_unknown_variable(self):
3369
 
        # BZR_HOME is known to be absent from os.environ
3370
 
        test = self.ScratchMonkey('test_me')
3371
 
        tests.override_os_environ(test, {'BZR_HOME': 'foo'})
3372
 
        self.assertEquals('foo', os.environ['BZR_HOME'])
3373
 
        tests.restore_os_environ(test)
3374
 
        self.assertFalse('BZR_HOME' in os.environ)
3375
 
 
3376
 
    def test_injecting_known_variable(self):
3377
 
        test = self.ScratchMonkey('test_me')
3378
 
        # LINES is known to be present in os.environ
3379
 
        tests.override_os_environ(test, {'LINES': '42'})
3380
 
        self.assertEquals('42', os.environ['LINES'])
3381
 
        tests.restore_os_environ(test)
3382
 
        self.assertEquals('25', os.environ['LINES'])
3383
 
 
3384
 
    def test_deleting_variable(self):
3385
 
        test = self.ScratchMonkey('test_me')
3386
 
        # LINES is known to be present in os.environ
3387
 
        tests.override_os_environ(test, {'LINES': None})
3388
 
        self.assertTrue('LINES' not in os.environ)
3389
 
        tests.restore_os_environ(test)
3390
 
        self.assertEquals('25', os.environ['LINES'])
3391
 
 
3392
 
 
3393
 
class TestDocTestSuiteIsolation(tests.TestCase):
3394
 
    """Test that `tests.DocTestSuite` isolates doc tests from os.environ.
3395
 
 
3396
 
    Since tests.TestCase alreay provides an isolation from os.environ, we use
3397
 
    the clean environment as a base for testing. To precisely capture the
3398
 
    isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
3399
 
    compare against.
3400
 
 
3401
 
    We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
3402
 
    not `os.environ` so each test overrides it to suit its needs.
3403
 
 
3404
 
    """
3405
 
 
3406
 
    def get_doctest_suite_for_string(self, klass, string):
3407
 
        class Finder(doctest.DocTestFinder):
3408
 
 
3409
 
            def find(*args, **kwargs):
3410
 
                test = doctest.DocTestParser().get_doctest(
3411
 
                    string, {}, 'foo', 'foo.py', 0)
3412
 
                return [test]
3413
 
 
3414
 
        suite = klass(test_finder=Finder())
3415
 
        return suite
3416
 
 
3417
 
    def run_doctest_suite_for_string(self, klass, string):
3418
 
        suite = self.get_doctest_suite_for_string(klass, string)
3419
 
        output = StringIO()
3420
 
        result = tests.TextTestResult(output, 0, 1)
3421
 
        suite.run(result)
3422
 
        return result, output
3423
 
 
3424
 
    def assertDocTestStringSucceds(self, klass, string):
3425
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3426
 
        if not result.wasStrictlySuccessful():
3427
 
            self.fail(output.getvalue())
3428
 
 
3429
 
    def assertDocTestStringFails(self, klass, string):
3430
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3431
 
        if result.wasStrictlySuccessful():
3432
 
            self.fail(output.getvalue())
3433
 
 
3434
 
    def test_injected_variable(self):
3435
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
3436
 
        test = """
3437
 
            >>> import os
3438
 
            >>> os.environ['LINES']
3439
 
            '42'
3440
 
            """
3441
 
        # doctest.DocTestSuite fails as it sees '25'
3442
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3443
 
        # tests.DocTestSuite sees '42'
3444
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3445
 
 
3446
 
    def test_deleted_variable(self):
3447
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
3448
 
        test = """
3449
 
            >>> import os
3450
 
            >>> os.environ.get('LINES')
3451
 
            """
3452
 
        # doctest.DocTestSuite fails as it sees '25'
3453
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3454
 
        # tests.DocTestSuite sees None
3455
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3456
 
 
3457
 
 
3458
 
class TestSelftestExcludePatterns(tests.TestCase):
3459
 
 
3460
 
    def setUp(self):
3461
 
        super(TestSelftestExcludePatterns, self).setUp()
3462
 
        self.overrideAttr(tests, 'test_suite', self.suite_factory)
3463
 
 
3464
 
    def suite_factory(self, keep_only=None, starting_with=None):
3465
 
        """A test suite factory with only a few tests."""
3466
 
        class Test(tests.TestCase):
3467
 
            def id(self):
3468
 
                # We don't need the full class path
3469
 
                return self._testMethodName
3470
 
            def a(self):
3471
 
                pass
3472
 
            def b(self):
3473
 
                pass
3474
 
            def c(self):
3475
 
                pass
3476
 
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
3477
 
 
3478
 
    def assertTestList(self, expected, *selftest_args):
3479
 
        # We rely on setUp installing the right test suite factory so we can
3480
 
        # test at the command level without loading the whole test suite
3481
 
        out, err = self.run_bzr(('selftest', '--list') + selftest_args)
3482
 
        actual = out.splitlines()
3483
 
        self.assertEquals(expected, actual)
3484
 
 
3485
 
    def test_full_list(self):
3486
 
        self.assertTestList(['a', 'b', 'c'])
3487
 
 
3488
 
    def test_single_exclude(self):
3489
 
        self.assertTestList(['b', 'c'], '-x', 'a')
3490
 
 
3491
 
    def test_mutiple_excludes(self):
3492
 
        self.assertTestList(['c'], '-x', 'a', '-x', 'b')
3493
 
 
3494
 
 
3495
 
class TestCounterHooks(tests.TestCase, SelfTestHelper):
3496
 
 
3497
 
    _test_needs_features = [features.subunit]
3498
 
 
3499
 
    def setUp(self):
3500
 
        super(TestCounterHooks, self).setUp()
3501
 
        class Test(tests.TestCase):
3502
 
 
3503
 
            def setUp(self):
3504
 
                super(Test, self).setUp()
3505
 
                self.hooks = hooks.Hooks()
3506
 
                self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
3507
 
                self.install_counter_hook(self.hooks, 'myhook')
3508
 
 
3509
 
            def no_hook(self):
3510
 
                pass
3511
 
 
3512
 
            def run_hook_once(self):
3513
 
                for hook in self.hooks['myhook']:
3514
 
                    hook(self)
3515
 
 
3516
 
        self.test_class = Test
3517
 
 
3518
 
    def assertHookCalls(self, expected_calls, test_name):
3519
 
        test = self.test_class(test_name)
3520
 
        result = unittest.TestResult()
3521
 
        test.run(result)
3522
 
        self.assertTrue(hasattr(test, '_counters'))
3523
 
        self.assertTrue(test._counters.has_key('myhook'))
3524
 
        self.assertEquals(expected_calls, test._counters['myhook'])
3525
 
 
3526
 
    def test_no_hook(self):
3527
 
        self.assertHookCalls(0, 'no_hook')
3528
 
 
3529
 
    def test_run_hook_once(self):
3530
 
        tt = features.testtools
3531
 
        if tt.module.__version__ < (0, 9, 8):
3532
 
            raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
3533
 
        self.assertHookCalls(1, 'run_hook_once')