~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Martin
  • Date: 2011-08-04 00:17:53 UTC
  • mto: This revision was merged to the branch mainline in revision 6055.
  • Revision ID: gzlist@googlemail.com-20110804001753-plgpwcpsxcum16yb
Make tests raising KnownFailure use the knownFailure method instead

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005-2011 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
 
from doctest import ELLIPSIS
 
20
import doctest
21
21
import os
22
22
import signal
23
23
import sys
 
24
import threading
24
25
import time
25
26
import unittest
26
27
import warnings
27
28
 
28
 
from testtools import MultiTestResult
 
29
from testtools import (
 
30
    ExtendedToOriginalDecorator,
 
31
    MultiTestResult,
 
32
    __version__ as testtools_version,
 
33
    )
 
34
from testtools.content import Content
29
35
from testtools.content_type import ContentType
30
36
from testtools.matchers import (
31
37
    DocTestMatches,
32
38
    Equals,
33
39
    )
34
 
import testtools.tests.helpers
 
40
import testtools.testresult.doubles
35
41
 
36
42
import bzrlib
37
43
from bzrlib import (
38
44
    branchbuilder,
39
45
    bzrdir,
40
 
    debug,
41
46
    errors,
 
47
    hooks,
42
48
    lockdir,
43
49
    memorytree,
44
50
    osutils,
45
 
    progress,
46
51
    remote,
47
52
    repository,
48
53
    symbol_versioning,
49
54
    tests,
50
55
    transport,
51
56
    workingtree,
 
57
    workingtree_3,
 
58
    workingtree_4,
52
59
    )
53
60
from bzrlib.repofmt import (
54
61
    groupcompress_repo,
55
 
    pack_repo,
56
 
    weaverepo,
57
62
    )
58
63
from bzrlib.symbol_versioning import (
59
64
    deprecated_function,
64
69
    features,
65
70
    test_lsprof,
66
71
    test_server,
67
 
    test_sftp_transport,
68
72
    TestUtil,
69
73
    )
70
 
from bzrlib.trace import note
 
74
from bzrlib.trace import note, mutter
71
75
from bzrlib.transport import memory
72
 
from bzrlib.version import _get_bzr_source_tree
73
76
 
74
77
 
75
78
def _test_ids(test_suite):
77
80
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
78
81
 
79
82
 
80
 
class SelftestTests(tests.TestCase):
81
 
 
82
 
    def test_import_tests(self):
83
 
        mod = TestUtil._load_module_by_name('bzrlib.tests.test_selftest')
84
 
        self.assertEqual(mod.SelftestTests, SelftestTests)
85
 
 
86
 
    def test_import_test_failure(self):
87
 
        self.assertRaises(ImportError,
88
 
                          TestUtil._load_module_by_name,
89
 
                          'bzrlib.no-name-yet')
90
 
 
91
 
 
92
83
class MetaTestLog(tests.TestCase):
93
84
 
94
85
    def test_logging(self):
100
91
            "text", "plain", {"charset": "utf8"})))
101
92
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
102
93
        self.assertThat(self.get_log(),
103
 
            DocTestMatches(u"...a test message\n", ELLIPSIS))
104
 
 
105
 
 
106
 
class TestUnicodeFilename(tests.TestCase):
107
 
 
108
 
    def test_probe_passes(self):
109
 
        """UnicodeFilename._probe passes."""
110
 
        # We can't test much more than that because the behaviour depends
111
 
        # on the platform.
112
 
        tests.UnicodeFilename._probe()
 
94
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
113
95
 
114
96
 
115
97
class TestTreeShape(tests.TestCaseInTempDir):
116
98
 
117
99
    def test_unicode_paths(self):
118
 
        self.requireFeature(tests.UnicodeFilename)
 
100
        self.requireFeature(features.UnicodeFilenameFeature)
119
101
 
120
102
        filename = u'hell\u00d8'
121
103
        self.build_tree_contents([(filename, 'contents of hello')])
122
 
        self.failUnlessExists(filename)
 
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
123
118
 
124
119
 
125
120
class TestTransportScenarios(tests.TestCase):
208
203
    def test_scenarios(self):
209
204
        # check that constructor parameters are passed through to the adapted
210
205
        # test.
211
 
        from bzrlib.tests.per_bzrdir import make_scenarios
 
206
        from bzrlib.tests.per_controldir import make_scenarios
212
207
        vfs_factory = "v"
213
208
        server1 = "a"
214
209
        server2 = "b"
312
307
        from bzrlib.tests.per_interrepository import make_scenarios
313
308
        server1 = "a"
314
309
        server2 = "b"
315
 
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
 
310
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
316
311
        scenarios = make_scenarios(server1, server2, formats)
317
312
        self.assertEqual([
318
313
            ('C0,str,str',
319
314
             {'repository_format': 'C1',
320
315
              'repository_format_to': 'C2',
321
316
              'transport_readonly_server': 'b',
322
 
              'transport_server': 'a'}),
 
317
              'transport_server': 'a',
 
318
              'extra_setup': 'C3'}),
323
319
            ('D0,str,str',
324
320
             {'repository_format': 'D1',
325
321
              'repository_format_to': 'D2',
326
322
              'transport_readonly_server': 'b',
327
 
              'transport_server': 'a'})],
 
323
              'transport_server': 'a',
 
324
              'extra_setup': 'D3'})],
328
325
            scenarios)
329
326
 
330
327
 
336
333
        from bzrlib.tests.per_workingtree import make_scenarios
337
334
        server1 = "a"
338
335
        server2 = "b"
339
 
        formats = [workingtree.WorkingTreeFormat2(),
340
 
                   workingtree.WorkingTreeFormat3(),]
 
336
        formats = [workingtree_4.WorkingTreeFormat4(),
 
337
                   workingtree_3.WorkingTreeFormat3(),]
341
338
        scenarios = make_scenarios(server1, server2, formats)
342
339
        self.assertEqual([
343
 
            ('WorkingTreeFormat2',
 
340
            ('WorkingTreeFormat4',
344
341
             {'bzrdir_format': formats[0]._matchingbzrdir,
345
342
              'transport_readonly_server': 'b',
346
343
              'transport_server': 'a',
373
370
            )
374
371
        server1 = "a"
375
372
        server2 = "b"
376
 
        formats = [workingtree.WorkingTreeFormat2(),
377
 
                   workingtree.WorkingTreeFormat3(),]
 
373
        formats = [workingtree_4.WorkingTreeFormat4(),
 
374
                   workingtree_3.WorkingTreeFormat3(),]
378
375
        scenarios = make_scenarios(server1, server2, formats)
379
376
        self.assertEqual(7, len(scenarios))
380
 
        default_wt_format = workingtree.WorkingTreeFormat4._default_format
381
 
        wt4_format = workingtree.WorkingTreeFormat4()
382
 
        wt5_format = workingtree.WorkingTreeFormat5()
 
377
        default_wt_format = workingtree.format_registry.get_default()
 
378
        wt4_format = workingtree_4.WorkingTreeFormat4()
 
379
        wt5_format = workingtree_4.WorkingTreeFormat5()
383
380
        expected_scenarios = [
384
 
            ('WorkingTreeFormat2',
 
381
            ('WorkingTreeFormat4',
385
382
             {'bzrdir_format': formats[0]._matchingbzrdir,
386
383
              'transport_readonly_server': 'b',
387
384
              'transport_server': 'a',
447
444
        # ones to add.
448
445
        from bzrlib.tests.per_tree import (
449
446
            return_parameter,
450
 
            revision_tree_from_workingtree
451
447
            )
452
448
        from bzrlib.tests.per_intertree import (
453
449
            make_scenarios,
454
450
            )
455
 
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
 
451
        from bzrlib.workingtree_3 import WorkingTreeFormat3
 
452
        from bzrlib.workingtree_4 import WorkingTreeFormat4
456
453
        input_test = TestInterTreeScenarios(
457
454
            "test_scenarios")
458
455
        server1 = "a"
459
456
        server2 = "b"
460
 
        format1 = WorkingTreeFormat2()
 
457
        format1 = WorkingTreeFormat4()
461
458
        format2 = WorkingTreeFormat3()
462
459
        formats = [("1", str, format1, format2, "converter1"),
463
460
            ("2", int, format2, format1, "converter2")]
509
506
        self.assertRaises(AssertionError, self.assertEqualStat,
510
507
            os.lstat("foo"), os.lstat("longname"))
511
508
 
 
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
 
512
528
 
513
529
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
514
530
 
548
564
        tree = self.make_branch_and_memory_tree('dir')
549
565
        # Guard against regression into MemoryTransport leaking
550
566
        # files to disk instead of keeping them in memory.
551
 
        self.failIf(osutils.lexists('dir'))
 
567
        self.assertFalse(osutils.lexists('dir'))
552
568
        self.assertIsInstance(tree, memorytree.MemoryTree)
553
569
 
554
570
    def test_make_branch_and_memory_tree_with_format(self):
555
571
        """make_branch_and_memory_tree should accept a format option."""
556
572
        format = bzrdir.BzrDirMetaFormat1()
557
 
        format.repository_format = weaverepo.RepositoryFormat7()
 
573
        format.repository_format = repository.format_registry.get_default()
558
574
        tree = self.make_branch_and_memory_tree('dir', format=format)
559
575
        # Guard against regression into MemoryTransport leaking
560
576
        # files to disk instead of keeping them in memory.
561
 
        self.failIf(osutils.lexists('dir'))
 
577
        self.assertFalse(osutils.lexists('dir'))
562
578
        self.assertIsInstance(tree, memorytree.MemoryTree)
563
579
        self.assertEqual(format.repository_format.__class__,
564
580
            tree.branch.repository._format.__class__)
568
584
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
569
585
        # Guard against regression into MemoryTransport leaking
570
586
        # files to disk instead of keeping them in memory.
571
 
        self.failIf(osutils.lexists('dir'))
 
587
        self.assertFalse(osutils.lexists('dir'))
572
588
 
573
589
    def test_make_branch_builder_with_format(self):
574
590
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
575
591
        # that the format objects are used.
576
592
        format = bzrdir.BzrDirMetaFormat1()
577
 
        repo_format = weaverepo.RepositoryFormat7()
 
593
        repo_format = repository.format_registry.get_default()
578
594
        format.repository_format = repo_format
579
595
        builder = self.make_branch_builder('dir', format=format)
580
596
        the_branch = builder.get_branch()
581
597
        # Guard against regression into MemoryTransport leaking
582
598
        # files to disk instead of keeping them in memory.
583
 
        self.failIf(osutils.lexists('dir'))
 
599
        self.assertFalse(osutils.lexists('dir'))
584
600
        self.assertEqual(format.repository_format.__class__,
585
601
                         the_branch.repository._format.__class__)
586
602
        self.assertEqual(repo_format.get_format_string(),
592
608
        the_branch = builder.get_branch()
593
609
        # Guard against regression into MemoryTransport leaking
594
610
        # files to disk instead of keeping them in memory.
595
 
        self.failIf(osutils.lexists('dir'))
 
611
        self.assertFalse(osutils.lexists('dir'))
596
612
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
597
613
        self.assertEqual(dir_format.repository_format.__class__,
598
614
                         the_branch.repository._format.__class__)
609
625
                l.attempt_lock()
610
626
        test = TestDanglingLock('test_function')
611
627
        result = test.run()
 
628
        total_failures = result.errors + result.failures
612
629
        if self._lock_check_thorough:
613
 
            self.assertEqual(1, len(result.errors))
 
630
            self.assertEqual(1, len(total_failures))
614
631
        else:
615
632
            # When _lock_check_thorough is disabled, then we don't trigger a
616
633
            # failure
617
 
            self.assertEqual(0, len(result.errors))
 
634
            self.assertEqual(0, len(total_failures))
618
635
 
619
636
 
620
637
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
621
638
    """Tests for the convenience functions TestCaseWithTransport introduces."""
622
639
 
623
640
    def test_get_readonly_url_none(self):
624
 
        from bzrlib.transport import get_transport
625
641
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
626
642
        self.vfs_transport_factory = memory.MemoryServer
627
643
        self.transport_readonly_server = None
629
645
        # for the server
630
646
        url = self.get_readonly_url()
631
647
        url2 = self.get_readonly_url('foo/bar')
632
 
        t = get_transport(url)
633
 
        t2 = get_transport(url2)
634
 
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
635
 
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
 
648
        t = transport.get_transport(url)
 
649
        t2 = transport.get_transport(url2)
 
650
        self.assertIsInstance(t, ReadonlyTransportDecorator)
 
651
        self.assertIsInstance(t2, ReadonlyTransportDecorator)
636
652
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
637
653
 
638
654
    def test_get_readonly_url_http(self):
639
655
        from bzrlib.tests.http_server import HttpServer
640
 
        from bzrlib.transport import get_transport
641
656
        from bzrlib.transport.http import HttpTransportBase
642
657
        self.transport_server = test_server.LocalURLServer
643
658
        self.transport_readonly_server = HttpServer
645
660
        url = self.get_readonly_url()
646
661
        url2 = self.get_readonly_url('foo/bar')
647
662
        # the transport returned may be any HttpTransportBase subclass
648
 
        t = get_transport(url)
649
 
        t2 = get_transport(url2)
650
 
        self.failUnless(isinstance(t, HttpTransportBase))
651
 
        self.failUnless(isinstance(t2, HttpTransportBase))
 
663
        t = transport.get_transport(url)
 
664
        t2 = transport.get_transport(url2)
 
665
        self.assertIsInstance(t, HttpTransportBase)
 
666
        self.assertIsInstance(t2, HttpTransportBase)
652
667
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
653
668
 
654
669
    def test_is_directory(self):
662
677
    def test_make_branch_builder(self):
663
678
        builder = self.make_branch_builder('dir')
664
679
        rev_id = builder.build_commit()
665
 
        self.failUnlessExists('dir')
 
680
        self.assertPathExists('dir')
666
681
        a_dir = bzrdir.BzrDir.open('dir')
667
682
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
668
683
        a_branch = a_dir.open_branch()
684
699
        self.assertIsInstance(result_bzrdir.transport,
685
700
                              memory.MemoryTransport)
686
701
        # should not be on disk, should only be in memory
687
 
        self.failIfExists('subdir')
 
702
        self.assertPathDoesNotExist('subdir')
688
703
 
689
704
 
690
705
class TestChrootedTest(tests.ChrootedTestCase):
691
706
 
692
707
    def test_root_is_root(self):
693
 
        from bzrlib.transport import get_transport
694
 
        t = get_transport(self.get_readonly_url())
 
708
        t = transport.get_transport(self.get_readonly_url())
695
709
        url = t.base
696
710
        self.assertEqual(url, t.clone('..').base)
697
711
 
699
713
class TestProfileResult(tests.TestCase):
700
714
 
701
715
    def test_profiles_tests(self):
702
 
        self.requireFeature(test_lsprof.LSProfFeature)
703
 
        terminal = testtools.tests.helpers.ExtendedTestResult()
 
716
        self.requireFeature(features.lsprof_feature)
 
717
        terminal = testtools.testresult.doubles.ExtendedTestResult()
704
718
        result = tests.ProfileResult(terminal)
705
719
        class Sample(tests.TestCase):
706
720
            def a(self):
723
737
                descriptions=0,
724
738
                verbosity=1,
725
739
                )
726
 
        capture = testtools.tests.helpers.ExtendedTestResult()
 
740
        capture = testtools.testresult.doubles.ExtendedTestResult()
727
741
        test_case.run(MultiTestResult(result, capture))
728
742
        run_case = capture._events[0][1]
729
743
        timed_string = result._testTimeString(run_case)
750
764
        self.check_timing(ShortDelayTestCase('test_short_delay'),
751
765
                          r"^ +[0-9]+ms$")
752
766
 
753
 
    def _patch_get_bzr_source_tree(self):
754
 
        # Reading from the actual source tree breaks isolation, but we don't
755
 
        # want to assume that thats *all* that would happen.
756
 
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', lambda: None)
757
 
 
758
 
    def test_assigned_benchmark_file_stores_date(self):
759
 
        self._patch_get_bzr_source_tree()
760
 
        output = StringIO()
761
 
        result = bzrlib.tests.TextTestResult(self._log_file,
762
 
                                        descriptions=0,
763
 
                                        verbosity=1,
764
 
                                        bench_history=output
765
 
                                        )
766
 
        output_string = output.getvalue()
767
 
        # if you are wondering about the regexp please read the comment in
768
 
        # test_bench_history (bzrlib.tests.test_selftest.TestRunner)
769
 
        # XXX: what comment?  -- Andrew Bennetts
770
 
        self.assertContainsRe(output_string, "--date [0-9.]+")
771
 
 
772
 
    def test_benchhistory_records_test_times(self):
773
 
        self._patch_get_bzr_source_tree()
774
 
        result_stream = StringIO()
775
 
        result = bzrlib.tests.TextTestResult(
776
 
            self._log_file,
777
 
            descriptions=0,
778
 
            verbosity=1,
779
 
            bench_history=result_stream
780
 
            )
781
 
 
782
 
        # we want profile a call and check that its test duration is recorded
783
 
        # make a new test instance that when run will generate a benchmark
784
 
        example_test_case = TestTestResult("_time_hello_world_encoding")
785
 
        # execute the test, which should succeed and record times
786
 
        example_test_case.run(result)
787
 
        lines = result_stream.getvalue().splitlines()
788
 
        self.assertEqual(2, len(lines))
789
 
        self.assertContainsRe(lines[1],
790
 
            " *[0-9]+ms bzrlib.tests.test_selftest.TestTestResult"
791
 
            "._time_hello_world_encoding")
792
 
 
793
767
    def _time_hello_world_encoding(self):
794
768
        """Profile two sleep calls
795
769
 
800
774
 
801
775
    def test_lsprofiling(self):
802
776
        """Verbose test result prints lsprof statistics from test cases."""
803
 
        self.requireFeature(test_lsprof.LSProfFeature)
 
777
        self.requireFeature(features.lsprof_feature)
804
778
        result_stream = StringIO()
805
779
        result = bzrlib.tests.VerboseTestResult(
806
 
            unittest._WritelnDecorator(result_stream),
 
780
            result_stream,
807
781
            descriptions=0,
808
782
            verbosity=2,
809
783
            )
835
809
        self.assertContainsRe(output,
836
810
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
837
811
 
 
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
 
838
827
    def test_known_failure(self):
839
828
        """A KnownFailure being raised should trigger several result actions."""
840
829
        class InstrumentedTestResult(tests.ExtendedTestResult):
841
830
            def stopTestRun(self): pass
842
 
            def startTests(self): pass
843
 
            def report_test_start(self, test): pass
 
831
            def report_tests_starting(self): pass
844
832
            def report_known_failure(self, test, err=None, details=None):
845
833
                self._call = test, 'known failure'
846
834
        result = InstrumentedTestResult(None, None, None, None)
864
852
        # verbose test output formatting
865
853
        result_stream = StringIO()
866
854
        result = bzrlib.tests.VerboseTestResult(
867
 
            unittest._WritelnDecorator(result_stream),
 
855
            result_stream,
868
856
            descriptions=0,
869
857
            verbosity=2,
870
858
            )
880
868
        output = result_stream.getvalue()[prefix:]
881
869
        lines = output.splitlines()
882
870
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
 
871
        if sys.version_info > (2, 7):
 
872
            self.expectFailure("_ExpectedFailure on 2.7 loses the message",
 
873
                self.assertNotEqual, lines[1], '    ')
883
874
        self.assertEqual(lines[1], '    foo')
884
875
        self.assertEqual(2, len(lines))
885
876
 
893
884
        """Test the behaviour of invoking addNotSupported."""
894
885
        class InstrumentedTestResult(tests.ExtendedTestResult):
895
886
            def stopTestRun(self): pass
896
 
            def startTests(self): pass
897
 
            def report_test_start(self, test): pass
 
887
            def report_tests_starting(self): pass
898
888
            def report_unsupported(self, test, feature):
899
889
                self._call = test, feature
900
890
        result = InstrumentedTestResult(None, None, None, None)
901
891
        test = SampleTestCase('_test_pass')
902
 
        feature = tests.Feature()
 
892
        feature = features.Feature()
903
893
        result.startTest(test)
904
894
        result.addNotSupported(test, feature)
905
895
        # it should invoke 'report_unsupported'.
919
909
        # verbose test output formatting
920
910
        result_stream = StringIO()
921
911
        result = bzrlib.tests.VerboseTestResult(
922
 
            unittest._WritelnDecorator(result_stream),
 
912
            result_stream,
923
913
            descriptions=0,
924
914
            verbosity=2,
925
915
            )
926
916
        test = self.get_passing_test()
927
 
        feature = tests.Feature()
 
917
        feature = features.Feature()
928
918
        result.startTest(test)
929
919
        prefix = len(result_stream.getvalue())
930
920
        result.report_unsupported(test, feature)
939
929
        """An UnavailableFeature being raised should invoke addNotSupported."""
940
930
        class InstrumentedTestResult(tests.ExtendedTestResult):
941
931
            def stopTestRun(self): pass
942
 
            def startTests(self): pass
943
 
            def report_test_start(self, test): pass
 
932
            def report_tests_starting(self): pass
944
933
            def addNotSupported(self, test, feature):
945
934
                self._call = test, feature
946
935
        result = InstrumentedTestResult(None, None, None, None)
947
 
        feature = tests.Feature()
 
936
        feature = features.Feature()
948
937
        class Test(tests.TestCase):
949
938
            def test_function(self):
950
939
                raise tests.UnavailableFeature(feature)
988
977
        class InstrumentedTestResult(tests.ExtendedTestResult):
989
978
            calls = 0
990
979
            def startTests(self): self.calls += 1
991
 
            def report_test_start(self, test): pass
992
980
        result = InstrumentedTestResult(None, None, None, None)
993
981
        def test_function():
994
982
            pass
996
984
        test.run(result)
997
985
        self.assertEquals(1, result.calls)
998
986
 
999
 
 
1000
 
class TestUnicodeFilenameFeature(tests.TestCase):
1001
 
 
1002
 
    def test_probe_passes(self):
1003
 
        """UnicodeFilenameFeature._probe passes."""
1004
 
        # We can't test much more than that because the behaviour depends
1005
 
        # on the platform.
1006
 
        tests.UnicodeFilenameFeature._probe()
 
987
    def test_startTests_only_once(self):
 
988
        """With multiple tests startTests should still only be called once"""
 
989
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
990
            calls = 0
 
991
            def startTests(self): self.calls += 1
 
992
        result = InstrumentedTestResult(None, None, None, None)
 
993
        suite = unittest.TestSuite([
 
994
            unittest.FunctionTestCase(lambda: None),
 
995
            unittest.FunctionTestCase(lambda: None)])
 
996
        suite.run(result)
 
997
        self.assertEquals(1, result.calls)
 
998
        self.assertEquals(2, result.count)
1007
999
 
1008
1000
 
1009
1001
class TestRunner(tests.TestCase):
1022
1014
        because of our use of global state.
1023
1015
        """
1024
1016
        old_root = tests.TestCaseInTempDir.TEST_ROOT
1025
 
        old_leak = tests.TestCase._first_thread_leaker_id
1026
1017
        try:
1027
1018
            tests.TestCaseInTempDir.TEST_ROOT = None
1028
 
            tests.TestCase._first_thread_leaker_id = None
1029
1019
            return testrunner.run(test)
1030
1020
        finally:
1031
1021
            tests.TestCaseInTempDir.TEST_ROOT = old_root
1032
 
            tests.TestCase._first_thread_leaker_id = old_leak
1033
1022
 
1034
1023
    def test_known_failure_failed_run(self):
1035
1024
        # run a test that generates a known failure which should be printed in
1040
1029
        test = unittest.TestSuite()
1041
1030
        test.addTest(Test("known_failure_test"))
1042
1031
        def failing_test():
1043
 
            self.fail('foo')
 
1032
            raise AssertionError('foo')
1044
1033
        test.addTest(unittest.FunctionTestCase(failing_test))
1045
1034
        stream = StringIO()
1046
1035
        runner = tests.TextTestRunner(stream=stream)
1054
1043
            '^----------------------------------------------------------------------\n'
1055
1044
            'Traceback \\(most recent call last\\):\n'
1056
1045
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1057
 
            '    self.fail\\(\'foo\'\\)\n'
 
1046
            '    raise AssertionError\\(\'foo\'\\)\n'
1058
1047
            '.*'
1059
1048
            '^----------------------------------------------------------------------\n'
1060
1049
            '.*'
1066
1055
        # the final output.
1067
1056
        class Test(tests.TestCase):
1068
1057
            def known_failure_test(self):
1069
 
                self.expectFailure('failed', self.assertTrue, False)
 
1058
                self.knownFailure("Never works...")
1070
1059
        test = Test("known_failure_test")
1071
1060
        stream = StringIO()
1072
1061
        runner = tests.TextTestRunner(stream=stream)
1078
1067
            '\n'
1079
1068
            'OK \\(known_failures=1\\)\n')
1080
1069
 
 
1070
    def test_unexpected_success_bad(self):
 
1071
        class Test(tests.TestCase):
 
1072
            def test_truth(self):
 
1073
                self.expectFailure("No absolute truth", self.assertTrue, True)
 
1074
        runner = tests.TextTestRunner(stream=StringIO())
 
1075
        result = self.run_test_runner(runner, Test("test_truth"))
 
1076
        if testtools_version <= (0, 9, 11):
 
1077
            self.assertContainsRe(runner.stream.getvalue(),
 
1078
                "=+\n"
 
1079
                "FAIL: \\S+\.test_truth\n"
 
1080
                "-+\n"
 
1081
                "(?:.*\n)*"
 
1082
                "No absolute truth\n"
 
1083
                "(?:.*\n)*"
 
1084
                "-+\n"
 
1085
                "Ran 1 test in .*\n"
 
1086
                "\n"
 
1087
                "FAILED \\(failures=1\\)\n\\Z")
 
1088
        else:
 
1089
            self.assertContainsRe(runner.stream.getvalue(),
 
1090
                "=+\n"
 
1091
                "FAIL: \\S+\.test_truth\n"
 
1092
                "-+\n"
 
1093
                "Empty attachments:\n"
 
1094
                "  log\n"
 
1095
                "\n"
 
1096
                "reason: {{{No absolute truth}}}\n"
 
1097
                "-+\n"
 
1098
                "Ran 1 test in .*\n"
 
1099
                "\n"
 
1100
                "FAILED \\(failures=1\\)\n\\Z")
 
1101
 
1081
1102
    def test_result_decorator(self):
1082
1103
        # decorate results
1083
1104
        calls = []
1084
 
        class LoggingDecorator(tests.ForwardingResult):
 
1105
        class LoggingDecorator(ExtendedToOriginalDecorator):
1085
1106
            def startTest(self, test):
1086
 
                tests.ForwardingResult.startTest(self, test)
 
1107
                ExtendedToOriginalDecorator.startTest(self, test)
1087
1108
                calls.append('start')
1088
1109
        test = unittest.FunctionTestCase(lambda:None)
1089
1110
        stream = StringIO()
1167
1188
 
1168
1189
    def test_unsupported_features_listed(self):
1169
1190
        """When unsupported features are encountered they are detailed."""
1170
 
        class Feature1(tests.Feature):
 
1191
        class Feature1(features.Feature):
1171
1192
            def _probe(self): return False
1172
 
        class Feature2(tests.Feature):
 
1193
        class Feature2(features.Feature):
1173
1194
            def _probe(self): return False
1174
1195
        # create sample tests
1175
1196
        test1 = SampleTestCase('_test_pass')
1190
1211
            ],
1191
1212
            lines[-3:])
1192
1213
 
1193
 
    def _patch_get_bzr_source_tree(self):
1194
 
        # Reading from the actual source tree breaks isolation, but we don't
1195
 
        # want to assume that thats *all* that would happen.
1196
 
        self._get_source_tree_calls = []
1197
 
        def new_get():
1198
 
            self._get_source_tree_calls.append("called")
1199
 
            return None
1200
 
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree',  new_get)
1201
 
 
1202
 
    def test_bench_history(self):
1203
 
        # tests that the running the benchmark passes bench_history into
1204
 
        # the test result object. We can tell that happens if
1205
 
        # _get_bzr_source_tree is called.
1206
 
        self._patch_get_bzr_source_tree()
1207
 
        test = TestRunner('dummy_test')
1208
 
        output = StringIO()
1209
 
        runner = tests.TextTestRunner(stream=self._log_file,
1210
 
                                      bench_history=output)
1211
 
        result = self.run_test_runner(runner, test)
1212
 
        output_string = output.getvalue()
1213
 
        self.assertContainsRe(output_string, "--date [0-9.]+")
1214
 
        self.assertLength(1, self._get_source_tree_calls)
 
1214
    def test_verbose_test_count(self):
 
1215
        """A verbose test run reports the right test count at the start"""
 
1216
        suite = TestUtil.TestSuite([
 
1217
            unittest.FunctionTestCase(lambda:None),
 
1218
            unittest.FunctionTestCase(lambda:None)])
 
1219
        self.assertEqual(suite.countTestCases(), 2)
 
1220
        stream = StringIO()
 
1221
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
 
1222
        # Need to use the CountingDecorator as that's what sets num_tests
 
1223
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
 
1224
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
1215
1225
 
1216
1226
    def test_startTestRun(self):
1217
1227
        """run should call result.startTestRun()"""
1218
1228
        calls = []
1219
 
        class LoggingDecorator(tests.ForwardingResult):
 
1229
        class LoggingDecorator(ExtendedToOriginalDecorator):
1220
1230
            def startTestRun(self):
1221
 
                tests.ForwardingResult.startTestRun(self)
 
1231
                ExtendedToOriginalDecorator.startTestRun(self)
1222
1232
                calls.append('startTestRun')
1223
1233
        test = unittest.FunctionTestCase(lambda:None)
1224
1234
        stream = StringIO()
1230
1240
    def test_stopTestRun(self):
1231
1241
        """run should call result.stopTestRun()"""
1232
1242
        calls = []
1233
 
        class LoggingDecorator(tests.ForwardingResult):
 
1243
        class LoggingDecorator(ExtendedToOriginalDecorator):
1234
1244
            def stopTestRun(self):
1235
 
                tests.ForwardingResult.stopTestRun(self)
 
1245
                ExtendedToOriginalDecorator.stopTestRun(self)
1236
1246
                calls.append('stopTestRun')
1237
1247
        test = unittest.FunctionTestCase(lambda:None)
1238
1248
        stream = StringIO()
1241
1251
        result = self.run_test_runner(runner, test)
1242
1252
        self.assertLength(1, calls)
1243
1253
 
 
1254
    def test_unicode_test_output_on_ascii_stream(self):
 
1255
        """Showing results should always succeed even on an ascii console"""
 
1256
        class FailureWithUnicode(tests.TestCase):
 
1257
            def test_log_unicode(self):
 
1258
                self.log(u"\u2606")
 
1259
                self.fail("Now print that log!")
 
1260
        out = StringIO()
 
1261
        self.overrideAttr(osutils, "get_terminal_encoding",
 
1262
            lambda trace=False: "ascii")
 
1263
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
 
1264
            FailureWithUnicode("test_log_unicode"))
 
1265
        if testtools_version > (0, 9, 11):
 
1266
            self.assertContainsRe(out.getvalue(), "log: {{{\d+\.\d+  \\\\u2606}}}")
 
1267
        else:
 
1268
            self.assertContainsRe(out.getvalue(),
 
1269
                "Text attachment: log\n"
 
1270
                "-+\n"
 
1271
                "\d+\.\d+  \\\\u2606\n"
 
1272
                "-+\n")
 
1273
 
1244
1274
 
1245
1275
class SampleTestCase(tests.TestCase):
1246
1276
 
1421
1451
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1422
1452
        output_stream = StringIO()
1423
1453
        result = bzrlib.tests.VerboseTestResult(
1424
 
            unittest._WritelnDecorator(output_stream),
 
1454
            output_stream,
1425
1455
            descriptions=0,
1426
1456
            verbosity=2)
1427
1457
        sample_test.run(result)
1434
1464
        # Note this test won't fail with hooks that the core library doesn't
1435
1465
        # use - but it trigger with a plugin that adds hooks, so its still a
1436
1466
        # useful warning in that case.
1437
 
        self.assertEqual(bzrlib.branch.BranchHooks(),
1438
 
            bzrlib.branch.Branch.hooks)
1439
 
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
 
1467
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
 
1468
        self.assertEqual(
 
1469
            bzrlib.smart.server.SmartServerHooks(),
1440
1470
            bzrlib.smart.server.SmartTCPServer.hooks)
1441
 
        self.assertEqual(bzrlib.commands.CommandHooks(),
1442
 
            bzrlib.commands.Command.hooks)
 
1471
        self.assertEqual(
 
1472
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
1443
1473
 
1444
1474
    def test__gather_lsprof_in_benchmarks(self):
1445
1475
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1446
1476
 
1447
1477
        Each self.time() call is individually and separately profiled.
1448
1478
        """
1449
 
        self.requireFeature(test_lsprof.LSProfFeature)
 
1479
        self.requireFeature(features.lsprof_feature)
1450
1480
        # overrides the class member with an instance member so no cleanup
1451
1481
        # needed.
1452
1482
        self._gather_lsprof_in_benchmarks = True
1482
1512
 
1483
1513
    def test_requireFeature_available(self):
1484
1514
        """self.requireFeature(available) is a no-op."""
1485
 
        class Available(tests.Feature):
 
1515
        class Available(features.Feature):
1486
1516
            def _probe(self):return True
1487
1517
        feature = Available()
1488
1518
        self.requireFeature(feature)
1489
1519
 
1490
1520
    def test_requireFeature_unavailable(self):
1491
1521
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1492
 
        class Unavailable(tests.Feature):
 
1522
        class Unavailable(features.Feature):
1493
1523
            def _probe(self):return False
1494
1524
        feature = Unavailable()
1495
1525
        self.assertRaises(tests.UnavailableFeature,
1654
1684
        test.run(unittest.TestResult())
1655
1685
        self.assertEqual('original', obj.test_attr)
1656
1686
 
 
1687
    def test_recordCalls(self):
 
1688
        from bzrlib.tests import test_selftest
 
1689
        calls = self.recordCalls(
 
1690
            test_selftest, '_add_numbers')
 
1691
        self.assertEqual(test_selftest._add_numbers(2, 10),
 
1692
            12)
 
1693
        self.assertEquals(calls, [((2, 10), {})])
 
1694
 
 
1695
 
 
1696
def _add_numbers(a, b):
 
1697
    return a + b
 
1698
 
 
1699
 
 
1700
class _MissingFeature(features.Feature):
 
1701
    def _probe(self):
 
1702
        return False
 
1703
missing_feature = _MissingFeature()
 
1704
 
 
1705
 
 
1706
def _get_test(name):
 
1707
    """Get an instance of a specific example test.
 
1708
 
 
1709
    We protect this in a function so that they don't auto-run in the test
 
1710
    suite.
 
1711
    """
 
1712
 
 
1713
    class ExampleTests(tests.TestCase):
 
1714
 
 
1715
        def test_fail(self):
 
1716
            mutter('this was a failing test')
 
1717
            self.fail('this test will fail')
 
1718
 
 
1719
        def test_error(self):
 
1720
            mutter('this test errored')
 
1721
            raise RuntimeError('gotcha')
 
1722
 
 
1723
        def test_missing_feature(self):
 
1724
            mutter('missing the feature')
 
1725
            self.requireFeature(missing_feature)
 
1726
 
 
1727
        def test_skip(self):
 
1728
            mutter('this test will be skipped')
 
1729
            raise tests.TestSkipped('reason')
 
1730
 
 
1731
        def test_success(self):
 
1732
            mutter('this test succeeds')
 
1733
 
 
1734
        def test_xfail(self):
 
1735
            mutter('test with expected failure')
 
1736
            self.knownFailure('this_fails')
 
1737
 
 
1738
        def test_unexpected_success(self):
 
1739
            mutter('test with unexpected success')
 
1740
            self.expectFailure('should_fail', lambda: None)
 
1741
 
 
1742
    return ExampleTests(name)
 
1743
 
 
1744
 
 
1745
class TestTestCaseLogDetails(tests.TestCase):
 
1746
 
 
1747
    def _run_test(self, test_name):
 
1748
        test = _get_test(test_name)
 
1749
        result = testtools.TestResult()
 
1750
        test.run(result)
 
1751
        return result
 
1752
 
 
1753
    def test_fail_has_log(self):
 
1754
        result = self._run_test('test_fail')
 
1755
        self.assertEqual(1, len(result.failures))
 
1756
        result_content = result.failures[0][1]
 
1757
        if testtools_version < (0, 9, 12):
 
1758
            self.assertContainsRe(result_content, 'Text attachment: log')
 
1759
        self.assertContainsRe(result_content, 'this was a failing test')
 
1760
 
 
1761
    def test_error_has_log(self):
 
1762
        result = self._run_test('test_error')
 
1763
        self.assertEqual(1, len(result.errors))
 
1764
        result_content = result.errors[0][1]
 
1765
        if testtools_version < (0, 9, 12):
 
1766
            self.assertContainsRe(result_content, 'Text attachment: log')
 
1767
        self.assertContainsRe(result_content, 'this test errored')
 
1768
 
 
1769
    def test_skip_has_no_log(self):
 
1770
        result = self._run_test('test_skip')
 
1771
        self.assertEqual(['reason'], result.skip_reasons.keys())
 
1772
        skips = result.skip_reasons['reason']
 
1773
        self.assertEqual(1, len(skips))
 
1774
        test = skips[0]
 
1775
        self.assertFalse('log' in test.getDetails())
 
1776
 
 
1777
    def test_missing_feature_has_no_log(self):
 
1778
        # testtools doesn't know about addNotSupported, so it just gets
 
1779
        # considered as a skip
 
1780
        result = self._run_test('test_missing_feature')
 
1781
        self.assertEqual([missing_feature], result.skip_reasons.keys())
 
1782
        skips = result.skip_reasons[missing_feature]
 
1783
        self.assertEqual(1, len(skips))
 
1784
        test = skips[0]
 
1785
        self.assertFalse('log' in test.getDetails())
 
1786
 
 
1787
    def test_xfail_has_no_log(self):
 
1788
        result = self._run_test('test_xfail')
 
1789
        self.assertEqual(1, len(result.expectedFailures))
 
1790
        result_content = result.expectedFailures[0][1]
 
1791
        self.assertNotContainsRe(result_content, 'Text attachment: log')
 
1792
        self.assertNotContainsRe(result_content, 'test with expected failure')
 
1793
 
 
1794
    def test_unexpected_success_has_log(self):
 
1795
        result = self._run_test('test_unexpected_success')
 
1796
        self.assertEqual(1, len(result.unexpectedSuccesses))
 
1797
        # Inconsistency, unexpectedSuccesses is a list of tests,
 
1798
        # expectedFailures is a list of reasons?
 
1799
        test = result.unexpectedSuccesses[0]
 
1800
        details = test.getDetails()
 
1801
        self.assertTrue('log' in details)
 
1802
 
 
1803
 
 
1804
class TestTestCloning(tests.TestCase):
 
1805
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
 
1806
 
 
1807
    def test_cloned_testcase_does_not_share_details(self):
 
1808
        """A TestCase cloned with clone_test does not share mutable attributes
 
1809
        such as details or cleanups.
 
1810
        """
 
1811
        class Test(tests.TestCase):
 
1812
            def test_foo(self):
 
1813
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
 
1814
        orig_test = Test('test_foo')
 
1815
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
 
1816
        orig_test.run(unittest.TestResult())
 
1817
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
 
1818
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
 
1819
 
 
1820
    def test_double_apply_scenario_preserves_first_scenario(self):
 
1821
        """Applying two levels of scenarios to a test preserves the attributes
 
1822
        added by both scenarios.
 
1823
        """
 
1824
        class Test(tests.TestCase):
 
1825
            def test_foo(self):
 
1826
                pass
 
1827
        test = Test('test_foo')
 
1828
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
 
1829
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
 
1830
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
 
1831
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
 
1832
        all_tests = list(tests.iter_suite_tests(suite))
 
1833
        self.assertLength(4, all_tests)
 
1834
        all_xys = sorted((t.x, t.y) for t in all_tests)
 
1835
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
 
1836
 
1657
1837
 
1658
1838
# NB: Don't delete this; it's not actually from 0.11!
1659
1839
@deprecated_function(deprecated_in((0, 11, 0)))
1787
1967
    def test_make_branch_and_tree_with_format(self):
1788
1968
        # we should be able to supply a format to make_branch_and_tree
1789
1969
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
1790
 
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
1791
1970
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
1792
1971
                              bzrlib.bzrdir.BzrDirMetaFormat1)
1793
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
1794
 
                              bzrlib.bzrdir.BzrDirFormat6)
1795
1972
 
1796
1973
    def test_make_branch_and_memory_tree(self):
1797
1974
        # we should be able to get a new branch and a mutable tree from
1814
1991
                tree.branch.repository.bzrdir.root_transport)
1815
1992
 
1816
1993
 
1817
 
class SelfTestHelper:
 
1994
class SelfTestHelper(object):
1818
1995
 
1819
1996
    def run_selftest(self, **kwargs):
1820
1997
        """Run selftest returning its output."""
1874
2051
        self.assertLength(2, output.readlines())
1875
2052
 
1876
2053
    def test_lsprof_tests(self):
1877
 
        self.requireFeature(test_lsprof.LSProfFeature)
1878
 
        calls = []
 
2054
        self.requireFeature(features.lsprof_feature)
 
2055
        results = []
1879
2056
        class Test(object):
1880
2057
            def __call__(test, result):
1881
2058
                test.run(result)
1882
2059
            def run(test, result):
1883
 
                self.assertIsInstance(result, tests.ForwardingResult)
1884
 
                calls.append("called")
 
2060
                results.append(result)
1885
2061
            def countTestCases(self):
1886
2062
                return 1
1887
2063
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
1888
 
        self.assertLength(1, calls)
 
2064
        self.assertLength(1, results)
 
2065
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
1889
2066
 
1890
2067
    def test_random(self):
1891
2068
        # test randomising by listing a number of tests.
1971
2148
            load_list='missing file name', list_only=True)
1972
2149
 
1973
2150
 
 
2151
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
 
2152
 
 
2153
    _test_needs_features = [features.subunit]
 
2154
 
 
2155
    def run_subunit_stream(self, test_name):
 
2156
        from subunit import ProtocolTestCase
 
2157
        def factory():
 
2158
            return TestUtil.TestSuite([_get_test(test_name)])
 
2159
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
 
2160
            test_suite_factory=factory)
 
2161
        test = ProtocolTestCase(stream)
 
2162
        result = testtools.TestResult()
 
2163
        test.run(result)
 
2164
        content = stream.getvalue()
 
2165
        return content, result
 
2166
 
 
2167
    def test_fail_has_log(self):
 
2168
        content, result = self.run_subunit_stream('test_fail')
 
2169
        self.assertEqual(1, len(result.failures))
 
2170
        self.assertContainsRe(content, '(?m)^log$')
 
2171
        self.assertContainsRe(content, 'this test will fail')
 
2172
 
 
2173
    def test_error_has_log(self):
 
2174
        content, result = self.run_subunit_stream('test_error')
 
2175
        self.assertContainsRe(content, '(?m)^log$')
 
2176
        self.assertContainsRe(content, 'this test errored')
 
2177
 
 
2178
    def test_skip_has_no_log(self):
 
2179
        content, result = self.run_subunit_stream('test_skip')
 
2180
        self.assertNotContainsRe(content, '(?m)^log$')
 
2181
        self.assertNotContainsRe(content, 'this test will be skipped')
 
2182
        self.assertEqual(['reason'], result.skip_reasons.keys())
 
2183
        skips = result.skip_reasons['reason']
 
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_missing_feature_has_no_log(self):
 
2190
        content, result = self.run_subunit_stream('test_missing_feature')
 
2191
        self.assertNotContainsRe(content, '(?m)^log$')
 
2192
        self.assertNotContainsRe(content, 'missing the feature')
 
2193
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
 
2194
        skips = result.skip_reasons['_MissingFeature\n']
 
2195
        self.assertEqual(1, len(skips))
 
2196
        test = skips[0]
 
2197
        # RemotedTestCase doesn't preserve the "details"
 
2198
        ## self.assertFalse('log' in test.getDetails())
 
2199
 
 
2200
    def test_xfail_has_no_log(self):
 
2201
        content, result = self.run_subunit_stream('test_xfail')
 
2202
        self.assertNotContainsRe(content, '(?m)^log$')
 
2203
        self.assertNotContainsRe(content, 'test with expected failure')
 
2204
        self.assertEqual(1, len(result.expectedFailures))
 
2205
        result_content = result.expectedFailures[0][1]
 
2206
        self.assertNotContainsRe(result_content, 'Text attachment: log')
 
2207
        self.assertNotContainsRe(result_content, 'test with expected failure')
 
2208
 
 
2209
    def test_unexpected_success_has_log(self):
 
2210
        content, result = self.run_subunit_stream('test_unexpected_success')
 
2211
        self.assertContainsRe(content, '(?m)^log$')
 
2212
        self.assertContainsRe(content, 'test with unexpected success')
 
2213
        # GZ 2011-05-18: Old versions of subunit treat unexpected success as a
 
2214
        #                success, if a min version check is added remove this
 
2215
        from subunit import TestProtocolClient as _Client
 
2216
        if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
 
2217
            self.expectFailure('subunit treats "unexpectedSuccess"'
 
2218
                               ' as a plain success',
 
2219
                self.assertEqual, 1, len(result.unexpectedSuccesses))
 
2220
        self.assertEqual(1, len(result.unexpectedSuccesses))
 
2221
        test = result.unexpectedSuccesses[0]
 
2222
        # RemotedTestCase doesn't preserve the "details"
 
2223
        ## self.assertTrue('log' in test.getDetails())
 
2224
 
 
2225
    def test_success_has_no_log(self):
 
2226
        content, result = self.run_subunit_stream('test_success')
 
2227
        self.assertEqual(1, result.testsRun)
 
2228
        self.assertNotContainsRe(content, '(?m)^log$')
 
2229
        self.assertNotContainsRe(content, 'this test succeeds')
 
2230
 
 
2231
 
1974
2232
class TestRunBzr(tests.TestCase):
1975
2233
 
1976
2234
    out = ''
2099
2357
        # stdout and stderr of the invoked run_bzr
2100
2358
        current_factory = bzrlib.ui.ui_factory
2101
2359
        self.run_bzr(['foo'])
2102
 
        self.failIf(current_factory is self.factory)
 
2360
        self.assertFalse(current_factory is self.factory)
2103
2361
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2104
2362
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2105
2363
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2262
2520
 
2263
2521
 
2264
2522
class TestStartBzrSubProcess(tests.TestCase):
 
2523
    """Stub test start_bzr_subprocess."""
2265
2524
 
2266
 
    def check_popen_state(self):
2267
 
        """Replace to make assertions when popen is called."""
 
2525
    def _subprocess_log_cleanup(self):
 
2526
        """Inhibits the base version as we don't produce a log file."""
2268
2527
 
2269
2528
    def _popen(self, *args, **kwargs):
2270
 
        """Record the command that is run, so that we can ensure it is correct"""
 
2529
        """Override the base version to record the command that is run.
 
2530
 
 
2531
        From there we can ensure it is correct without spawning a real process.
 
2532
        """
2271
2533
        self.check_popen_state()
2272
2534
        self._popen_args = args
2273
2535
        self._popen_kwargs = kwargs
2274
2536
        raise _DontSpawnProcess()
2275
2537
 
 
2538
    def check_popen_state(self):
 
2539
        """Replace to make assertions when popen is called."""
 
2540
 
2276
2541
    def test_run_bzr_subprocess_no_plugins(self):
2277
2542
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2278
2543
        command = self._popen_args[0]
2282
2547
 
2283
2548
    def test_allow_plugins(self):
2284
2549
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2285
 
            allow_plugins=True)
 
2550
                          allow_plugins=True)
2286
2551
        command = self._popen_args[0]
2287
2552
        self.assertEqual([], command[2:])
2288
2553
 
2289
2554
    def test_set_env(self):
2290
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2555
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2291
2556
        # set in the child
2292
2557
        def check_environment():
2293
2558
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2294
2559
        self.check_popen_state = check_environment
2295
2560
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2296
 
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
 
2561
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
2297
2562
        # not set in theparent
2298
2563
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2299
2564
 
2300
2565
    def test_run_bzr_subprocess_env_del(self):
2301
2566
        """run_bzr_subprocess can remove environment variables too."""
2302
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2567
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2303
2568
        def check_environment():
2304
2569
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2305
2570
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2306
2571
        self.check_popen_state = check_environment
2307
2572
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2308
 
            env_changes={'EXISTANT_ENV_VAR':None})
 
2573
                          env_changes={'EXISTANT_ENV_VAR':None})
2309
2574
        # Still set in parent
2310
2575
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2311
2576
        del os.environ['EXISTANT_ENV_VAR']
2312
2577
 
2313
2578
    def test_env_del_missing(self):
2314
 
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
 
2579
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2315
2580
        def check_environment():
2316
2581
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2317
2582
        self.check_popen_state = check_environment
2318
2583
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2319
 
            env_changes={'NON_EXISTANT_ENV_VAR':None})
 
2584
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
2320
2585
 
2321
2586
    def test_working_dir(self):
2322
2587
        """Test that we can specify the working dir for the child"""
2325
2590
        chdirs = []
2326
2591
        def chdir(path):
2327
2592
            chdirs.append(path)
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
 
2593
        self.overrideAttr(os, 'chdir', chdir)
 
2594
        def getcwd():
 
2595
            return 'current'
 
2596
        self.overrideAttr(osutils, 'getcwd', getcwd)
 
2597
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2598
                          working_dir='foo')
2340
2599
        self.assertEqual(['foo', 'current'], chdirs)
2341
2600
 
 
2601
    def test_get_bzr_path_with_cwd_bzrlib(self):
 
2602
        self.get_source_path = lambda: ""
 
2603
        self.overrideAttr(os.path, "isfile", lambda path: True)
 
2604
        self.assertEqual(self.get_bzr_path(), "bzr")
 
2605
 
2342
2606
 
2343
2607
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2344
2608
    """Tests that really need to do things with an external bzr."""
2357
2621
        self.assertEqual('bzr: interrupted\n', result[1])
2358
2622
 
2359
2623
 
2360
 
class TestFeature(tests.TestCase):
2361
 
 
2362
 
    def test_caching(self):
2363
 
        """Feature._probe is called by the feature at most once."""
2364
 
        class InstrumentedFeature(tests.Feature):
2365
 
            def __init__(self):
2366
 
                super(InstrumentedFeature, self).__init__()
2367
 
                self.calls = []
2368
 
            def _probe(self):
2369
 
                self.calls.append('_probe')
2370
 
                return False
2371
 
        feature = InstrumentedFeature()
2372
 
        feature.available()
2373
 
        self.assertEqual(['_probe'], feature.calls)
2374
 
        feature.available()
2375
 
        self.assertEqual(['_probe'], feature.calls)
2376
 
 
2377
 
    def test_named_str(self):
2378
 
        """Feature.__str__ should thunk to feature_name()."""
2379
 
        class NamedFeature(tests.Feature):
2380
 
            def feature_name(self):
2381
 
                return 'symlinks'
2382
 
        feature = NamedFeature()
2383
 
        self.assertEqual('symlinks', str(feature))
2384
 
 
2385
 
    def test_default_str(self):
2386
 
        """Feature.__str__ should default to __class__.__name__."""
2387
 
        class NamedFeature(tests.Feature):
2388
 
            pass
2389
 
        feature = NamedFeature()
2390
 
        self.assertEqual('NamedFeature', str(feature))
2391
 
 
2392
 
 
2393
 
class TestUnavailableFeature(tests.TestCase):
2394
 
 
2395
 
    def test_access_feature(self):
2396
 
        feature = tests.Feature()
2397
 
        exception = tests.UnavailableFeature(feature)
2398
 
        self.assertIs(feature, exception.args[0])
2399
 
 
2400
 
 
2401
 
simple_thunk_feature = tests._CompatabilityThunkFeature(
2402
 
    deprecated_in((2, 1, 0)),
2403
 
    'bzrlib.tests.test_selftest',
2404
 
    'simple_thunk_feature','UnicodeFilename',
2405
 
    replacement_module='bzrlib.tests'
2406
 
    )
2407
 
 
2408
 
class Test_CompatibilityFeature(tests.TestCase):
2409
 
 
2410
 
    def test_does_thunk(self):
2411
 
        res = self.callDeprecated(
2412
 
            ['bzrlib.tests.test_selftest.simple_thunk_feature was deprecated'
2413
 
             ' in version 2.1.0. Use bzrlib.tests.UnicodeFilename instead.'],
2414
 
            simple_thunk_feature.available)
2415
 
        self.assertEqual(tests.UnicodeFilename.available(), res)
2416
 
 
2417
 
 
2418
 
class TestModuleAvailableFeature(tests.TestCase):
2419
 
 
2420
 
    def test_available_module(self):
2421
 
        feature = tests.ModuleAvailableFeature('bzrlib.tests')
2422
 
        self.assertEqual('bzrlib.tests', feature.module_name)
2423
 
        self.assertEqual('bzrlib.tests', str(feature))
2424
 
        self.assertTrue(feature.available())
2425
 
        self.assertIs(tests, feature.module)
2426
 
 
2427
 
    def test_unavailable_module(self):
2428
 
        feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
2429
 
        self.assertEqual('bzrlib.no_such_module_exists', str(feature))
2430
 
        self.assertFalse(feature.available())
2431
 
        self.assertIs(None, feature.module)
2432
 
 
2433
 
 
2434
2624
class TestSelftestFiltering(tests.TestCase):
2435
2625
 
2436
2626
    def setUp(self):
2587
2777
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2588
2778
 
2589
2779
 
2590
 
class TestCheckInventoryShape(tests.TestCaseWithTransport):
 
2780
class TestCheckTreeShape(tests.TestCaseWithTransport):
2591
2781
 
2592
 
    def test_check_inventory_shape(self):
 
2782
    def test_check_tree_shape(self):
2593
2783
        files = ['a', 'b/', 'b/c']
2594
2784
        tree = self.make_branch_and_tree('.')
2595
2785
        self.build_tree(files)
2596
2786
        tree.add(files)
2597
2787
        tree.lock_read()
2598
2788
        try:
2599
 
            self.check_inventory_shape(tree.inventory, files)
 
2789
            self.check_tree_shape(tree, files)
2600
2790
        finally:
2601
2791
            tree.unlock()
2602
2792
 
2934
3124
        tpr.register('bar', 'bBB.aAA.rRR')
2935
3125
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
2936
3126
        self.assertThat(self.get_log(),
2937
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
 
3127
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
 
3128
                           doctest.ELLIPSIS))
2938
3129
 
2939
3130
    def test_get_unknown_prefix(self):
2940
3131
        tpr = self._get_registry()
2960
3151
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
2961
3152
 
2962
3153
 
 
3154
class TestThreadLeakDetection(tests.TestCase):
 
3155
    """Ensure when tests leak threads we detect and report it"""
 
3156
 
 
3157
    class LeakRecordingResult(tests.ExtendedTestResult):
 
3158
        def __init__(self):
 
3159
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
 
3160
            self.leaks = []
 
3161
        def _report_thread_leak(self, test, leaks, alive):
 
3162
            self.leaks.append((test, leaks))
 
3163
 
 
3164
    def test_testcase_without_addCleanups(self):
 
3165
        """Check old TestCase instances don't break with leak detection"""
 
3166
        class Test(unittest.TestCase):
 
3167
            def runTest(self):
 
3168
                pass
 
3169
        result = self.LeakRecordingResult()
 
3170
        test = Test()
 
3171
        result.startTestRun()
 
3172
        test.run(result)
 
3173
        result.stopTestRun()
 
3174
        self.assertEqual(result._tests_leaking_threads_count, 0)
 
3175
        self.assertEqual(result.leaks, [])
 
3176
        
 
3177
    def test_thread_leak(self):
 
3178
        """Ensure a thread that outlives the running of a test is reported
 
3179
 
 
3180
        Uses a thread that blocks on an event, and is started by the inner
 
3181
        test case. As the thread outlives the inner case's run, it should be
 
3182
        detected as a leak, but the event is then set so that the thread can
 
3183
        be safely joined in cleanup so it's not leaked for real.
 
3184
        """
 
3185
        event = threading.Event()
 
3186
        thread = threading.Thread(name="Leaker", target=event.wait)
 
3187
        class Test(tests.TestCase):
 
3188
            def test_leak(self):
 
3189
                thread.start()
 
3190
        result = self.LeakRecordingResult()
 
3191
        test = Test("test_leak")
 
3192
        self.addCleanup(thread.join)
 
3193
        self.addCleanup(event.set)
 
3194
        result.startTestRun()
 
3195
        test.run(result)
 
3196
        result.stopTestRun()
 
3197
        self.assertEqual(result._tests_leaking_threads_count, 1)
 
3198
        self.assertEqual(result._first_thread_leaker_id, test.id())
 
3199
        self.assertEqual(result.leaks, [(test, set([thread]))])
 
3200
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
 
3201
 
 
3202
    def test_multiple_leaks(self):
 
3203
        """Check multiple leaks are blamed on the test cases at fault
 
3204
 
 
3205
        Same concept as the previous test, but has one inner test method that
 
3206
        leaks two threads, and one that doesn't leak at all.
 
3207
        """
 
3208
        event = threading.Event()
 
3209
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
 
3210
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
 
3211
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
 
3212
        class Test(tests.TestCase):
 
3213
            def test_first_leak(self):
 
3214
                thread_b.start()
 
3215
            def test_second_no_leak(self):
 
3216
                pass
 
3217
            def test_third_leak(self):
 
3218
                thread_c.start()
 
3219
                thread_a.start()
 
3220
        result = self.LeakRecordingResult()
 
3221
        first_test = Test("test_first_leak")
 
3222
        third_test = Test("test_third_leak")
 
3223
        self.addCleanup(thread_a.join)
 
3224
        self.addCleanup(thread_b.join)
 
3225
        self.addCleanup(thread_c.join)
 
3226
        self.addCleanup(event.set)
 
3227
        result.startTestRun()
 
3228
        unittest.TestSuite(
 
3229
            [first_test, Test("test_second_no_leak"), third_test]
 
3230
            ).run(result)
 
3231
        result.stopTestRun()
 
3232
        self.assertEqual(result._tests_leaking_threads_count, 2)
 
3233
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
 
3234
        self.assertEqual(result.leaks, [
 
3235
            (first_test, set([thread_b])),
 
3236
            (third_test, set([thread_a, thread_c]))])
 
3237
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
 
3238
 
 
3239
 
 
3240
class TestPostMortemDebugging(tests.TestCase):
 
3241
    """Check post mortem debugging works when tests fail or error"""
 
3242
 
 
3243
    class TracebackRecordingResult(tests.ExtendedTestResult):
 
3244
        def __init__(self):
 
3245
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
 
3246
            self.postcode = None
 
3247
        def _post_mortem(self, tb=None):
 
3248
            """Record the code object at the end of the current traceback"""
 
3249
            tb = tb or sys.exc_info()[2]
 
3250
            if tb is not None:
 
3251
                next = tb.tb_next
 
3252
                while next is not None:
 
3253
                    tb = next
 
3254
                    next = next.tb_next
 
3255
                self.postcode = tb.tb_frame.f_code
 
3256
        def report_error(self, test, err):
 
3257
            pass
 
3258
        def report_failure(self, test, err):
 
3259
            pass
 
3260
 
 
3261
    def test_location_unittest_error(self):
 
3262
        """Needs right post mortem traceback with erroring unittest case"""
 
3263
        class Test(unittest.TestCase):
 
3264
            def runTest(self):
 
3265
                raise RuntimeError
 
3266
        result = self.TracebackRecordingResult()
 
3267
        Test().run(result)
 
3268
        self.assertEqual(result.postcode, Test.runTest.func_code)
 
3269
 
 
3270
    def test_location_unittest_failure(self):
 
3271
        """Needs right post mortem traceback with failing unittest case"""
 
3272
        class Test(unittest.TestCase):
 
3273
            def runTest(self):
 
3274
                raise self.failureException
 
3275
        result = self.TracebackRecordingResult()
 
3276
        Test().run(result)
 
3277
        self.assertEqual(result.postcode, Test.runTest.func_code)
 
3278
 
 
3279
    def test_location_bt_error(self):
 
3280
        """Needs right post mortem traceback with erroring bzrlib.tests case"""
 
3281
        class Test(tests.TestCase):
 
3282
            def test_error(self):
 
3283
                raise RuntimeError
 
3284
        result = self.TracebackRecordingResult()
 
3285
        Test("test_error").run(result)
 
3286
        self.assertEqual(result.postcode, Test.test_error.func_code)
 
3287
 
 
3288
    def test_location_bt_failure(self):
 
3289
        """Needs right post mortem traceback with failing bzrlib.tests case"""
 
3290
        class Test(tests.TestCase):
 
3291
            def test_failure(self):
 
3292
                raise self.failureException
 
3293
        result = self.TracebackRecordingResult()
 
3294
        Test("test_failure").run(result)
 
3295
        self.assertEqual(result.postcode, Test.test_failure.func_code)
 
3296
 
 
3297
    def test_env_var_triggers_post_mortem(self):
 
3298
        """Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
 
3299
        import pdb
 
3300
        result = tests.ExtendedTestResult(StringIO(), 0, 1)
 
3301
        post_mortem_calls = []
 
3302
        self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
 
3303
        self.overrideEnv('BZR_TEST_PDB', None)
 
3304
        result._post_mortem(1)
 
3305
        self.overrideEnv('BZR_TEST_PDB', 'on')
 
3306
        result._post_mortem(2)
 
3307
        self.assertEqual([2], post_mortem_calls)
 
3308
 
 
3309
 
2963
3310
class TestRunSuite(tests.TestCase):
2964
3311
 
2965
3312
    def test_runner_class(self):
2976
3323
                                                self.verbosity)
2977
3324
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2978
3325
        self.assertLength(1, calls)
 
3326
 
 
3327
 
 
3328
class TestEnvironHandling(tests.TestCase):
 
3329
 
 
3330
    def test_overrideEnv_None_called_twice_doesnt_leak(self):
 
3331
        self.assertFalse('MYVAR' in os.environ)
 
3332
        self.overrideEnv('MYVAR', '42')
 
3333
        # We use an embedded test to make sure we fix the _captureVar bug
 
3334
        class Test(tests.TestCase):
 
3335
            def test_me(self):
 
3336
                # The first call save the 42 value
 
3337
                self.overrideEnv('MYVAR', None)
 
3338
                self.assertEquals(None, os.environ.get('MYVAR'))
 
3339
                # Make sure we can call it twice
 
3340
                self.overrideEnv('MYVAR', None)
 
3341
                self.assertEquals(None, os.environ.get('MYVAR'))
 
3342
        output = StringIO()
 
3343
        result = tests.TextTestResult(output, 0, 1)
 
3344
        Test('test_me').run(result)
 
3345
        if not result.wasStrictlySuccessful():
 
3346
            self.fail(output.getvalue())
 
3347
        # We get our value back
 
3348
        self.assertEquals('42', os.environ.get('MYVAR'))
 
3349
 
 
3350
 
 
3351
class TestIsolatedEnv(tests.TestCase):
 
3352
    """Test isolating tests from os.environ.
 
3353
 
 
3354
    Since we use tests that are already isolated from os.environ a bit of care
 
3355
    should be taken when designing the tests to avoid bootstrap side-effects.
 
3356
    The tests start an already clean os.environ which allow doing valid
 
3357
    assertions about which variables are present or not and design tests around
 
3358
    these assertions.
 
3359
    """
 
3360
 
 
3361
    class ScratchMonkey(tests.TestCase):
 
3362
 
 
3363
        def test_me(self):
 
3364
            pass
 
3365
 
 
3366
    def test_basics(self):
 
3367
        # Make sure we know the definition of BZR_HOME: not part of os.environ
 
3368
        # for tests.TestCase.
 
3369
        self.assertTrue('BZR_HOME' in tests.isolated_environ)
 
3370
        self.assertEquals(None, tests.isolated_environ['BZR_HOME'])
 
3371
        # Being part of isolated_environ, BZR_HOME should not appear here
 
3372
        self.assertFalse('BZR_HOME' in os.environ)
 
3373
        # Make sure we know the definition of LINES: part of os.environ for
 
3374
        # tests.TestCase
 
3375
        self.assertTrue('LINES' in tests.isolated_environ)
 
3376
        self.assertEquals('25', tests.isolated_environ['LINES'])
 
3377
        self.assertEquals('25', os.environ['LINES'])
 
3378
 
 
3379
    def test_injecting_unknown_variable(self):
 
3380
        # BZR_HOME is known to be absent from os.environ
 
3381
        test = self.ScratchMonkey('test_me')
 
3382
        tests.override_os_environ(test, {'BZR_HOME': 'foo'})
 
3383
        self.assertEquals('foo', os.environ['BZR_HOME'])
 
3384
        tests.restore_os_environ(test)
 
3385
        self.assertFalse('BZR_HOME' in os.environ)
 
3386
 
 
3387
    def test_injecting_known_variable(self):
 
3388
        test = self.ScratchMonkey('test_me')
 
3389
        # LINES is known to be present in os.environ
 
3390
        tests.override_os_environ(test, {'LINES': '42'})
 
3391
        self.assertEquals('42', os.environ['LINES'])
 
3392
        tests.restore_os_environ(test)
 
3393
        self.assertEquals('25', os.environ['LINES'])
 
3394
 
 
3395
    def test_deleting_variable(self):
 
3396
        test = self.ScratchMonkey('test_me')
 
3397
        # LINES is known to be present in os.environ
 
3398
        tests.override_os_environ(test, {'LINES': None})
 
3399
        self.assertTrue('LINES' not in os.environ)
 
3400
        tests.restore_os_environ(test)
 
3401
        self.assertEquals('25', os.environ['LINES'])
 
3402
 
 
3403
 
 
3404
class TestDocTestSuiteIsolation(tests.TestCase):
 
3405
    """Test that `tests.DocTestSuite` isolates doc tests from os.environ.
 
3406
 
 
3407
    Since tests.TestCase alreay provides an isolation from os.environ, we use
 
3408
    the clean environment as a base for testing. To precisely capture the
 
3409
    isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
 
3410
    compare against.
 
3411
 
 
3412
    We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
 
3413
    not `os.environ` so each test overrides it to suit its needs.
 
3414
 
 
3415
    """
 
3416
 
 
3417
    def get_doctest_suite_for_string(self, klass, string):
 
3418
        class Finder(doctest.DocTestFinder):
 
3419
 
 
3420
            def find(*args, **kwargs):
 
3421
                test = doctest.DocTestParser().get_doctest(
 
3422
                    string, {}, 'foo', 'foo.py', 0)
 
3423
                return [test]
 
3424
 
 
3425
        suite = klass(test_finder=Finder())
 
3426
        return suite
 
3427
 
 
3428
    def run_doctest_suite_for_string(self, klass, string):
 
3429
        suite = self.get_doctest_suite_for_string(klass, string)
 
3430
        output = StringIO()
 
3431
        result = tests.TextTestResult(output, 0, 1)
 
3432
        suite.run(result)
 
3433
        return result, output
 
3434
 
 
3435
    def assertDocTestStringSucceds(self, klass, string):
 
3436
        result, output = self.run_doctest_suite_for_string(klass, string)
 
3437
        if not result.wasStrictlySuccessful():
 
3438
            self.fail(output.getvalue())
 
3439
 
 
3440
    def assertDocTestStringFails(self, klass, string):
 
3441
        result, output = self.run_doctest_suite_for_string(klass, string)
 
3442
        if result.wasStrictlySuccessful():
 
3443
            self.fail(output.getvalue())
 
3444
 
 
3445
    def test_injected_variable(self):
 
3446
        self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
 
3447
        test = """
 
3448
            >>> import os
 
3449
            >>> os.environ['LINES']
 
3450
            '42'
 
3451
            """
 
3452
        # doctest.DocTestSuite fails as it sees '25'
 
3453
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
 
3454
        # tests.DocTestSuite sees '42'
 
3455
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
 
3456
 
 
3457
    def test_deleted_variable(self):
 
3458
        self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
 
3459
        test = """
 
3460
            >>> import os
 
3461
            >>> os.environ.get('LINES')
 
3462
            """
 
3463
        # doctest.DocTestSuite fails as it sees '25'
 
3464
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
 
3465
        # tests.DocTestSuite sees None
 
3466
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
 
3467
 
 
3468
 
 
3469
class TestSelftestExcludePatterns(tests.TestCase):
 
3470
 
 
3471
    def setUp(self):
 
3472
        super(TestSelftestExcludePatterns, self).setUp()
 
3473
        self.overrideAttr(tests, 'test_suite', self.suite_factory)
 
3474
 
 
3475
    def suite_factory(self, keep_only=None, starting_with=None):
 
3476
        """A test suite factory with only a few tests."""
 
3477
        class Test(tests.TestCase):
 
3478
            def id(self):
 
3479
                # We don't need the full class path
 
3480
                return self._testMethodName
 
3481
            def a(self):
 
3482
                pass
 
3483
            def b(self):
 
3484
                pass
 
3485
            def c(self):
 
3486
                pass
 
3487
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
 
3488
 
 
3489
    def assertTestList(self, expected, *selftest_args):
 
3490
        # We rely on setUp installing the right test suite factory so we can
 
3491
        # test at the command level without loading the whole test suite
 
3492
        out, err = self.run_bzr(('selftest', '--list') + selftest_args)
 
3493
        actual = out.splitlines()
 
3494
        self.assertEquals(expected, actual)
 
3495
 
 
3496
    def test_full_list(self):
 
3497
        self.assertTestList(['a', 'b', 'c'])
 
3498
 
 
3499
    def test_single_exclude(self):
 
3500
        self.assertTestList(['b', 'c'], '-x', 'a')
 
3501
 
 
3502
    def test_mutiple_excludes(self):
 
3503
        self.assertTestList(['c'], '-x', 'a', '-x', 'b')
 
3504
 
 
3505
 
 
3506
class TestCounterHooks(tests.TestCase, SelfTestHelper):
 
3507
 
 
3508
    _test_needs_features = [features.subunit]
 
3509
 
 
3510
    def setUp(self):
 
3511
        super(TestCounterHooks, self).setUp()
 
3512
        class Test(tests.TestCase):
 
3513
 
 
3514
            def setUp(self):
 
3515
                super(Test, self).setUp()
 
3516
                self.hooks = hooks.Hooks()
 
3517
                self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
 
3518
                self.install_counter_hook(self.hooks, 'myhook')
 
3519
 
 
3520
            def no_hook(self):
 
3521
                pass
 
3522
 
 
3523
            def run_hook_once(self):
 
3524
                for hook in self.hooks['myhook']:
 
3525
                    hook(self)
 
3526
 
 
3527
        self.test_class = Test
 
3528
 
 
3529
    def assertHookCalls(self, expected_calls, test_name):
 
3530
        test = self.test_class(test_name)
 
3531
        result = unittest.TestResult()
 
3532
        test.run(result)
 
3533
        self.assertTrue(hasattr(test, '_counters'))
 
3534
        self.assertTrue(test._counters.has_key('myhook'))
 
3535
        self.assertEquals(expected_calls, test._counters['myhook'])
 
3536
 
 
3537
    def test_no_hook(self):
 
3538
        self.assertHookCalls(0, 'no_hook')
 
3539
 
 
3540
    def test_run_hook_once(self):
 
3541
        tt = features.testtools
 
3542
        if tt.module.__version__ < (0, 9, 8):
 
3543
            raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
 
3544
        self.assertHookCalls(1, 'run_hook_once')