← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-02-16 00:53:04 UTC
  • mfrom: (4997.1.3 find-branches-local)
  • Revision ID: pqm@pqm.ubuntu.com-20100216005304-1p8xafkhiizh6ugi
(Jelmer) Add BzrDir.list_branches().

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_selftest.py
1
 
# Copyright (C) 2005-2011 Canonical Ltd
 
1
# Copyright (C) 2005-2010 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
 
    )
33
 
from testtools.content import Content
 
28
from testtools import MultiTestResult
34
29
from testtools.content_type import ContentType
35
30
from testtools.matchers import (
36
31
    DocTestMatches,
37
32
    Equals,
38
33
    )
39
 
import testtools.testresult.doubles
 
34
import testtools.tests.helpers
40
35
 
41
36
import bzrlib
42
37
from bzrlib import (
43
38
    branchbuilder,
44
39
    bzrdir,
 
40
    debug,
45
41
    errors,
46
 
    hooks,
47
42
    lockdir,
48
43
    memorytree,
49
44
    osutils,
 
45
    progress,
50
46
    remote,
51
47
    repository,
52
48
    symbol_versioning,
53
49
    tests,
54
50
    transport,
55
51
    workingtree,
56
 
    workingtree_3,
57
 
    workingtree_4,
58
52
    )
59
53
from bzrlib.repofmt import (
60
54
    groupcompress_repo,
 
55
    pack_repo,
 
56
    weaverepo,
61
57
    )
62
58
from bzrlib.symbol_versioning import (
63
59
    deprecated_function,
66
62
    )
67
63
from bzrlib.tests import (
68
64
    features,
 
65
    stub_sftp,
69
66
    test_lsprof,
70
67
    test_server,
 
68
    test_sftp_transport,
71
69
    TestUtil,
72
70
    )
73
 
from bzrlib.trace import note, mutter
 
71
from bzrlib.trace import note
74
72
from bzrlib.transport import memory
 
73
from bzrlib.version import _get_bzr_source_tree
75
74
 
76
75
 
77
76
def _test_ids(test_suite):
79
78
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
80
79
 
81
80
 
 
81
class SelftestTests(tests.TestCase):
 
82
 
 
83
    def test_import_tests(self):
 
84
        mod = TestUtil._load_module_by_name('bzrlib.tests.test_selftest')
 
85
        self.assertEqual(mod.SelftestTests, SelftestTests)
 
86
 
 
87
    def test_import_test_failure(self):
 
88
        self.assertRaises(ImportError,
 
89
                          TestUtil._load_module_by_name,
 
90
                          'bzrlib.no-name-yet')
 
91
 
 
92
 
82
93
class MetaTestLog(tests.TestCase):
83
94
 
84
95
    def test_logging(self):
90
101
            "text", "plain", {"charset": "utf8"})))
91
102
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
92
103
        self.assertThat(self.get_log(),
93
 
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
 
104
            DocTestMatches(u"...a test message\n", ELLIPSIS))
94
105
 
95
106
 
96
107
class TestUnicodeFilename(tests.TestCase):
109
120
 
110
121
        filename = u'hell\u00d8'
111
122
        self.build_tree_contents([(filename, 'contents of hello')])
112
 
        self.assertPathExists(filename)
113
 
 
114
 
 
115
 
class TestClassesAvailable(tests.TestCase):
116
 
    """As a convenience we expose Test* classes from bzrlib.tests"""
117
 
 
118
 
    def test_test_case(self):
119
 
        from bzrlib.tests import TestCase
120
 
 
121
 
    def test_test_loader(self):
122
 
        from bzrlib.tests import TestLoader
123
 
 
124
 
    def test_test_suite(self):
125
 
        from bzrlib.tests import TestSuite
 
123
        self.failUnlessExists(filename)
126
124
 
127
125
 
128
126
class TestTransportScenarios(tests.TestCase):
211
209
    def test_scenarios(self):
212
210
        # check that constructor parameters are passed through to the adapted
213
211
        # test.
214
 
        from bzrlib.tests.per_controldir import make_scenarios
 
212
        from bzrlib.tests.per_bzrdir import make_scenarios
215
213
        vfs_factory = "v"
216
214
        server1 = "a"
217
215
        server2 = "b"
315
313
        from bzrlib.tests.per_interrepository import make_scenarios
316
314
        server1 = "a"
317
315
        server2 = "b"
318
 
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
 
316
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
319
317
        scenarios = make_scenarios(server1, server2, formats)
320
318
        self.assertEqual([
321
319
            ('C0,str,str',
322
320
             {'repository_format': 'C1',
323
321
              'repository_format_to': 'C2',
324
322
              'transport_readonly_server': 'b',
325
 
              'transport_server': 'a',
326
 
              'extra_setup': 'C3'}),
 
323
              'transport_server': 'a'}),
327
324
            ('D0,str,str',
328
325
             {'repository_format': 'D1',
329
326
              'repository_format_to': 'D2',
330
327
              'transport_readonly_server': 'b',
331
 
              'transport_server': 'a',
332
 
              'extra_setup': 'D3'})],
 
328
              'transport_server': 'a'})],
333
329
            scenarios)
334
330
 
335
331
 
341
337
        from bzrlib.tests.per_workingtree import make_scenarios
342
338
        server1 = "a"
343
339
        server2 = "b"
344
 
        formats = [workingtree_4.WorkingTreeFormat4(),
345
 
                   workingtree_3.WorkingTreeFormat3(),]
 
340
        formats = [workingtree.WorkingTreeFormat2(),
 
341
                   workingtree.WorkingTreeFormat3(),]
346
342
        scenarios = make_scenarios(server1, server2, formats)
347
343
        self.assertEqual([
348
 
            ('WorkingTreeFormat4',
 
344
            ('WorkingTreeFormat2',
349
345
             {'bzrdir_format': formats[0]._matchingbzrdir,
350
346
              'transport_readonly_server': 'b',
351
347
              'transport_server': 'a',
378
374
            )
379
375
        server1 = "a"
380
376
        server2 = "b"
381
 
        formats = [workingtree_4.WorkingTreeFormat4(),
382
 
                   workingtree_3.WorkingTreeFormat3(),]
 
377
        formats = [workingtree.WorkingTreeFormat2(),
 
378
                   workingtree.WorkingTreeFormat3(),]
383
379
        scenarios = make_scenarios(server1, server2, formats)
384
380
        self.assertEqual(7, len(scenarios))
385
 
        default_wt_format = workingtree.format_registry.get_default()
386
 
        wt4_format = workingtree_4.WorkingTreeFormat4()
387
 
        wt5_format = workingtree_4.WorkingTreeFormat5()
 
381
        default_wt_format = workingtree.WorkingTreeFormat4._default_format
 
382
        wt4_format = workingtree.WorkingTreeFormat4()
 
383
        wt5_format = workingtree.WorkingTreeFormat5()
388
384
        expected_scenarios = [
389
 
            ('WorkingTreeFormat4',
 
385
            ('WorkingTreeFormat2',
390
386
             {'bzrdir_format': formats[0]._matchingbzrdir,
391
387
              'transport_readonly_server': 'b',
392
388
              'transport_server': 'a',
452
448
        # ones to add.
453
449
        from bzrlib.tests.per_tree import (
454
450
            return_parameter,
 
451
            revision_tree_from_workingtree
455
452
            )
456
453
        from bzrlib.tests.per_intertree import (
457
454
            make_scenarios,
458
455
            )
459
 
        from bzrlib.workingtree_3 import WorkingTreeFormat3
460
 
        from bzrlib.workingtree_4 import WorkingTreeFormat4
 
456
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
461
457
        input_test = TestInterTreeScenarios(
462
458
            "test_scenarios")
463
459
        server1 = "a"
464
460
        server2 = "b"
465
 
        format1 = WorkingTreeFormat4()
 
461
        format1 = WorkingTreeFormat2()
466
462
        format2 = WorkingTreeFormat3()
467
463
        formats = [("1", str, format1, format2, "converter1"),
468
464
            ("2", int, format2, format1, "converter2")]
514
510
        self.assertRaises(AssertionError, self.assertEqualStat,
515
511
            os.lstat("foo"), os.lstat("longname"))
516
512
 
517
 
    def test_failUnlessExists(self):
518
 
        """Deprecated failUnlessExists and failIfExists"""
519
 
        self.applyDeprecated(
520
 
            deprecated_in((2, 4)),
521
 
            self.failUnlessExists, '.')
522
 
        self.build_tree(['foo/', 'foo/bar'])
523
 
        self.applyDeprecated(
524
 
            deprecated_in((2, 4)),
525
 
            self.failUnlessExists, 'foo/bar')
526
 
        self.applyDeprecated(
527
 
            deprecated_in((2, 4)),
528
 
            self.failIfExists, 'foo/foo')
529
 
 
530
 
    def test_assertPathExists(self):
531
 
        self.assertPathExists('.')
532
 
        self.build_tree(['foo/', 'foo/bar'])
533
 
        self.assertPathExists('foo/bar')
534
 
        self.assertPathDoesNotExist('foo/foo')
535
 
 
536
513
 
537
514
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
538
515
 
572
549
        tree = self.make_branch_and_memory_tree('dir')
573
550
        # Guard against regression into MemoryTransport leaking
574
551
        # files to disk instead of keeping them in memory.
575
 
        self.assertFalse(osutils.lexists('dir'))
 
552
        self.failIf(osutils.lexists('dir'))
576
553
        self.assertIsInstance(tree, memorytree.MemoryTree)
577
554
 
578
555
    def test_make_branch_and_memory_tree_with_format(self):
579
556
        """make_branch_and_memory_tree should accept a format option."""
580
557
        format = bzrdir.BzrDirMetaFormat1()
581
 
        format.repository_format = repository.format_registry.get_default()
 
558
        format.repository_format = weaverepo.RepositoryFormat7()
582
559
        tree = self.make_branch_and_memory_tree('dir', format=format)
583
560
        # Guard against regression into MemoryTransport leaking
584
561
        # files to disk instead of keeping them in memory.
585
 
        self.assertFalse(osutils.lexists('dir'))
 
562
        self.failIf(osutils.lexists('dir'))
586
563
        self.assertIsInstance(tree, memorytree.MemoryTree)
587
564
        self.assertEqual(format.repository_format.__class__,
588
565
            tree.branch.repository._format.__class__)
592
569
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
593
570
        # Guard against regression into MemoryTransport leaking
594
571
        # files to disk instead of keeping them in memory.
595
 
        self.assertFalse(osutils.lexists('dir'))
 
572
        self.failIf(osutils.lexists('dir'))
596
573
 
597
574
    def test_make_branch_builder_with_format(self):
598
575
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
599
576
        # that the format objects are used.
600
577
        format = bzrdir.BzrDirMetaFormat1()
601
 
        repo_format = repository.format_registry.get_default()
 
578
        repo_format = weaverepo.RepositoryFormat7()
602
579
        format.repository_format = repo_format
603
580
        builder = self.make_branch_builder('dir', format=format)
604
581
        the_branch = builder.get_branch()
605
582
        # Guard against regression into MemoryTransport leaking
606
583
        # files to disk instead of keeping them in memory.
607
 
        self.assertFalse(osutils.lexists('dir'))
 
584
        self.failIf(osutils.lexists('dir'))
608
585
        self.assertEqual(format.repository_format.__class__,
609
586
                         the_branch.repository._format.__class__)
610
587
        self.assertEqual(repo_format.get_format_string(),
616
593
        the_branch = builder.get_branch()
617
594
        # Guard against regression into MemoryTransport leaking
618
595
        # files to disk instead of keeping them in memory.
619
 
        self.assertFalse(osutils.lexists('dir'))
 
596
        self.failIf(osutils.lexists('dir'))
620
597
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
621
598
        self.assertEqual(dir_format.repository_format.__class__,
622
599
                         the_branch.repository._format.__class__)
633
610
                l.attempt_lock()
634
611
        test = TestDanglingLock('test_function')
635
612
        result = test.run()
636
 
        total_failures = result.errors + result.failures
637
613
        if self._lock_check_thorough:
638
 
            self.assertEqual(1, len(total_failures))
 
614
            self.assertEqual(1, len(result.errors))
639
615
        else:
640
616
            # When _lock_check_thorough is disabled, then we don't trigger a
641
617
            # failure
642
 
            self.assertEqual(0, len(total_failures))
 
618
            self.assertEqual(0, len(result.errors))
643
619
 
644
620
 
645
621
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
646
622
    """Tests for the convenience functions TestCaseWithTransport introduces."""
647
623
 
648
624
    def test_get_readonly_url_none(self):
 
625
        from bzrlib.transport import get_transport
649
626
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
650
627
        self.vfs_transport_factory = memory.MemoryServer
651
628
        self.transport_readonly_server = None
653
630
        # for the server
654
631
        url = self.get_readonly_url()
655
632
        url2 = self.get_readonly_url('foo/bar')
656
 
        t = transport.get_transport(url)
657
 
        t2 = transport.get_transport(url2)
658
 
        self.assertIsInstance(t, ReadonlyTransportDecorator)
659
 
        self.assertIsInstance(t2, ReadonlyTransportDecorator)
 
633
        t = get_transport(url)
 
634
        t2 = get_transport(url2)
 
635
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
 
636
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
660
637
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
661
638
 
662
639
    def test_get_readonly_url_http(self):
663
640
        from bzrlib.tests.http_server import HttpServer
 
641
        from bzrlib.transport import get_transport
664
642
        from bzrlib.transport.http import HttpTransportBase
665
643
        self.transport_server = test_server.LocalURLServer
666
644
        self.transport_readonly_server = HttpServer
668
646
        url = self.get_readonly_url()
669
647
        url2 = self.get_readonly_url('foo/bar')
670
648
        # the transport returned may be any HttpTransportBase subclass
671
 
        t = transport.get_transport(url)
672
 
        t2 = transport.get_transport(url2)
673
 
        self.assertIsInstance(t, HttpTransportBase)
674
 
        self.assertIsInstance(t2, HttpTransportBase)
 
649
        t = get_transport(url)
 
650
        t2 = get_transport(url2)
 
651
        self.failUnless(isinstance(t, HttpTransportBase))
 
652
        self.failUnless(isinstance(t2, HttpTransportBase))
675
653
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
676
654
 
677
655
    def test_is_directory(self):
685
663
    def test_make_branch_builder(self):
686
664
        builder = self.make_branch_builder('dir')
687
665
        rev_id = builder.build_commit()
688
 
        self.assertPathExists('dir')
 
666
        self.failUnlessExists('dir')
689
667
        a_dir = bzrdir.BzrDir.open('dir')
690
668
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
691
669
        a_branch = a_dir.open_branch()
707
685
        self.assertIsInstance(result_bzrdir.transport,
708
686
                              memory.MemoryTransport)
709
687
        # should not be on disk, should only be in memory
710
 
        self.assertPathDoesNotExist('subdir')
 
688
        self.failIfExists('subdir')
711
689
 
712
690
 
713
691
class TestChrootedTest(tests.ChrootedTestCase):
714
692
 
715
693
    def test_root_is_root(self):
716
 
        t = transport.get_transport(self.get_readonly_url())
 
694
        from bzrlib.transport import get_transport
 
695
        t = get_transport(self.get_readonly_url())
717
696
        url = t.base
718
697
        self.assertEqual(url, t.clone('..').base)
719
698
 
722
701
 
723
702
    def test_profiles_tests(self):
724
703
        self.requireFeature(test_lsprof.LSProfFeature)
725
 
        terminal = testtools.testresult.doubles.ExtendedTestResult()
 
704
        terminal = testtools.tests.helpers.ExtendedTestResult()
726
705
        result = tests.ProfileResult(terminal)
727
706
        class Sample(tests.TestCase):
728
707
            def a(self):
745
724
                descriptions=0,
746
725
                verbosity=1,
747
726
                )
748
 
        capture = testtools.testresult.doubles.ExtendedTestResult()
 
727
        capture = testtools.tests.helpers.ExtendedTestResult()
749
728
        test_case.run(MultiTestResult(result, capture))
750
729
        run_case = capture._events[0][1]
751
730
        timed_string = result._testTimeString(run_case)
772
751
        self.check_timing(ShortDelayTestCase('test_short_delay'),
773
752
                          r"^ +[0-9]+ms$")
774
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
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', lambda: None)
 
758
 
 
759
    def test_assigned_benchmark_file_stores_date(self):
 
760
        self._patch_get_bzr_source_tree()
 
761
        output = StringIO()
 
762
        result = bzrlib.tests.TextTestResult(self._log_file,
 
763
                                        descriptions=0,
 
764
                                        verbosity=1,
 
765
                                        bench_history=output
 
766
                                        )
 
767
        output_string = output.getvalue()
 
768
        # if you are wondering about the regexp please read the comment in
 
769
        # test_bench_history (bzrlib.tests.test_selftest.TestRunner)
 
770
        # XXX: what comment?  -- Andrew Bennetts
 
771
        self.assertContainsRe(output_string, "--date [0-9.]+")
 
772
 
 
773
    def test_benchhistory_records_test_times(self):
 
774
        self._patch_get_bzr_source_tree()
 
775
        result_stream = StringIO()
 
776
        result = bzrlib.tests.TextTestResult(
 
777
            self._log_file,
 
778
            descriptions=0,
 
779
            verbosity=1,
 
780
            bench_history=result_stream
 
781
            )
 
782
 
 
783
        # we want profile a call and check that its test duration is recorded
 
784
        # make a new test instance that when run will generate a benchmark
 
785
        example_test_case = TestTestResult("_time_hello_world_encoding")
 
786
        # execute the test, which should succeed and record times
 
787
        example_test_case.run(result)
 
788
        lines = result_stream.getvalue().splitlines()
 
789
        self.assertEqual(2, len(lines))
 
790
        self.assertContainsRe(lines[1],
 
791
            " *[0-9]+ms bzrlib.tests.test_selftest.TestTestResult"
 
792
            "._time_hello_world_encoding")
 
793
 
775
794
    def _time_hello_world_encoding(self):
776
795
        """Profile two sleep calls
777
796
 
785
804
        self.requireFeature(test_lsprof.LSProfFeature)
786
805
        result_stream = StringIO()
787
806
        result = bzrlib.tests.VerboseTestResult(
788
 
            result_stream,
 
807
            unittest._WritelnDecorator(result_stream),
789
808
            descriptions=0,
790
809
            verbosity=2,
791
810
            )
817
836
        self.assertContainsRe(output,
818
837
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
819
838
 
820
 
    def test_uses_time_from_testtools(self):
821
 
        """Test case timings in verbose results should use testtools times"""
822
 
        import datetime
823
 
        class TimeAddedVerboseTestResult(tests.VerboseTestResult):
824
 
            def startTest(self, test):
825
 
                self.time(datetime.datetime.utcfromtimestamp(1.145))
826
 
                super(TimeAddedVerboseTestResult, self).startTest(test)
827
 
            def addSuccess(self, test):
828
 
                self.time(datetime.datetime.utcfromtimestamp(51.147))
829
 
                super(TimeAddedVerboseTestResult, self).addSuccess(test)
830
 
            def report_tests_starting(self): pass
831
 
        sio = StringIO()
832
 
        self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
833
 
        self.assertEndsWith(sio.getvalue(), "OK    50002ms\n")
834
 
 
835
839
    def test_known_failure(self):
836
840
        """A KnownFailure being raised should trigger several result actions."""
837
841
        class InstrumentedTestResult(tests.ExtendedTestResult):
838
842
            def stopTestRun(self): pass
839
 
            def report_tests_starting(self): pass
 
843
            def startTests(self): pass
 
844
            def report_test_start(self, test): pass
840
845
            def report_known_failure(self, test, err=None, details=None):
841
846
                self._call = test, 'known failure'
842
847
        result = InstrumentedTestResult(None, None, None, None)
860
865
        # verbose test output formatting
861
866
        result_stream = StringIO()
862
867
        result = bzrlib.tests.VerboseTestResult(
863
 
            result_stream,
 
868
            unittest._WritelnDecorator(result_stream),
864
869
            descriptions=0,
865
870
            verbosity=2,
866
871
            )
876
881
        output = result_stream.getvalue()[prefix:]
877
882
        lines = output.splitlines()
878
883
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
879
 
        if sys.version_info > (2, 7):
880
 
            self.expectFailure("_ExpectedFailure on 2.7 loses the message",
881
 
                self.assertNotEqual, lines[1], '    ')
882
884
        self.assertEqual(lines[1], '    foo')
883
885
        self.assertEqual(2, len(lines))
884
886
 
892
894
        """Test the behaviour of invoking addNotSupported."""
893
895
        class InstrumentedTestResult(tests.ExtendedTestResult):
894
896
            def stopTestRun(self): pass
895
 
            def report_tests_starting(self): pass
 
897
            def startTests(self): pass
 
898
            def report_test_start(self, test): pass
896
899
            def report_unsupported(self, test, feature):
897
900
                self._call = test, feature
898
901
        result = InstrumentedTestResult(None, None, None, None)
917
920
        # verbose test output formatting
918
921
        result_stream = StringIO()
919
922
        result = bzrlib.tests.VerboseTestResult(
920
 
            result_stream,
 
923
            unittest._WritelnDecorator(result_stream),
921
924
            descriptions=0,
922
925
            verbosity=2,
923
926
            )
937
940
        """An UnavailableFeature being raised should invoke addNotSupported."""
938
941
        class InstrumentedTestResult(tests.ExtendedTestResult):
939
942
            def stopTestRun(self): pass
940
 
            def report_tests_starting(self): pass
 
943
            def startTests(self): pass
 
944
            def report_test_start(self, test): pass
941
945
            def addNotSupported(self, test, feature):
942
946
                self._call = test, feature
943
947
        result = InstrumentedTestResult(None, None, None, None)
985
989
        class InstrumentedTestResult(tests.ExtendedTestResult):
986
990
            calls = 0
987
991
            def startTests(self): self.calls += 1
 
992
            def report_test_start(self, test): pass
988
993
        result = InstrumentedTestResult(None, None, None, None)
989
994
        def test_function():
990
995
            pass
992
997
        test.run(result)
993
998
        self.assertEquals(1, result.calls)
994
999
 
995
 
    def test_startTests_only_once(self):
996
 
        """With multiple tests startTests should still only be called once"""
997
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
998
 
            calls = 0
999
 
            def startTests(self): self.calls += 1
1000
 
        result = InstrumentedTestResult(None, None, None, None)
1001
 
        suite = unittest.TestSuite([
1002
 
            unittest.FunctionTestCase(lambda: None),
1003
 
            unittest.FunctionTestCase(lambda: None)])
1004
 
        suite.run(result)
1005
 
        self.assertEquals(1, result.calls)
1006
 
        self.assertEquals(2, result.count)
1007
 
 
1008
1000
 
1009
1001
class TestUnicodeFilenameFeature(tests.TestCase):
1010
1002
 
1031
1023
        because of our use of global state.
1032
1024
        """
1033
1025
        old_root = tests.TestCaseInTempDir.TEST_ROOT
 
1026
        old_leak = tests.TestCase._first_thread_leaker_id
1034
1027
        try:
1035
1028
            tests.TestCaseInTempDir.TEST_ROOT = None
 
1029
            tests.TestCase._first_thread_leaker_id = None
1036
1030
            return testrunner.run(test)
1037
1031
        finally:
1038
1032
            tests.TestCaseInTempDir.TEST_ROOT = old_root
 
1033
            tests.TestCase._first_thread_leaker_id = old_leak
1039
1034
 
1040
1035
    def test_known_failure_failed_run(self):
1041
1036
        # run a test that generates a known failure which should be printed in
1046
1041
        test = unittest.TestSuite()
1047
1042
        test.addTest(Test("known_failure_test"))
1048
1043
        def failing_test():
1049
 
            raise AssertionError('foo')
 
1044
            self.fail('foo')
1050
1045
        test.addTest(unittest.FunctionTestCase(failing_test))
1051
1046
        stream = StringIO()
1052
1047
        runner = tests.TextTestRunner(stream=stream)
1060
1055
            '^----------------------------------------------------------------------\n'
1061
1056
            'Traceback \\(most recent call last\\):\n'
1062
1057
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1063
 
            '    raise AssertionError\\(\'foo\'\\)\n'
 
1058
            '    self.fail\\(\'foo\'\\)\n'
1064
1059
            '.*'
1065
1060
            '^----------------------------------------------------------------------\n'
1066
1061
            '.*'
1072
1067
        # the final output.
1073
1068
        class Test(tests.TestCase):
1074
1069
            def known_failure_test(self):
1075
 
                self.knownFailure("Never works...")
 
1070
                self.expectFailure('failed', self.assertTrue, False)
1076
1071
        test = Test("known_failure_test")
1077
1072
        stream = StringIO()
1078
1073
        runner = tests.TextTestRunner(stream=stream)
1084
1079
            '\n'
1085
1080
            'OK \\(known_failures=1\\)\n')
1086
1081
 
1087
 
    def test_unexpected_success_bad(self):
1088
 
        class Test(tests.TestCase):
1089
 
            def test_truth(self):
1090
 
                self.expectFailure("No absolute truth", self.assertTrue, True)
1091
 
        runner = tests.TextTestRunner(stream=StringIO())
1092
 
        result = self.run_test_runner(runner, Test("test_truth"))
1093
 
        self.assertContainsRe(runner.stream.getvalue(),
1094
 
            "=+\n"
1095
 
            "FAIL: \\S+\.test_truth\n"
1096
 
            "-+\n"
1097
 
            "(?:.*\n)*"
1098
 
            "No absolute truth\n"
1099
 
            "(?:.*\n)*"
1100
 
            "-+\n"
1101
 
            "Ran 1 test in .*\n"
1102
 
            "\n"
1103
 
            "FAILED \\(failures=1\\)\n\\Z")
1104
 
 
1105
1082
    def test_result_decorator(self):
1106
1083
        # decorate results
1107
1084
        calls = []
1108
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
 
1085
        class LoggingDecorator(tests.ForwardingResult):
1109
1086
            def startTest(self, test):
1110
 
                ExtendedToOriginalDecorator.startTest(self, test)
 
1087
                tests.ForwardingResult.startTest(self, test)
1111
1088
                calls.append('start')
1112
1089
        test = unittest.FunctionTestCase(lambda:None)
1113
1090
        stream = StringIO()
1214
1191
            ],
1215
1192
            lines[-3:])
1216
1193
 
1217
 
    def test_verbose_test_count(self):
1218
 
        """A verbose test run reports the right test count at the start"""
1219
 
        suite = TestUtil.TestSuite([
1220
 
            unittest.FunctionTestCase(lambda:None),
1221
 
            unittest.FunctionTestCase(lambda:None)])
1222
 
        self.assertEqual(suite.countTestCases(), 2)
1223
 
        stream = StringIO()
1224
 
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
1225
 
        # Need to use the CountingDecorator as that's what sets num_tests
1226
 
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
1227
 
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
 
1194
    def _patch_get_bzr_source_tree(self):
 
1195
        # Reading from the actual source tree breaks isolation, but we don't
 
1196
        # want to assume that thats *all* that would happen.
 
1197
        self._get_source_tree_calls = []
 
1198
        def new_get():
 
1199
            self._get_source_tree_calls.append("called")
 
1200
            return None
 
1201
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree',  new_get)
 
1202
 
 
1203
    def test_bench_history(self):
 
1204
        # tests that the running the benchmark passes bench_history into
 
1205
        # the test result object. We can tell that happens if
 
1206
        # _get_bzr_source_tree is called.
 
1207
        self._patch_get_bzr_source_tree()
 
1208
        test = TestRunner('dummy_test')
 
1209
        output = StringIO()
 
1210
        runner = tests.TextTestRunner(stream=self._log_file,
 
1211
                                      bench_history=output)
 
1212
        result = self.run_test_runner(runner, test)
 
1213
        output_string = output.getvalue()
 
1214
        self.assertContainsRe(output_string, "--date [0-9.]+")
 
1215
        self.assertLength(1, self._get_source_tree_calls)
1228
1216
 
1229
1217
    def test_startTestRun(self):
1230
1218
        """run should call result.startTestRun()"""
1231
1219
        calls = []
1232
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
 
1220
        class LoggingDecorator(tests.ForwardingResult):
1233
1221
            def startTestRun(self):
1234
 
                ExtendedToOriginalDecorator.startTestRun(self)
 
1222
                tests.ForwardingResult.startTestRun(self)
1235
1223
                calls.append('startTestRun')
1236
1224
        test = unittest.FunctionTestCase(lambda:None)
1237
1225
        stream = StringIO()
1243
1231
    def test_stopTestRun(self):
1244
1232
        """run should call result.stopTestRun()"""
1245
1233
        calls = []
1246
 
        class LoggingDecorator(ExtendedToOriginalDecorator):
 
1234
        class LoggingDecorator(tests.ForwardingResult):
1247
1235
            def stopTestRun(self):
1248
 
                ExtendedToOriginalDecorator.stopTestRun(self)
 
1236
                tests.ForwardingResult.stopTestRun(self)
1249
1237
                calls.append('stopTestRun')
1250
1238
        test = unittest.FunctionTestCase(lambda:None)
1251
1239
        stream = StringIO()
1254
1242
        result = self.run_test_runner(runner, test)
1255
1243
        self.assertLength(1, calls)
1256
1244
 
1257
 
    def test_unicode_test_output_on_ascii_stream(self):
1258
 
        """Showing results should always succeed even on an ascii console"""
1259
 
        class FailureWithUnicode(tests.TestCase):
1260
 
            def test_log_unicode(self):
1261
 
                self.log(u"\u2606")
1262
 
                self.fail("Now print that log!")
1263
 
        out = StringIO()
1264
 
        self.overrideAttr(osutils, "get_terminal_encoding",
1265
 
            lambda trace=False: "ascii")
1266
 
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
1267
 
            FailureWithUnicode("test_log_unicode"))
1268
 
        self.assertContainsRe(out.getvalue(),
1269
 
            "Text attachment: log\n"
1270
 
            "-+\n"
1271
 
            "\d+\.\d+  \\\\u2606\n"
1272
 
            "-+\n")
1273
 
 
1274
1245
 
1275
1246
class SampleTestCase(tests.TestCase):
1276
1247
 
1451
1422
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1452
1423
        output_stream = StringIO()
1453
1424
        result = bzrlib.tests.VerboseTestResult(
1454
 
            output_stream,
 
1425
            unittest._WritelnDecorator(output_stream),
1455
1426
            descriptions=0,
1456
1427
            verbosity=2)
1457
1428
        sample_test.run(result)
1464
1435
        # Note this test won't fail with hooks that the core library doesn't
1465
1436
        # use - but it trigger with a plugin that adds hooks, so its still a
1466
1437
        # useful warning in that case.
1467
 
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
1468
 
        self.assertEqual(
1469
 
            bzrlib.smart.server.SmartServerHooks(),
 
1438
        self.assertEqual(bzrlib.branch.BranchHooks(),
 
1439
            bzrlib.branch.Branch.hooks)
 
1440
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
1470
1441
            bzrlib.smart.server.SmartTCPServer.hooks)
1471
 
        self.assertEqual(
1472
 
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
 
1442
        self.assertEqual(bzrlib.commands.CommandHooks(),
 
1443
            bzrlib.commands.Command.hooks)
1473
1444
 
1474
1445
    def test__gather_lsprof_in_benchmarks(self):
1475
1446
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1685
1656
        self.assertEqual('original', obj.test_attr)
1686
1657
 
1687
1658
 
1688
 
class _MissingFeature(tests.Feature):
1689
 
    def _probe(self):
1690
 
        return False
1691
 
missing_feature = _MissingFeature()
1692
 
 
1693
 
 
1694
 
def _get_test(name):
1695
 
    """Get an instance of a specific example test.
1696
 
 
1697
 
    We protect this in a function so that they don't auto-run in the test
1698
 
    suite.
1699
 
    """
1700
 
 
1701
 
    class ExampleTests(tests.TestCase):
1702
 
 
1703
 
        def test_fail(self):
1704
 
            mutter('this was a failing test')
1705
 
            self.fail('this test will fail')
1706
 
 
1707
 
        def test_error(self):
1708
 
            mutter('this test errored')
1709
 
            raise RuntimeError('gotcha')
1710
 
 
1711
 
        def test_missing_feature(self):
1712
 
            mutter('missing the feature')
1713
 
            self.requireFeature(missing_feature)
1714
 
 
1715
 
        def test_skip(self):
1716
 
            mutter('this test will be skipped')
1717
 
            raise tests.TestSkipped('reason')
1718
 
 
1719
 
        def test_success(self):
1720
 
            mutter('this test succeeds')
1721
 
 
1722
 
        def test_xfail(self):
1723
 
            mutter('test with expected failure')
1724
 
            self.knownFailure('this_fails')
1725
 
 
1726
 
        def test_unexpected_success(self):
1727
 
            mutter('test with unexpected success')
1728
 
            self.expectFailure('should_fail', lambda: None)
1729
 
 
1730
 
    return ExampleTests(name)
1731
 
 
1732
 
 
1733
 
class TestTestCaseLogDetails(tests.TestCase):
1734
 
 
1735
 
    def _run_test(self, test_name):
1736
 
        test = _get_test(test_name)
1737
 
        result = testtools.TestResult()
1738
 
        test.run(result)
1739
 
        return result
1740
 
 
1741
 
    def test_fail_has_log(self):
1742
 
        result = self._run_test('test_fail')
1743
 
        self.assertEqual(1, len(result.failures))
1744
 
        result_content = result.failures[0][1]
1745
 
        self.assertContainsRe(result_content, 'Text attachment: log')
1746
 
        self.assertContainsRe(result_content, 'this was a failing test')
1747
 
 
1748
 
    def test_error_has_log(self):
1749
 
        result = self._run_test('test_error')
1750
 
        self.assertEqual(1, len(result.errors))
1751
 
        result_content = result.errors[0][1]
1752
 
        self.assertContainsRe(result_content, 'Text attachment: log')
1753
 
        self.assertContainsRe(result_content, 'this test errored')
1754
 
 
1755
 
    def test_skip_has_no_log(self):
1756
 
        result = self._run_test('test_skip')
1757
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
1758
 
        skips = result.skip_reasons['reason']
1759
 
        self.assertEqual(1, len(skips))
1760
 
        test = skips[0]
1761
 
        self.assertFalse('log' in test.getDetails())
1762
 
 
1763
 
    def test_missing_feature_has_no_log(self):
1764
 
        # testtools doesn't know about addNotSupported, so it just gets
1765
 
        # considered as a skip
1766
 
        result = self._run_test('test_missing_feature')
1767
 
        self.assertEqual([missing_feature], result.skip_reasons.keys())
1768
 
        skips = result.skip_reasons[missing_feature]
1769
 
        self.assertEqual(1, len(skips))
1770
 
        test = skips[0]
1771
 
        self.assertFalse('log' in test.getDetails())
1772
 
 
1773
 
    def test_xfail_has_no_log(self):
1774
 
        result = self._run_test('test_xfail')
1775
 
        self.assertEqual(1, len(result.expectedFailures))
1776
 
        result_content = result.expectedFailures[0][1]
1777
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
1778
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
1779
 
 
1780
 
    def test_unexpected_success_has_log(self):
1781
 
        result = self._run_test('test_unexpected_success')
1782
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
1783
 
        # Inconsistency, unexpectedSuccesses is a list of tests,
1784
 
        # expectedFailures is a list of reasons?
1785
 
        test = result.unexpectedSuccesses[0]
1786
 
        details = test.getDetails()
1787
 
        self.assertTrue('log' in details)
1788
 
 
1789
 
 
1790
 
class TestTestCloning(tests.TestCase):
1791
 
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
1792
 
 
1793
 
    def test_cloned_testcase_does_not_share_details(self):
1794
 
        """A TestCase cloned with clone_test does not share mutable attributes
1795
 
        such as details or cleanups.
1796
 
        """
1797
 
        class Test(tests.TestCase):
1798
 
            def test_foo(self):
1799
 
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
1800
 
        orig_test = Test('test_foo')
1801
 
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
1802
 
        orig_test.run(unittest.TestResult())
1803
 
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
1804
 
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
1805
 
 
1806
 
    def test_double_apply_scenario_preserves_first_scenario(self):
1807
 
        """Applying two levels of scenarios to a test preserves the attributes
1808
 
        added by both scenarios.
1809
 
        """
1810
 
        class Test(tests.TestCase):
1811
 
            def test_foo(self):
1812
 
                pass
1813
 
        test = Test('test_foo')
1814
 
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
1815
 
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
1816
 
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
1817
 
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
1818
 
        all_tests = list(tests.iter_suite_tests(suite))
1819
 
        self.assertLength(4, all_tests)
1820
 
        all_xys = sorted((t.x, t.y) for t in all_tests)
1821
 
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
1822
 
 
1823
 
 
1824
1659
# NB: Don't delete this; it's not actually from 0.11!
1825
1660
@deprecated_function(deprecated_in((0, 11, 0)))
1826
1661
def sample_deprecated_function():
1953
1788
    def test_make_branch_and_tree_with_format(self):
1954
1789
        # we should be able to supply a format to make_branch_and_tree
1955
1790
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
 
1791
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
1956
1792
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
1957
1793
                              bzrlib.bzrdir.BzrDirMetaFormat1)
 
1794
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
 
1795
                              bzrlib.bzrdir.BzrDirFormat6)
1958
1796
 
1959
1797
    def test_make_branch_and_memory_tree(self):
1960
1798
        # we should be able to get a new branch and a mutable tree from
1977
1815
                tree.branch.repository.bzrdir.root_transport)
1978
1816
 
1979
1817
 
1980
 
class SelfTestHelper(object):
 
1818
class SelfTestHelper:
1981
1819
 
1982
1820
    def run_selftest(self, **kwargs):
1983
1821
        """Run selftest returning its output."""
2038
1876
 
2039
1877
    def test_lsprof_tests(self):
2040
1878
        self.requireFeature(test_lsprof.LSProfFeature)
2041
 
        results = []
 
1879
        calls = []
2042
1880
        class Test(object):
2043
1881
            def __call__(test, result):
2044
1882
                test.run(result)
2045
1883
            def run(test, result):
2046
 
                results.append(result)
 
1884
                self.assertIsInstance(result, tests.ForwardingResult)
 
1885
                calls.append("called")
2047
1886
            def countTestCases(self):
2048
1887
                return 1
2049
1888
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
2050
 
        self.assertLength(1, results)
2051
 
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
 
1889
        self.assertLength(1, calls)
2052
1890
 
2053
1891
    def test_random(self):
2054
1892
        # test randomising by listing a number of tests.
2109
1947
 
2110
1948
    def test_transport_sftp(self):
2111
1949
        self.requireFeature(features.paramiko)
2112
 
        from bzrlib.tests import stub_sftp
2113
1950
        self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
2114
1951
 
2115
1952
    def test_transport_memory(self):
2134
1971
            load_list='missing file name', list_only=True)
2135
1972
 
2136
1973
 
2137
 
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
2138
 
 
2139
 
    _test_needs_features = [features.subunit]
2140
 
 
2141
 
    def run_subunit_stream(self, test_name):
2142
 
        from subunit import ProtocolTestCase
2143
 
        def factory():
2144
 
            return TestUtil.TestSuite([_get_test(test_name)])
2145
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
2146
 
            test_suite_factory=factory)
2147
 
        test = ProtocolTestCase(stream)
2148
 
        result = testtools.TestResult()
2149
 
        test.run(result)
2150
 
        content = stream.getvalue()
2151
 
        return content, result
2152
 
 
2153
 
    def test_fail_has_log(self):
2154
 
        content, result = self.run_subunit_stream('test_fail')
2155
 
        self.assertEqual(1, len(result.failures))
2156
 
        self.assertContainsRe(content, '(?m)^log$')
2157
 
        self.assertContainsRe(content, 'this test will fail')
2158
 
 
2159
 
    def test_error_has_log(self):
2160
 
        content, result = self.run_subunit_stream('test_error')
2161
 
        self.assertContainsRe(content, '(?m)^log$')
2162
 
        self.assertContainsRe(content, 'this test errored')
2163
 
 
2164
 
    def test_skip_has_no_log(self):
2165
 
        content, result = self.run_subunit_stream('test_skip')
2166
 
        self.assertNotContainsRe(content, '(?m)^log$')
2167
 
        self.assertNotContainsRe(content, 'this test will be skipped')
2168
 
        self.assertEqual(['reason'], result.skip_reasons.keys())
2169
 
        skips = result.skip_reasons['reason']
2170
 
        self.assertEqual(1, len(skips))
2171
 
        test = skips[0]
2172
 
        # RemotedTestCase doesn't preserve the "details"
2173
 
        ## self.assertFalse('log' in test.getDetails())
2174
 
 
2175
 
    def test_missing_feature_has_no_log(self):
2176
 
        content, result = self.run_subunit_stream('test_missing_feature')
2177
 
        self.assertNotContainsRe(content, '(?m)^log$')
2178
 
        self.assertNotContainsRe(content, 'missing the feature')
2179
 
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
2180
 
        skips = result.skip_reasons['_MissingFeature\n']
2181
 
        self.assertEqual(1, len(skips))
2182
 
        test = skips[0]
2183
 
        # RemotedTestCase doesn't preserve the "details"
2184
 
        ## self.assertFalse('log' in test.getDetails())
2185
 
 
2186
 
    def test_xfail_has_no_log(self):
2187
 
        content, result = self.run_subunit_stream('test_xfail')
2188
 
        self.assertNotContainsRe(content, '(?m)^log$')
2189
 
        self.assertNotContainsRe(content, 'test with expected failure')
2190
 
        self.assertEqual(1, len(result.expectedFailures))
2191
 
        result_content = result.expectedFailures[0][1]
2192
 
        self.assertNotContainsRe(result_content, 'Text attachment: log')
2193
 
        self.assertNotContainsRe(result_content, 'test with expected failure')
2194
 
 
2195
 
    def test_unexpected_success_has_log(self):
2196
 
        content, result = self.run_subunit_stream('test_unexpected_success')
2197
 
        self.assertContainsRe(content, '(?m)^log$')
2198
 
        self.assertContainsRe(content, 'test with unexpected success')
2199
 
        # GZ 2011-05-18: Old versions of subunit treat unexpected success as a
2200
 
        #                success, if a min version check is added remove this
2201
 
        from subunit import TestProtocolClient as _Client
2202
 
        if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
2203
 
            self.expectFailure('subunit treats "unexpectedSuccess"'
2204
 
                               ' as a plain success',
2205
 
                self.assertEqual, 1, len(result.unexpectedSuccesses))
2206
 
        self.assertEqual(1, len(result.unexpectedSuccesses))
2207
 
        test = result.unexpectedSuccesses[0]
2208
 
        # RemotedTestCase doesn't preserve the "details"
2209
 
        ## self.assertTrue('log' in test.getDetails())
2210
 
 
2211
 
    def test_success_has_no_log(self):
2212
 
        content, result = self.run_subunit_stream('test_success')
2213
 
        self.assertEqual(1, result.testsRun)
2214
 
        self.assertNotContainsRe(content, '(?m)^log$')
2215
 
        self.assertNotContainsRe(content, 'this test succeeds')
2216
 
 
2217
 
 
2218
1974
class TestRunBzr(tests.TestCase):
2219
1975
 
2220
1976
    out = ''
2343
2099
        # stdout and stderr of the invoked run_bzr
2344
2100
        current_factory = bzrlib.ui.ui_factory
2345
2101
        self.run_bzr(['foo'])
2346
 
        self.assertFalse(current_factory is self.factory)
 
2102
        self.failIf(current_factory is self.factory)
2347
2103
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2348
2104
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2349
2105
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2506
2262
 
2507
2263
 
2508
2264
class TestStartBzrSubProcess(tests.TestCase):
2509
 
    """Stub test start_bzr_subprocess."""
2510
2265
 
2511
 
    def _subprocess_log_cleanup(self):
2512
 
        """Inhibits the base version as we don't produce a log file."""
 
2266
    def check_popen_state(self):
 
2267
        """Replace to make assertions when popen is called."""
2513
2268
 
2514
2269
    def _popen(self, *args, **kwargs):
2515
 
        """Override the base version to record the command that is run.
2516
 
 
2517
 
        From there we can ensure it is correct without spawning a real process.
2518
 
        """
 
2270
        """Record the command that is run, so that we can ensure it is correct"""
2519
2271
        self.check_popen_state()
2520
2272
        self._popen_args = args
2521
2273
        self._popen_kwargs = kwargs
2522
2274
        raise _DontSpawnProcess()
2523
2275
 
2524
 
    def check_popen_state(self):
2525
 
        """Replace to make assertions when popen is called."""
2526
 
 
2527
2276
    def test_run_bzr_subprocess_no_plugins(self):
2528
2277
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2529
2278
        command = self._popen_args[0]
2533
2282
 
2534
2283
    def test_allow_plugins(self):
2535
2284
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2536
 
                          allow_plugins=True)
 
2285
            allow_plugins=True)
2537
2286
        command = self._popen_args[0]
2538
2287
        self.assertEqual([], command[2:])
2539
2288
 
2540
2289
    def test_set_env(self):
2541
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2290
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2542
2291
        # set in the child
2543
2292
        def check_environment():
2544
2293
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2545
2294
        self.check_popen_state = check_environment
2546
2295
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2547
 
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
 
2296
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
2548
2297
        # not set in theparent
2549
2298
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2550
2299
 
2551
2300
    def test_run_bzr_subprocess_env_del(self):
2552
2301
        """run_bzr_subprocess can remove environment variables too."""
2553
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2302
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2554
2303
        def check_environment():
2555
2304
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2556
2305
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2557
2306
        self.check_popen_state = check_environment
2558
2307
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2559
 
                          env_changes={'EXISTANT_ENV_VAR':None})
 
2308
            env_changes={'EXISTANT_ENV_VAR':None})
2560
2309
        # Still set in parent
2561
2310
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2562
2311
        del os.environ['EXISTANT_ENV_VAR']
2563
2312
 
2564
2313
    def test_env_del_missing(self):
2565
 
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
 
2314
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2566
2315
        def check_environment():
2567
2316
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2568
2317
        self.check_popen_state = check_environment
2569
2318
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2570
 
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
 
2319
            env_changes={'NON_EXISTANT_ENV_VAR':None})
2571
2320
 
2572
2321
    def test_working_dir(self):
2573
2322
        """Test that we can specify the working dir for the child"""
2576
2325
        chdirs = []
2577
2326
        def chdir(path):
2578
2327
            chdirs.append(path)
2579
 
        self.overrideAttr(os, 'chdir', chdir)
2580
 
        def getcwd():
2581
 
            return 'current'
2582
 
        self.overrideAttr(osutils, 'getcwd', getcwd)
2583
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2584
 
                          working_dir='foo')
 
2328
        os.chdir = chdir
 
2329
        try:
 
2330
            def getcwd():
 
2331
                return 'current'
 
2332
            osutils.getcwd = getcwd
 
2333
            try:
 
2334
                self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2335
                    working_dir='foo')
 
2336
            finally:
 
2337
                osutils.getcwd = orig_getcwd
 
2338
        finally:
 
2339
            os.chdir = orig_chdir
2585
2340
        self.assertEqual(['foo', 'current'], chdirs)
2586
2341
 
2587
 
    def test_get_bzr_path_with_cwd_bzrlib(self):
2588
 
        self.get_source_path = lambda: ""
2589
 
        self.overrideAttr(os.path, "isfile", lambda path: True)
2590
 
        self.assertEqual(self.get_bzr_path(), "bzr")
2591
 
 
2592
2342
 
2593
2343
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2594
2344
    """Tests that really need to do things with an external bzr."""
2837
2587
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2838
2588
 
2839
2589
 
2840
 
class TestCheckTreeShape(tests.TestCaseWithTransport):
 
2590
class TestCheckInventoryShape(tests.TestCaseWithTransport):
2841
2591
 
2842
 
    def test_check_tree_shape(self):
 
2592
    def test_check_inventory_shape(self):
2843
2593
        files = ['a', 'b/', 'b/c']
2844
2594
        tree = self.make_branch_and_tree('.')
2845
2595
        self.build_tree(files)
2846
2596
        tree.add(files)
2847
2597
        tree.lock_read()
2848
2598
        try:
2849
 
            self.check_tree_shape(tree, files)
 
2599
            self.check_inventory_shape(tree.inventory, files)
2850
2600
        finally:
2851
2601
            tree.unlock()
2852
2602
 
3023
2773
        # Test that a plausible list of modules to doctest is returned
3024
2774
        # by _test_suite_modules_to_doctest.
3025
2775
        test_list = tests._test_suite_modules_to_doctest()
3026
 
        if __doc__ is None:
3027
 
            # When docstrings are stripped, there are no modules to doctest
3028
 
            self.assertEqual([], test_list)
3029
 
            return
3030
2776
        self.assertSubset([
3031
2777
            'bzrlib.timestamp',
3032
2778
            ],
3049
2795
        self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
3050
2796
        def doctests():
3051
2797
            calls.append("modules_to_doctest")
3052
 
            if __doc__ is None:
3053
 
                return []
3054
2798
            return ['bzrlib.timestamp']
3055
2799
        self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
3056
2800
        expected_test_list = [
3059
2803
            ('bzrlib.tests.per_transport.TransportTests'
3060
2804
             '.test_abspath(LocalTransport,LocalURLServer)'),
3061
2805
            'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
 
2806
            # modules_to_doctest
 
2807
            'bzrlib.timestamp.format_highres_date',
3062
2808
            # plugins can't be tested that way since selftest may be run with
3063
2809
            # --no-plugins
3064
2810
            ]
3065
 
        if __doc__ is not None:
3066
 
            expected_test_list.extend([
3067
 
                # modules_to_doctest
3068
 
                'bzrlib.timestamp.format_highres_date',
3069
 
                ])
3070
2811
        suite = tests.test_suite()
3071
2812
        self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
3072
2813
            set(calls))
3184
2925
        tpr.register('bar', 'bBB.aAA.rRR')
3185
2926
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
3186
2927
        self.assertThat(self.get_log(),
3187
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
3188
 
                           doctest.ELLIPSIS))
 
2928
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
3189
2929
 
3190
2930
    def test_get_unknown_prefix(self):
3191
2931
        tpr = self._get_registry()
3211
2951
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
3212
2952
 
3213
2953
 
3214
 
class TestThreadLeakDetection(tests.TestCase):
3215
 
    """Ensure when tests leak threads we detect and report it"""
3216
 
 
3217
 
    class LeakRecordingResult(tests.ExtendedTestResult):
3218
 
        def __init__(self):
3219
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3220
 
            self.leaks = []
3221
 
        def _report_thread_leak(self, test, leaks, alive):
3222
 
            self.leaks.append((test, leaks))
3223
 
 
3224
 
    def test_testcase_without_addCleanups(self):
3225
 
        """Check old TestCase instances don't break with leak detection"""
3226
 
        class Test(unittest.TestCase):
3227
 
            def runTest(self):
3228
 
                pass
3229
 
        result = self.LeakRecordingResult()
3230
 
        test = Test()
3231
 
        result.startTestRun()
3232
 
        test.run(result)
3233
 
        result.stopTestRun()
3234
 
        self.assertEqual(result._tests_leaking_threads_count, 0)
3235
 
        self.assertEqual(result.leaks, [])
3236
 
        
3237
 
    def test_thread_leak(self):
3238
 
        """Ensure a thread that outlives the running of a test is reported
3239
 
 
3240
 
        Uses a thread that blocks on an event, and is started by the inner
3241
 
        test case. As the thread outlives the inner case's run, it should be
3242
 
        detected as a leak, but the event is then set so that the thread can
3243
 
        be safely joined in cleanup so it's not leaked for real.
3244
 
        """
3245
 
        event = threading.Event()
3246
 
        thread = threading.Thread(name="Leaker", target=event.wait)
3247
 
        class Test(tests.TestCase):
3248
 
            def test_leak(self):
3249
 
                thread.start()
3250
 
        result = self.LeakRecordingResult()
3251
 
        test = Test("test_leak")
3252
 
        self.addCleanup(thread.join)
3253
 
        self.addCleanup(event.set)
3254
 
        result.startTestRun()
3255
 
        test.run(result)
3256
 
        result.stopTestRun()
3257
 
        self.assertEqual(result._tests_leaking_threads_count, 1)
3258
 
        self.assertEqual(result._first_thread_leaker_id, test.id())
3259
 
        self.assertEqual(result.leaks, [(test, set([thread]))])
3260
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3261
 
 
3262
 
    def test_multiple_leaks(self):
3263
 
        """Check multiple leaks are blamed on the test cases at fault
3264
 
 
3265
 
        Same concept as the previous test, but has one inner test method that
3266
 
        leaks two threads, and one that doesn't leak at all.
3267
 
        """
3268
 
        event = threading.Event()
3269
 
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
3270
 
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
3271
 
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
3272
 
        class Test(tests.TestCase):
3273
 
            def test_first_leak(self):
3274
 
                thread_b.start()
3275
 
            def test_second_no_leak(self):
3276
 
                pass
3277
 
            def test_third_leak(self):
3278
 
                thread_c.start()
3279
 
                thread_a.start()
3280
 
        result = self.LeakRecordingResult()
3281
 
        first_test = Test("test_first_leak")
3282
 
        third_test = Test("test_third_leak")
3283
 
        self.addCleanup(thread_a.join)
3284
 
        self.addCleanup(thread_b.join)
3285
 
        self.addCleanup(thread_c.join)
3286
 
        self.addCleanup(event.set)
3287
 
        result.startTestRun()
3288
 
        unittest.TestSuite(
3289
 
            [first_test, Test("test_second_no_leak"), third_test]
3290
 
            ).run(result)
3291
 
        result.stopTestRun()
3292
 
        self.assertEqual(result._tests_leaking_threads_count, 2)
3293
 
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
3294
 
        self.assertEqual(result.leaks, [
3295
 
            (first_test, set([thread_b])),
3296
 
            (third_test, set([thread_a, thread_c]))])
3297
 
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
3298
 
 
3299
 
 
3300
 
class TestPostMortemDebugging(tests.TestCase):
3301
 
    """Check post mortem debugging works when tests fail or error"""
3302
 
 
3303
 
    class TracebackRecordingResult(tests.ExtendedTestResult):
3304
 
        def __init__(self):
3305
 
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
3306
 
            self.postcode = None
3307
 
        def _post_mortem(self, tb=None):
3308
 
            """Record the code object at the end of the current traceback"""
3309
 
            tb = tb or sys.exc_info()[2]
3310
 
            if tb is not None:
3311
 
                next = tb.tb_next
3312
 
                while next is not None:
3313
 
                    tb = next
3314
 
                    next = next.tb_next
3315
 
                self.postcode = tb.tb_frame.f_code
3316
 
        def report_error(self, test, err):
3317
 
            pass
3318
 
        def report_failure(self, test, err):
3319
 
            pass
3320
 
 
3321
 
    def test_location_unittest_error(self):
3322
 
        """Needs right post mortem traceback with erroring unittest case"""
3323
 
        class Test(unittest.TestCase):
3324
 
            def runTest(self):
3325
 
                raise RuntimeError
3326
 
        result = self.TracebackRecordingResult()
3327
 
        Test().run(result)
3328
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3329
 
 
3330
 
    def test_location_unittest_failure(self):
3331
 
        """Needs right post mortem traceback with failing unittest case"""
3332
 
        class Test(unittest.TestCase):
3333
 
            def runTest(self):
3334
 
                raise self.failureException
3335
 
        result = self.TracebackRecordingResult()
3336
 
        Test().run(result)
3337
 
        self.assertEqual(result.postcode, Test.runTest.func_code)
3338
 
 
3339
 
    def test_location_bt_error(self):
3340
 
        """Needs right post mortem traceback with erroring bzrlib.tests case"""
3341
 
        class Test(tests.TestCase):
3342
 
            def test_error(self):
3343
 
                raise RuntimeError
3344
 
        result = self.TracebackRecordingResult()
3345
 
        Test("test_error").run(result)
3346
 
        self.assertEqual(result.postcode, Test.test_error.func_code)
3347
 
 
3348
 
    def test_location_bt_failure(self):
3349
 
        """Needs right post mortem traceback with failing bzrlib.tests case"""
3350
 
        class Test(tests.TestCase):
3351
 
            def test_failure(self):
3352
 
                raise self.failureException
3353
 
        result = self.TracebackRecordingResult()
3354
 
        Test("test_failure").run(result)
3355
 
        self.assertEqual(result.postcode, Test.test_failure.func_code)
3356
 
 
3357
 
    def test_env_var_triggers_post_mortem(self):
3358
 
        """Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
3359
 
        import pdb
3360
 
        result = tests.ExtendedTestResult(StringIO(), 0, 1)
3361
 
        post_mortem_calls = []
3362
 
        self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
3363
 
        self.overrideEnv('BZR_TEST_PDB', None)
3364
 
        result._post_mortem(1)
3365
 
        self.overrideEnv('BZR_TEST_PDB', 'on')
3366
 
        result._post_mortem(2)
3367
 
        self.assertEqual([2], post_mortem_calls)
3368
 
 
3369
 
 
3370
2954
class TestRunSuite(tests.TestCase):
3371
2955
 
3372
2956
    def test_runner_class(self):
3383
2967
                                                self.verbosity)
3384
2968
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
3385
2969
        self.assertLength(1, calls)
3386
 
 
3387
 
 
3388
 
class TestEnvironHandling(tests.TestCase):
3389
 
 
3390
 
    def test_overrideEnv_None_called_twice_doesnt_leak(self):
3391
 
        self.assertFalse('MYVAR' in os.environ)
3392
 
        self.overrideEnv('MYVAR', '42')
3393
 
        # We use an embedded test to make sure we fix the _captureVar bug
3394
 
        class Test(tests.TestCase):
3395
 
            def test_me(self):
3396
 
                # The first call save the 42 value
3397
 
                self.overrideEnv('MYVAR', None)
3398
 
                self.assertEquals(None, os.environ.get('MYVAR'))
3399
 
                # Make sure we can call it twice
3400
 
                self.overrideEnv('MYVAR', None)
3401
 
                self.assertEquals(None, os.environ.get('MYVAR'))
3402
 
        output = StringIO()
3403
 
        result = tests.TextTestResult(output, 0, 1)
3404
 
        Test('test_me').run(result)
3405
 
        if not result.wasStrictlySuccessful():
3406
 
            self.fail(output.getvalue())
3407
 
        # We get our value back
3408
 
        self.assertEquals('42', os.environ.get('MYVAR'))
3409
 
 
3410
 
 
3411
 
class TestIsolatedEnv(tests.TestCase):
3412
 
    """Test isolating tests from os.environ.
3413
 
 
3414
 
    Since we use tests that are already isolated from os.environ a bit of care
3415
 
    should be taken when designing the tests to avoid bootstrap side-effects.
3416
 
    The tests start an already clean os.environ which allow doing valid
3417
 
    assertions about which variables are present or not and design tests around
3418
 
    these assertions.
3419
 
    """
3420
 
 
3421
 
    class ScratchMonkey(tests.TestCase):
3422
 
 
3423
 
        def test_me(self):
3424
 
            pass
3425
 
 
3426
 
    def test_basics(self):
3427
 
        # Make sure we know the definition of BZR_HOME: not part of os.environ
3428
 
        # for tests.TestCase.
3429
 
        self.assertTrue('BZR_HOME' in tests.isolated_environ)
3430
 
        self.assertEquals(None, tests.isolated_environ['BZR_HOME'])
3431
 
        # Being part of isolated_environ, BZR_HOME should not appear here
3432
 
        self.assertFalse('BZR_HOME' in os.environ)
3433
 
        # Make sure we know the definition of LINES: part of os.environ for
3434
 
        # tests.TestCase
3435
 
        self.assertTrue('LINES' in tests.isolated_environ)
3436
 
        self.assertEquals('25', tests.isolated_environ['LINES'])
3437
 
        self.assertEquals('25', os.environ['LINES'])
3438
 
 
3439
 
    def test_injecting_unknown_variable(self):
3440
 
        # BZR_HOME is known to be absent from os.environ
3441
 
        test = self.ScratchMonkey('test_me')
3442
 
        tests.override_os_environ(test, {'BZR_HOME': 'foo'})
3443
 
        self.assertEquals('foo', os.environ['BZR_HOME'])
3444
 
        tests.restore_os_environ(test)
3445
 
        self.assertFalse('BZR_HOME' in os.environ)
3446
 
 
3447
 
    def test_injecting_known_variable(self):
3448
 
        test = self.ScratchMonkey('test_me')
3449
 
        # LINES is known to be present in os.environ
3450
 
        tests.override_os_environ(test, {'LINES': '42'})
3451
 
        self.assertEquals('42', os.environ['LINES'])
3452
 
        tests.restore_os_environ(test)
3453
 
        self.assertEquals('25', os.environ['LINES'])
3454
 
 
3455
 
    def test_deleting_variable(self):
3456
 
        test = self.ScratchMonkey('test_me')
3457
 
        # LINES is known to be present in os.environ
3458
 
        tests.override_os_environ(test, {'LINES': None})
3459
 
        self.assertTrue('LINES' not in os.environ)
3460
 
        tests.restore_os_environ(test)
3461
 
        self.assertEquals('25', os.environ['LINES'])
3462
 
 
3463
 
 
3464
 
class TestDocTestSuiteIsolation(tests.TestCase):
3465
 
    """Test that `tests.DocTestSuite` isolates doc tests from os.environ.
3466
 
 
3467
 
    Since tests.TestCase alreay provides an isolation from os.environ, we use
3468
 
    the clean environment as a base for testing. To precisely capture the
3469
 
    isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
3470
 
    compare against.
3471
 
 
3472
 
    We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
3473
 
    not `os.environ` so each test overrides it to suit its needs.
3474
 
 
3475
 
    """
3476
 
 
3477
 
    def get_doctest_suite_for_string(self, klass, string):
3478
 
        class Finder(doctest.DocTestFinder):
3479
 
 
3480
 
            def find(*args, **kwargs):
3481
 
                test = doctest.DocTestParser().get_doctest(
3482
 
                    string, {}, 'foo', 'foo.py', 0)
3483
 
                return [test]
3484
 
 
3485
 
        suite = klass(test_finder=Finder())
3486
 
        return suite
3487
 
 
3488
 
    def run_doctest_suite_for_string(self, klass, string):
3489
 
        suite = self.get_doctest_suite_for_string(klass, string)
3490
 
        output = StringIO()
3491
 
        result = tests.TextTestResult(output, 0, 1)
3492
 
        suite.run(result)
3493
 
        return result, output
3494
 
 
3495
 
    def assertDocTestStringSucceds(self, klass, string):
3496
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3497
 
        if not result.wasStrictlySuccessful():
3498
 
            self.fail(output.getvalue())
3499
 
 
3500
 
    def assertDocTestStringFails(self, klass, string):
3501
 
        result, output = self.run_doctest_suite_for_string(klass, string)
3502
 
        if result.wasStrictlySuccessful():
3503
 
            self.fail(output.getvalue())
3504
 
 
3505
 
    def test_injected_variable(self):
3506
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
3507
 
        test = """
3508
 
            >>> import os
3509
 
            >>> os.environ['LINES']
3510
 
            '42'
3511
 
            """
3512
 
        # doctest.DocTestSuite fails as it sees '25'
3513
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3514
 
        # tests.DocTestSuite sees '42'
3515
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3516
 
 
3517
 
    def test_deleted_variable(self):
3518
 
        self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
3519
 
        test = """
3520
 
            >>> import os
3521
 
            >>> os.environ.get('LINES')
3522
 
            """
3523
 
        # doctest.DocTestSuite fails as it sees '25'
3524
 
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
3525
 
        # tests.DocTestSuite sees None
3526
 
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
3527
 
 
3528
 
 
3529
 
class TestSelftestExcludePatterns(tests.TestCase):
3530
 
 
3531
 
    def setUp(self):
3532
 
        super(TestSelftestExcludePatterns, self).setUp()
3533
 
        self.overrideAttr(tests, 'test_suite', self.suite_factory)
3534
 
 
3535
 
    def suite_factory(self, keep_only=None, starting_with=None):
3536
 
        """A test suite factory with only a few tests."""
3537
 
        class Test(tests.TestCase):
3538
 
            def id(self):
3539
 
                # We don't need the full class path
3540
 
                return self._testMethodName
3541
 
            def a(self):
3542
 
                pass
3543
 
            def b(self):
3544
 
                pass
3545
 
            def c(self):
3546
 
                pass
3547
 
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
3548
 
 
3549
 
    def assertTestList(self, expected, *selftest_args):
3550
 
        # We rely on setUp installing the right test suite factory so we can
3551
 
        # test at the command level without loading the whole test suite
3552
 
        out, err = self.run_bzr(('selftest', '--list') + selftest_args)
3553
 
        actual = out.splitlines()
3554
 
        self.assertEquals(expected, actual)
3555
 
 
3556
 
    def test_full_list(self):
3557
 
        self.assertTestList(['a', 'b', 'c'])
3558
 
 
3559
 
    def test_single_exclude(self):
3560
 
        self.assertTestList(['b', 'c'], '-x', 'a')
3561
 
 
3562
 
    def test_mutiple_excludes(self):
3563
 
        self.assertTestList(['c'], '-x', 'a', '-x', 'b')
3564
 
 
3565
 
 
3566
 
class TestCounterHooks(tests.TestCase, SelfTestHelper):
3567
 
 
3568
 
    _test_needs_features = [features.subunit]
3569
 
 
3570
 
    def setUp(self):
3571
 
        super(TestCounterHooks, self).setUp()
3572
 
        class Test(tests.TestCase):
3573
 
 
3574
 
            def setUp(self):
3575
 
                super(Test, self).setUp()
3576
 
                self.hooks = hooks.Hooks()
3577
 
                self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
3578
 
                self.install_counter_hook(self.hooks, 'myhook')
3579
 
 
3580
 
            def no_hook(self):
3581
 
                pass
3582
 
 
3583
 
            def run_hook_once(self):
3584
 
                for hook in self.hooks['myhook']:
3585
 
                    hook(self)
3586
 
 
3587
 
        self.test_class = Test
3588
 
 
3589
 
    def assertHookCalls(self, expected_calls, test_name):
3590
 
        test = self.test_class(test_name)
3591
 
        result = unittest.TestResult()
3592
 
        test.run(result)
3593
 
        self.assertTrue(hasattr(test, '_counters'))
3594
 
        self.assertTrue(test._counters.has_key('myhook'))
3595
 
        self.assertEquals(expected_calls, test._counters['myhook'])
3596
 
 
3597
 
    def test_no_hook(self):
3598
 
        self.assertHookCalls(0, 'no_hook')
3599
 
 
3600
 
    def test_run_hook_once(self):
3601
 
        tt = features.testtools
3602
 
        if tt.module.__version__ < (0, 9, 8):
3603
 
            raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
3604
 
        self.assertHookCalls(1, 'run_hook_once')