← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

Move all features to bzrlib.tests.features in 2.5

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_selftest.py
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
    )
 
33
from testtools.content import Content
29
34
from testtools.content_type import ContentType
30
35
from testtools.matchers import (
31
36
    DocTestMatches,
32
37
    Equals,
33
38
    )
34
 
import testtools.tests.helpers
 
39
import testtools.testresult.doubles
35
40
 
36
41
import bzrlib
37
42
from bzrlib import (
38
43
    branchbuilder,
39
44
    bzrdir,
40
 
    debug,
41
45
    errors,
 
46
    hooks,
42
47
    lockdir,
43
48
    memorytree,
44
49
    osutils,
45
 
    progress,
46
50
    remote,
47
51
    repository,
48
52
    symbol_versioning,
49
53
    tests,
50
54
    transport,
51
55
    workingtree,
 
56
    workingtree_3,
 
57
    workingtree_4,
52
58
    )
53
59
from bzrlib.repofmt import (
54
60
    groupcompress_repo,
55
 
    pack_repo,
56
 
    weaverepo,
57
61
    )
58
62
from bzrlib.symbol_versioning import (
59
63
    deprecated_function,
64
68
    features,
65
69
    test_lsprof,
66
70
    test_server,
67
 
    test_sftp_transport,
68
71
    TestUtil,
69
72
    )
70
 
from bzrlib.trace import note
 
73
from bzrlib.trace import note, mutter
71
74
from bzrlib.transport import memory
72
 
from bzrlib.version import _get_bzr_source_tree
73
75
 
74
76
 
75
77
def _test_ids(test_suite):
77
79
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
78
80
 
79
81
 
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
82
class MetaTestLog(tests.TestCase):
93
83
 
94
84
    def test_logging(self):
100
90
            "text", "plain", {"charset": "utf8"})))
101
91
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
102
92
        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()
 
93
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
113
94
 
114
95
 
115
96
class TestTreeShape(tests.TestCaseInTempDir):
116
97
 
117
98
    def test_unicode_paths(self):
118
 
        self.requireFeature(tests.UnicodeFilename)
 
99
        self.requireFeature(features.UnicodeFilename)
119
100
 
120
101
        filename = u'hell\u00d8'
121
102
        self.build_tree_contents([(filename, 'contents of hello')])
122
 
        self.failUnlessExists(filename)
 
103
        self.assertPathExists(filename)
 
104
 
 
105
 
 
106
class TestClassesAvailable(tests.TestCase):
 
107
    """As a convenience we expose Test* classes from bzrlib.tests"""
 
108
 
 
109
    def test_test_case(self):
 
110
        from bzrlib.tests import TestCase
 
111
 
 
112
    def test_test_loader(self):
 
113
        from bzrlib.tests import TestLoader
 
114
 
 
115
    def test_test_suite(self):
 
116
        from bzrlib.tests import TestSuite
123
117
 
124
118
 
125
119
class TestTransportScenarios(tests.TestCase):
208
202
    def test_scenarios(self):
209
203
        # check that constructor parameters are passed through to the adapted
210
204
        # test.
211
 
        from bzrlib.tests.per_bzrdir import make_scenarios
 
205
        from bzrlib.tests.per_controldir import make_scenarios
212
206
        vfs_factory = "v"
213
207
        server1 = "a"
214
208
        server2 = "b"
312
306
        from bzrlib.tests.per_interrepository import make_scenarios
313
307
        server1 = "a"
314
308
        server2 = "b"
315
 
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
 
309
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
316
310
        scenarios = make_scenarios(server1, server2, formats)
317
311
        self.assertEqual([
318
312
            ('C0,str,str',
319
313
             {'repository_format': 'C1',
320
314
              'repository_format_to': 'C2',
321
315
              'transport_readonly_server': 'b',
322
 
              'transport_server': 'a'}),
 
316
              'transport_server': 'a',
 
317
              'extra_setup': 'C3'}),
323
318
            ('D0,str,str',
324
319
             {'repository_format': 'D1',
325
320
              'repository_format_to': 'D2',
326
321
              'transport_readonly_server': 'b',
327
 
              'transport_server': 'a'})],
 
322
              'transport_server': 'a',
 
323
              'extra_setup': 'D3'})],
328
324
            scenarios)
329
325
 
330
326
 
336
332
        from bzrlib.tests.per_workingtree import make_scenarios
337
333
        server1 = "a"
338
334
        server2 = "b"
339
 
        formats = [workingtree.WorkingTreeFormat2(),
340
 
                   workingtree.WorkingTreeFormat3(),]
 
335
        formats = [workingtree_4.WorkingTreeFormat4(),
 
336
                   workingtree_3.WorkingTreeFormat3(),]
341
337
        scenarios = make_scenarios(server1, server2, formats)
342
338
        self.assertEqual([
343
 
            ('WorkingTreeFormat2',
 
339
            ('WorkingTreeFormat4',
344
340
             {'bzrdir_format': formats[0]._matchingbzrdir,
345
341
              'transport_readonly_server': 'b',
346
342
              'transport_server': 'a',
373
369
            )
374
370
        server1 = "a"
375
371
        server2 = "b"
376
 
        formats = [workingtree.WorkingTreeFormat2(),
377
 
                   workingtree.WorkingTreeFormat3(),]
 
372
        formats = [workingtree_4.WorkingTreeFormat4(),
 
373
                   workingtree_3.WorkingTreeFormat3(),]
378
374
        scenarios = make_scenarios(server1, server2, formats)
379
375
        self.assertEqual(7, len(scenarios))
380
 
        default_wt_format = workingtree.WorkingTreeFormat4._default_format
381
 
        wt4_format = workingtree.WorkingTreeFormat4()
382
 
        wt5_format = workingtree.WorkingTreeFormat5()
 
376
        default_wt_format = workingtree.format_registry.get_default()
 
377
        wt4_format = workingtree_4.WorkingTreeFormat4()
 
378
        wt5_format = workingtree_4.WorkingTreeFormat5()
383
379
        expected_scenarios = [
384
 
            ('WorkingTreeFormat2',
 
380
            ('WorkingTreeFormat4',
385
381
             {'bzrdir_format': formats[0]._matchingbzrdir,
386
382
              'transport_readonly_server': 'b',
387
383
              'transport_server': 'a',
447
443
        # ones to add.
448
444
        from bzrlib.tests.per_tree import (
449
445
            return_parameter,
450
 
            revision_tree_from_workingtree
451
446
            )
452
447
        from bzrlib.tests.per_intertree import (
453
448
            make_scenarios,
454
449
            )
455
 
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
 
450
        from bzrlib.workingtree_3 import WorkingTreeFormat3
 
451
        from bzrlib.workingtree_4 import WorkingTreeFormat4
456
452
        input_test = TestInterTreeScenarios(
457
453
            "test_scenarios")
458
454
        server1 = "a"
459
455
        server2 = "b"
460
 
        format1 = WorkingTreeFormat2()
 
456
        format1 = WorkingTreeFormat4()
461
457
        format2 = WorkingTreeFormat3()
462
458
        formats = [("1", str, format1, format2, "converter1"),
463
459
            ("2", int, format2, format1, "converter2")]
509
505
        self.assertRaises(AssertionError, self.assertEqualStat,
510
506
            os.lstat("foo"), os.lstat("longname"))
511
507
 
 
508
    def test_failUnlessExists(self):
 
509
        """Deprecated failUnlessExists and failIfExists"""
 
510
        self.applyDeprecated(
 
511
            deprecated_in((2, 4)),
 
512
            self.failUnlessExists, '.')
 
513
        self.build_tree(['foo/', 'foo/bar'])
 
514
        self.applyDeprecated(
 
515
            deprecated_in((2, 4)),
 
516
            self.failUnlessExists, 'foo/bar')
 
517
        self.applyDeprecated(
 
518
            deprecated_in((2, 4)),
 
519
            self.failIfExists, 'foo/foo')
 
520
 
 
521
    def test_assertPathExists(self):
 
522
        self.assertPathExists('.')
 
523
        self.build_tree(['foo/', 'foo/bar'])
 
524
        self.assertPathExists('foo/bar')
 
525
        self.assertPathDoesNotExist('foo/foo')
 
526
 
512
527
 
513
528
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
514
529
 
548
563
        tree = self.make_branch_and_memory_tree('dir')
549
564
        # Guard against regression into MemoryTransport leaking
550
565
        # files to disk instead of keeping them in memory.
551
 
        self.failIf(osutils.lexists('dir'))
 
566
        self.assertFalse(osutils.lexists('dir'))
552
567
        self.assertIsInstance(tree, memorytree.MemoryTree)
553
568
 
554
569
    def test_make_branch_and_memory_tree_with_format(self):
555
570
        """make_branch_and_memory_tree should accept a format option."""
556
571
        format = bzrdir.BzrDirMetaFormat1()
557
 
        format.repository_format = weaverepo.RepositoryFormat7()
 
572
        format.repository_format = repository.format_registry.get_default()
558
573
        tree = self.make_branch_and_memory_tree('dir', format=format)
559
574
        # Guard against regression into MemoryTransport leaking
560
575
        # files to disk instead of keeping them in memory.
561
 
        self.failIf(osutils.lexists('dir'))
 
576
        self.assertFalse(osutils.lexists('dir'))
562
577
        self.assertIsInstance(tree, memorytree.MemoryTree)
563
578
        self.assertEqual(format.repository_format.__class__,
564
579
            tree.branch.repository._format.__class__)
568
583
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
569
584
        # Guard against regression into MemoryTransport leaking
570
585
        # files to disk instead of keeping them in memory.
571
 
        self.failIf(osutils.lexists('dir'))
 
586
        self.assertFalse(osutils.lexists('dir'))
572
587
 
573
588
    def test_make_branch_builder_with_format(self):
574
589
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
575
590
        # that the format objects are used.
576
591
        format = bzrdir.BzrDirMetaFormat1()
577
 
        repo_format = weaverepo.RepositoryFormat7()
 
592
        repo_format = repository.format_registry.get_default()
578
593
        format.repository_format = repo_format
579
594
        builder = self.make_branch_builder('dir', format=format)
580
595
        the_branch = builder.get_branch()
581
596
        # Guard against regression into MemoryTransport leaking
582
597
        # files to disk instead of keeping them in memory.
583
 
        self.failIf(osutils.lexists('dir'))
 
598
        self.assertFalse(osutils.lexists('dir'))
584
599
        self.assertEqual(format.repository_format.__class__,
585
600
                         the_branch.repository._format.__class__)
586
601
        self.assertEqual(repo_format.get_format_string(),
592
607
        the_branch = builder.get_branch()
593
608
        # Guard against regression into MemoryTransport leaking
594
609
        # files to disk instead of keeping them in memory.
595
 
        self.failIf(osutils.lexists('dir'))
 
610
        self.assertFalse(osutils.lexists('dir'))
596
611
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
597
612
        self.assertEqual(dir_format.repository_format.__class__,
598
613
                         the_branch.repository._format.__class__)
609
624
                l.attempt_lock()
610
625
        test = TestDanglingLock('test_function')
611
626
        result = test.run()
 
627
        total_failures = result.errors + result.failures
612
628
        if self._lock_check_thorough:
613
 
            self.assertEqual(1, len(result.errors))
 
629
            self.assertEqual(1, len(total_failures))
614
630
        else:
615
631
            # When _lock_check_thorough is disabled, then we don't trigger a
616
632
            # failure
617
 
            self.assertEqual(0, len(result.errors))
 
633
            self.assertEqual(0, len(total_failures))
618
634
 
619
635
 
620
636
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
621
637
    """Tests for the convenience functions TestCaseWithTransport introduces."""
622
638
 
623
639
    def test_get_readonly_url_none(self):
624
 
        from bzrlib.transport import get_transport
625
640
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
626
641
        self.vfs_transport_factory = memory.MemoryServer
627
642
        self.transport_readonly_server = None
629
644
        # for the server
630
645
        url = self.get_readonly_url()
631
646
        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))
 
647
        t = transport.get_transport(url)
 
648
        t2 = transport.get_transport(url2)
 
649
        self.assertIsInstance(t, ReadonlyTransportDecorator)
 
650
        self.assertIsInstance(t2, ReadonlyTransportDecorator)
636
651
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
637
652
 
638
653
    def test_get_readonly_url_http(self):
639
654
        from bzrlib.tests.http_server import HttpServer
640
 
        from bzrlib.transport import get_transport
641
655
        from bzrlib.transport.http import HttpTransportBase
642
656
        self.transport_server = test_server.LocalURLServer
643
657
        self.transport_readonly_server = HttpServer
645
659
        url = self.get_readonly_url()
646
660
        url2 = self.get_readonly_url('foo/bar')
647
661
        # 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))
 
662
        t = transport.get_transport(url)
 
663
        t2 = transport.get_transport(url2)
 
664
        self.assertIsInstance(t, HttpTransportBase)
 
665
        self.assertIsInstance(t2, HttpTransportBase)
652
666
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
653
667
 
654
668
    def test_is_directory(self):
662
676
    def test_make_branch_builder(self):
663
677
        builder = self.make_branch_builder('dir')
664
678
        rev_id = builder.build_commit()
665
 
        self.failUnlessExists('dir')
 
679
        self.assertPathExists('dir')
666
680
        a_dir = bzrdir.BzrDir.open('dir')
667
681
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
668
682
        a_branch = a_dir.open_branch()
684
698
        self.assertIsInstance(result_bzrdir.transport,
685
699
                              memory.MemoryTransport)
686
700
        # should not be on disk, should only be in memory
687
 
        self.failIfExists('subdir')
 
701
        self.assertPathDoesNotExist('subdir')
688
702
 
689
703
 
690
704
class TestChrootedTest(tests.ChrootedTestCase):
691
705
 
692
706
    def test_root_is_root(self):
693
 
        from bzrlib.transport import get_transport
694
 
        t = get_transport(self.get_readonly_url())
 
707
        t = transport.get_transport(self.get_readonly_url())
695
708
        url = t.base
696
709
        self.assertEqual(url, t.clone('..').base)
697
710
 
699
712
class TestProfileResult(tests.TestCase):
700
713
 
701
714
    def test_profiles_tests(self):
702
 
        self.requireFeature(test_lsprof.LSProfFeature)
703
 
        terminal = testtools.tests.helpers.ExtendedTestResult()
 
715
        self.requireFeature(features.lsprof_feature)
 
716
        terminal = testtools.testresult.doubles.ExtendedTestResult()
704
717
        result = tests.ProfileResult(terminal)
705
718
        class Sample(tests.TestCase):
706
719
            def a(self):
723
736
                descriptions=0,
724
737
                verbosity=1,
725
738
                )
726
 
        capture = testtools.tests.helpers.ExtendedTestResult()
 
739
        capture = testtools.testresult.doubles.ExtendedTestResult()
727
740
        test_case.run(MultiTestResult(result, capture))
728
741
        run_case = capture._events[0][1]
729
742
        timed_string = result._testTimeString(run_case)
750
763
        self.check_timing(ShortDelayTestCase('test_short_delay'),
751
764
                          r"^ +[0-9]+ms$")
752
765
 
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
766
    def _time_hello_world_encoding(self):
794
767
        """Profile two sleep calls
795
768
 
800
773
 
801
774
    def test_lsprofiling(self):
802
775
        """Verbose test result prints lsprof statistics from test cases."""
803
 
        self.requireFeature(test_lsprof.LSProfFeature)
 
776
        self.requireFeature(features.lsprof_feature)
804
777
        result_stream = StringIO()
805
778
        result = bzrlib.tests.VerboseTestResult(
806
 
            unittest._WritelnDecorator(result_stream),
 
779
            result_stream,
807
780
            descriptions=0,
808
781
            verbosity=2,
809
782
            )
835
808
        self.assertContainsRe(output,
836
809
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
837
810
 
 
811
    def test_uses_time_from_testtools(self):
 
812
        """Test case timings in verbose results should use testtools times"""
 
813
        import datetime
 
814
        class TimeAddedVerboseTestResult(tests.VerboseTestResult):
 
815
            def startTest(self, test):
 
816
                self.time(datetime.datetime.utcfromtimestamp(1.145))
 
817
                super(TimeAddedVerboseTestResult, self).startTest(test)
 
818
            def addSuccess(self, test):
 
819
                self.time(datetime.datetime.utcfromtimestamp(51.147))
 
820
                super(TimeAddedVerboseTestResult, self).addSuccess(test)
 
821
            def report_tests_starting(self): pass
 
822
        sio = StringIO()
 
823
        self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
 
824
        self.assertEndsWith(sio.getvalue(), "OK    50002ms\n")
 
825
 
838
826
    def test_known_failure(self):
839
827
        """A KnownFailure being raised should trigger several result actions."""
840
828
        class InstrumentedTestResult(tests.ExtendedTestResult):
841
829
            def stopTestRun(self): pass
842
 
            def startTests(self): pass
843
 
            def report_test_start(self, test): pass
 
830
            def report_tests_starting(self): pass
844
831
            def report_known_failure(self, test, err=None, details=None):
845
832
                self._call = test, 'known failure'
846
833
        result = InstrumentedTestResult(None, None, None, None)
864
851
        # verbose test output formatting
865
852
        result_stream = StringIO()
866
853
        result = bzrlib.tests.VerboseTestResult(
867
 
            unittest._WritelnDecorator(result_stream),
 
854
            result_stream,
868
855
            descriptions=0,
869
856
            verbosity=2,
870
857
            )
880
867
        output = result_stream.getvalue()[prefix:]
881
868
        lines = output.splitlines()
882
869
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
 
870
        if sys.version_info > (2, 7):
 
871
            self.expectFailure("_ExpectedFailure on 2.7 loses the message",
 
872
                self.assertNotEqual, lines[1], '    ')
883
873
        self.assertEqual(lines[1], '    foo')
884
874
        self.assertEqual(2, len(lines))
885
875
 
893
883
        """Test the behaviour of invoking addNotSupported."""
894
884
        class InstrumentedTestResult(tests.ExtendedTestResult):
895
885
            def stopTestRun(self): pass
896
 
            def startTests(self): pass
897
 
            def report_test_start(self, test): pass
 
886
            def report_tests_starting(self): pass
898
887
            def report_unsupported(self, test, feature):
899
888
                self._call = test, feature
900
889
        result = InstrumentedTestResult(None, None, None, None)
901
890
        test = SampleTestCase('_test_pass')
902
 
        feature = tests.Feature()
 
891
        feature = features.Feature()
903
892
        result.startTest(test)
904
893
        result.addNotSupported(test, feature)
905
894
        # it should invoke 'report_unsupported'.
919
908
        # verbose test output formatting
920
909
        result_stream = StringIO()
921
910
        result = bzrlib.tests.VerboseTestResult(
922
 
            unittest._WritelnDecorator(result_stream),
 
911
            result_stream,
923
912
            descriptions=0,
924
913
            verbosity=2,
925
914
            )
926
915
        test = self.get_passing_test()
927
 
        feature = tests.Feature()
 
916
        feature = features.Feature()
928
917
        result.startTest(test)
929
918
        prefix = len(result_stream.getvalue())
930
919
        result.report_unsupported(test, feature)
939
928
        """An UnavailableFeature being raised should invoke addNotSupported."""
940
929
        class InstrumentedTestResult(tests.ExtendedTestResult):
941
930
            def stopTestRun(self): pass
942
 
            def startTests(self): pass
943
 
            def report_test_start(self, test): pass
 
931
            def report_tests_starting(self): pass
944
932
            def addNotSupported(self, test, feature):
945
933
                self._call = test, feature
946
934
        result = InstrumentedTestResult(None, None, None, None)
947
 
        feature = tests.Feature()
 
935
        feature = features.Feature()
948
936
        class Test(tests.TestCase):
949
937
            def test_function(self):
950
938
                raise tests.UnavailableFeature(feature)
988
976
        class InstrumentedTestResult(tests.ExtendedTestResult):
989
977
            calls = 0
990
978
            def startTests(self): self.calls += 1
991
 
            def report_test_start(self, test): pass
992
979
        result = InstrumentedTestResult(None, None, None, None)
993
980
        def test_function():
994
981
            pass
996
983
        test.run(result)
997
984
        self.assertEquals(1, result.calls)
998
985
 
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()
 
986
    def test_startTests_only_once(self):
 
987
        """With multiple tests startTests should still only be called once"""
 
988
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
989
            calls = 0
 
990
            def startTests(self): self.calls += 1
 
991
        result = InstrumentedTestResult(None, None, None, None)
 
992
        suite = unittest.TestSuite([
 
993
            unittest.FunctionTestCase(lambda: None),
 
994
            unittest.FunctionTestCase(lambda: None)])
 
995
        suite.run(result)
 
996
        self.assertEquals(1, result.calls)
 
997
        self.assertEquals(2, result.count)
1007
998
 
1008
999
 
1009
1000
class TestRunner(tests.TestCase):
1022
1013
        because of our use of global state.
1023
1014
        """
1024
1015
        old_root = tests.TestCaseInTempDir.TEST_ROOT
1025
 
        old_leak = tests.TestCase._first_thread_leaker_id
1026
1016
        try:
1027
1017
            tests.TestCaseInTempDir.TEST_ROOT = None
1028
 
            tests.TestCase._first_thread_leaker_id = None
1029
1018
            return testrunner.run(test)
1030
1019
        finally:
1031
1020
            tests.TestCaseInTempDir.TEST_ROOT = old_root
1032
 
            tests.TestCase._first_thread_leaker_id = old_leak
1033
1021
 
1034
1022
    def test_known_failure_failed_run(self):
1035
1023
        # run a test that generates a known failure which should be printed in
1040
1028
        test = unittest.TestSuite()
1041
1029
        test.addTest(Test("known_failure_test"))
1042
1030
        def failing_test():
1043
 
            self.fail('foo')
 
1031
            raise AssertionError('foo')
1044
1032
        test.addTest(unittest.FunctionTestCase(failing_test))
1045
1033
        stream = StringIO()
1046
1034
        runner = tests.TextTestRunner(stream=stream)
1054
1042
            '^----------------------------------------------------------------------\n'
1055
1043
            'Traceback \\(most recent call last\\):\n'
1056
1044
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1057
 
            '    self.fail\\(\'foo\'\\)\n'
 
1045
            '    raise AssertionError\\(\'foo\'\\)\n'
1058
1046
            '.*'
1059
1047
            '^----------------------------------------------------------------------\n'
1060
1048
            '.*'
1066
1054
        # the final output.
1067
1055
        class Test(tests.TestCase):
1068
1056
            def known_failure_test(self):
1069
 
                self.expectFailure('failed', self.assertTrue, False)
 
1057
                self.knownFailure("Never works...")
1070
1058
        test = Test("known_failure_test")
1071
1059
        stream = StringIO()
1072
1060
        runner = tests.TextTestRunner(stream=stream)
1078
1066
            '\n'
1079
1067
            'OK \\(known_failures=1\\)\n')
1080
1068
 
 
1069
    def test_unexpected_success_bad(self):
 
1070
        class Test(tests.TestCase):
 
1071
            def test_truth(self):
 
1072
                self.expectFailure("No absolute truth", self.assertTrue, True)
 
1073
        runner = tests.TextTestRunner(stream=StringIO())
 
1074
        result = self.run_test_runner(runner, Test("test_truth"))
 
1075
        self.assertContainsRe(runner.stream.getvalue(),
 
1076
            "=+\n"
 
1077
            "FAIL: \\S+\.test_truth\n"
 
1078
            "-+\n"
 
1079
            "(?:.*\n)*"
 
1080
            "No absolute truth\n"
 
1081
            "(?:.*\n)*"
 
1082
            "-+\n"
 
1083
            "Ran 1 test in .*\n"
 
1084
            "\n"
 
1085
            "FAILED \\(failures=1\\)\n\\Z")
 
1086
 
1081
1087
    def test_result_decorator(self):
1082
1088
        # decorate results
1083
1089
        calls = []
1084
 
        class LoggingDecorator(tests.ForwardingResult):
 
1090
        class LoggingDecorator(ExtendedToOriginalDecorator):
1085
1091
            def startTest(self, test):
1086
 
                tests.ForwardingResult.startTest(self, test)
 
1092
                ExtendedToOriginalDecorator.startTest(self, test)
1087
1093
                calls.append('start')
1088
1094
        test = unittest.FunctionTestCase(lambda:None)
1089
1095
        stream = StringIO()
1167
1173
 
1168
1174
    def test_unsupported_features_listed(self):
1169
1175
        """When unsupported features are encountered they are detailed."""
1170
 
        class Feature1(tests.Feature):
 
1176
        class Feature1(features.Feature):
1171
1177
            def _probe(self): return False
1172
 
        class Feature2(tests.Feature):
 
1178
        class Feature2(features.Feature):
1173
1179
            def _probe(self): return False
1174
1180
        # create sample tests
1175
1181
        test1 = SampleTestCase('_test_pass')
1190
1196
            ],
1191
1197
            lines[-3:])
1192
1198
 
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)
 
1199
    def test_verbose_test_count(self):
 
1200
        """A verbose test run reports the right test count at the start"""
 
1201
        suite = TestUtil.TestSuite([
 
1202
            unittest.FunctionTestCase(lambda:None),
 
1203
            unittest.FunctionTestCase(lambda:None)])
 
1204
        self.assertEqual(suite.countTestCases(), 2)
 
1205
        stream = StringIO()
 
1206
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
 
1207
        # Need to use the CountingDecorator as that's what sets num_tests
 
1208
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
 
1209
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
1215
1210
 
1216
1211
    def test_startTestRun(self):
1217
1212
        """run should call result.startTestRun()"""
1218
1213
        calls = []
1219
 
        class LoggingDecorator(tests.ForwardingResult):
 
1214
        class LoggingDecorator(ExtendedToOriginalDecorator):
1220
1215
            def startTestRun(self):
1221
 
                tests.ForwardingResult.startTestRun(self)
 
1216
                ExtendedToOriginalDecorator.startTestRun(self)
1222
1217
                calls.append('startTestRun')
1223
1218
        test = unittest.FunctionTestCase(lambda:None)
1224
1219
        stream = StringIO()
1230
1225
    def test_stopTestRun(self):
1231
1226
        """run should call result.stopTestRun()"""
1232
1227
        calls = []
1233
 
        class LoggingDecorator(tests.ForwardingResult):
 
1228
        class LoggingDecorator(ExtendedToOriginalDecorator):
1234
1229
            def stopTestRun(self):
1235
 
                tests.ForwardingResult.stopTestRun(self)
 
1230
                ExtendedToOriginalDecorator.stopTestRun(self)
1236
1231
                calls.append('stopTestRun')
1237
1232
        test = unittest.FunctionTestCase(lambda:None)
1238
1233
        stream = StringIO()
1241
1236
        result = self.run_test_runner(runner, test)
1242
1237
        self.assertLength(1, calls)
1243
1238
 
 
1239
    def test_unicode_test_output_on_ascii_stream(self):
 
1240
        """Showing results should always succeed even on an ascii console"""
 
1241
        class FailureWithUnicode(tests.TestCase):
 
1242
            def test_log_unicode(self):
 
1243
                self.log(u"\u2606")
 
1244
                self.fail("Now print that log!")
 
1245
        out = StringIO()
 
1246
        self.overrideAttr(osutils, "get_terminal_encoding",
 
1247
            lambda trace=False: "ascii")
 
1248
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
 
1249
            FailureWithUnicode("test_log_unicode"))
 
1250
        self.assertContainsRe(out.getvalue(),
 
1251
            "Text attachment: log\n"
 
1252
            "-+\n"
 
1253
            "\d+\.\d+  \\\\u2606\n"
 
1254
            "-+\n")
 
1255
 
1244
1256
 
1245
1257
class SampleTestCase(tests.TestCase):
1246
1258
 
1421
1433
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1422
1434
        output_stream = StringIO()
1423
1435
        result = bzrlib.tests.VerboseTestResult(
1424
 
            unittest._WritelnDecorator(output_stream),
 
1436
            output_stream,
1425
1437
            descriptions=0,
1426
1438
            verbosity=2)
1427
1439
        sample_test.run(result)
1434
1446
        # Note this test won't fail with hooks that the core library doesn't
1435
1447
        # use - but it trigger with a plugin that adds hooks, so its still a
1436
1448
        # useful warning in that case.
1437
 
        self.assertEqual(bzrlib.branch.BranchHooks(),
1438
 
            bzrlib.branch.Branch.hooks)
1439
 
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
 
1449
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
 
1450
        self.assertEqual(
 
1451
            bzrlib.smart.server.SmartServerHooks(),
1440
1452
            bzrlib.smart.server.SmartTCPServer.hooks)
1441
 
        self.assertEqual(bzrlib.commands.CommandHooks(),
1442
 
            bzrlib.commands.Command.hooks)
 
1453
        self.assertEqual(
 
1454
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
1443
1455
 
1444
1456
    def test__gather_lsprof_in_benchmarks(self):
1445
1457
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1446
1458
 
1447
1459
        Each self.time() call is individually and separately profiled.
1448
1460
        """
1449
 
        self.requireFeature(test_lsprof.LSProfFeature)
 
1461
        self.requireFeature(features.lsprof_feature)
1450
1462
        # overrides the class member with an instance member so no cleanup
1451
1463
        # needed.
1452
1464
        self._gather_lsprof_in_benchmarks = True
1482
1494
 
1483
1495
    def test_requireFeature_available(self):
1484
1496
        """self.requireFeature(available) is a no-op."""
1485
 
        class Available(tests.Feature):
 
1497
        class Available(features.Feature):
1486
1498
            def _probe(self):return True
1487
1499
        feature = Available()
1488
1500
        self.requireFeature(feature)
1489
1501
 
1490
1502
    def test_requireFeature_unavailable(self):
1491
1503
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1492
 
        class Unavailable(tests.Feature):
 
1504
        class Unavailable(features.Feature):
1493
1505
            def _probe(self):return False
1494
1506
        feature = Unavailable()
1495
1507
        self.assertRaises(tests.UnavailableFeature,
1655
1667
        self.assertEqual('original', obj.test_attr)
1656
1668
 
1657
1669
 
 
1670
class _MissingFeature(features.Feature):
 
1671
    def _probe(self):
 
1672
        return False
 
1673
missing_feature = _MissingFeature()
 
1674
 
 
1675
 
 
1676
def _get_test(name):
 
1677
    """Get an instance of a specific example test.
 
1678
 
 
1679
    We protect this in a function so that they don't auto-run in the test
 
1680
    suite.
 
1681
    """
 
1682
 
 
1683
    class ExampleTests(tests.TestCase):
 
1684
 
 
1685
        def test_fail(self):
 
1686
            mutter('this was a failing test')
 
1687
            self.fail('this test will fail')
 
1688
 
 
1689
        def test_error(self):
 
1690
            mutter('this test errored')
 
1691
            raise RuntimeError('gotcha')
 
1692
 
 
1693
        def test_missing_feature(self):
 
1694
            mutter('missing the feature')
 
1695
            self.requireFeature(missing_feature)
 
1696
 
 
1697
        def test_skip(self):
 
1698
            mutter('this test will be skipped')
 
1699
            raise tests.TestSkipped('reason')
 
1700
 
 
1701
        def test_success(self):
 
1702
            mutter('this test succeeds')
 
1703
 
 
1704
        def test_xfail(self):
 
1705
            mutter('test with expected failure')
 
1706
            self.knownFailure('this_fails')
 
1707
 
 
1708
        def test_unexpected_success(self):
 
1709
            mutter('test with unexpected success')
 
1710
            self.expectFailure('should_fail', lambda: None)
 
1711
 
 
1712
    return ExampleTests(name)
 
1713
 
 
1714
 
 
1715
class TestTestCaseLogDetails(tests.TestCase):
 
1716
 
 
1717
    def _run_test(self, test_name):
 
1718
        test = _get_test(test_name)
 
1719
        result = testtools.TestResult()
 
1720
        test.run(result)
 
1721
        return result
 
1722
 
 
1723
    def test_fail_has_log(self):
 
1724
        result = self._run_test('test_fail')
 
1725
        self.assertEqual(1, len(result.failures))
 
1726
        result_content = result.failures[0][1]
 
1727
        self.assertContainsRe(result_content, 'Text attachment: log')
 
1728
        self.assertContainsRe(result_content, 'this was a failing test')
 
1729
 
 
1730
    def test_error_has_log(self):
 
1731
        result = self._run_test('test_error')
 
1732
        self.assertEqual(1, len(result.errors))
 
1733
        result_content = result.errors[0][1]
 
1734
        self.assertContainsRe(result_content, 'Text attachment: log')
 
1735
        self.assertContainsRe(result_content, 'this test errored')
 
1736
 
 
1737
    def test_skip_has_no_log(self):
 
1738
        result = self._run_test('test_skip')
 
1739
        self.assertEqual(['reason'], result.skip_reasons.keys())
 
1740
        skips = result.skip_reasons['reason']
 
1741
        self.assertEqual(1, len(skips))
 
1742
        test = skips[0]
 
1743
        self.assertFalse('log' in test.getDetails())
 
1744
 
 
1745
    def test_missing_feature_has_no_log(self):
 
1746
        # testtools doesn't know about addNotSupported, so it just gets
 
1747
        # considered as a skip
 
1748
        result = self._run_test('test_missing_feature')
 
1749
        self.assertEqual([missing_feature], result.skip_reasons.keys())
 
1750
        skips = result.skip_reasons[missing_feature]
 
1751
        self.assertEqual(1, len(skips))
 
1752
        test = skips[0]
 
1753
        self.assertFalse('log' in test.getDetails())
 
1754
 
 
1755
    def test_xfail_has_no_log(self):
 
1756
        result = self._run_test('test_xfail')
 
1757
        self.assertEqual(1, len(result.expectedFailures))
 
1758
        result_content = result.expectedFailures[0][1]
 
1759
        self.assertNotContainsRe(result_content, 'Text attachment: log')
 
1760
        self.assertNotContainsRe(result_content, 'test with expected failure')
 
1761
 
 
1762
    def test_unexpected_success_has_log(self):
 
1763
        result = self._run_test('test_unexpected_success')
 
1764
        self.assertEqual(1, len(result.unexpectedSuccesses))
 
1765
        # Inconsistency, unexpectedSuccesses is a list of tests,
 
1766
        # expectedFailures is a list of reasons?
 
1767
        test = result.unexpectedSuccesses[0]
 
1768
        details = test.getDetails()
 
1769
        self.assertTrue('log' in details)
 
1770
 
 
1771
 
 
1772
class TestTestCloning(tests.TestCase):
 
1773
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
 
1774
 
 
1775
    def test_cloned_testcase_does_not_share_details(self):
 
1776
        """A TestCase cloned with clone_test does not share mutable attributes
 
1777
        such as details or cleanups.
 
1778
        """
 
1779
        class Test(tests.TestCase):
 
1780
            def test_foo(self):
 
1781
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
 
1782
        orig_test = Test('test_foo')
 
1783
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
 
1784
        orig_test.run(unittest.TestResult())
 
1785
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
 
1786
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
 
1787
 
 
1788
    def test_double_apply_scenario_preserves_first_scenario(self):
 
1789
        """Applying two levels of scenarios to a test preserves the attributes
 
1790
        added by both scenarios.
 
1791
        """
 
1792
        class Test(tests.TestCase):
 
1793
            def test_foo(self):
 
1794
                pass
 
1795
        test = Test('test_foo')
 
1796
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
 
1797
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
 
1798
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
 
1799
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
 
1800
        all_tests = list(tests.iter_suite_tests(suite))
 
1801
        self.assertLength(4, all_tests)
 
1802
        all_xys = sorted((t.x, t.y) for t in all_tests)
 
1803
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
 
1804
 
 
1805
 
1658
1806
# NB: Don't delete this; it's not actually from 0.11!
1659
1807
@deprecated_function(deprecated_in((0, 11, 0)))
1660
1808
def sample_deprecated_function():
1787
1935
    def test_make_branch_and_tree_with_format(self):
1788
1936
        # we should be able to supply a format to make_branch_and_tree
1789
1937
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
1790
 
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
1791
1938
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
1792
1939
                              bzrlib.bzrdir.BzrDirMetaFormat1)
1793
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
1794
 
                              bzrlib.bzrdir.BzrDirFormat6)
1795
1940
 
1796
1941
    def test_make_branch_and_memory_tree(self):
1797
1942
        # we should be able to get a new branch and a mutable tree from
1814
1959
                tree.branch.repository.bzrdir.root_transport)
1815
1960
 
1816
1961
 
1817
 
class SelfTestHelper:
 
1962
class SelfTestHelper(object):
1818
1963
 
1819
1964
    def run_selftest(self, **kwargs):
1820
1965
        """Run selftest returning its output."""
1874
2019
        self.assertLength(2, output.readlines())
1875
2020
 
1876
2021
    def test_lsprof_tests(self):
1877
 
        self.requireFeature(test_lsprof.LSProfFeature)
1878
 
        calls = []
 
2022
        self.requireFeature(features.lsprof_feature)
 
2023
        results = []
1879
2024
        class Test(object):
1880
2025
            def __call__(test, result):
1881
2026
                test.run(result)
1882
2027
            def run(test, result):
1883
 
                self.assertIsInstance(result, tests.ForwardingResult)
1884
 
                calls.append("called")
 
2028
                results.append(result)
1885
2029
            def countTestCases(self):
1886
2030
                return 1
1887
2031
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
1888
 
        self.assertLength(1, calls)
 
2032
        self.assertLength(1, results)
 
2033
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
1889
2034
 
1890
2035
    def test_random(self):
1891
2036
        # test randomising by listing a number of tests.
1971
2116
            load_list='missing file name', list_only=True)
1972
2117
 
1973
2118
 
 
2119
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
 
2120
 
 
2121
    _test_needs_features = [features.subunit]
 
2122
 
 
2123
    def run_subunit_stream(self, test_name):
 
2124
        from subunit import ProtocolTestCase
 
2125
        def factory():
 
2126
            return TestUtil.TestSuite([_get_test(test_name)])
 
2127
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
 
2128
            test_suite_factory=factory)
 
2129
        test = ProtocolTestCase(stream)
 
2130
        result = testtools.TestResult()
 
2131
        test.run(result)
 
2132
        content = stream.getvalue()
 
2133
        return content, result
 
2134
 
 
2135
    def test_fail_has_log(self):
 
2136
        content, result = self.run_subunit_stream('test_fail')
 
2137
        self.assertEqual(1, len(result.failures))
 
2138
        self.assertContainsRe(content, '(?m)^log$')
 
2139
        self.assertContainsRe(content, 'this test will fail')
 
2140
 
 
2141
    def test_error_has_log(self):
 
2142
        content, result = self.run_subunit_stream('test_error')
 
2143
        self.assertContainsRe(content, '(?m)^log$')
 
2144
        self.assertContainsRe(content, 'this test errored')
 
2145
 
 
2146
    def test_skip_has_no_log(self):
 
2147
        content, result = self.run_subunit_stream('test_skip')
 
2148
        self.assertNotContainsRe(content, '(?m)^log$')
 
2149
        self.assertNotContainsRe(content, 'this test will be skipped')
 
2150
        self.assertEqual(['reason'], result.skip_reasons.keys())
 
2151
        skips = result.skip_reasons['reason']
 
2152
        self.assertEqual(1, len(skips))
 
2153
        test = skips[0]
 
2154
        # RemotedTestCase doesn't preserve the "details"
 
2155
        ## self.assertFalse('log' in test.getDetails())
 
2156
 
 
2157
    def test_missing_feature_has_no_log(self):
 
2158
        content, result = self.run_subunit_stream('test_missing_feature')
 
2159
        self.assertNotContainsRe(content, '(?m)^log$')
 
2160
        self.assertNotContainsRe(content, 'missing the feature')
 
2161
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
 
2162
        skips = result.skip_reasons['_MissingFeature\n']
 
2163
        self.assertEqual(1, len(skips))
 
2164
        test = skips[0]
 
2165
        # RemotedTestCase doesn't preserve the "details"
 
2166
        ## self.assertFalse('log' in test.getDetails())
 
2167
 
 
2168
    def test_xfail_has_no_log(self):
 
2169
        content, result = self.run_subunit_stream('test_xfail')
 
2170
        self.assertNotContainsRe(content, '(?m)^log$')
 
2171
        self.assertNotContainsRe(content, 'test with expected failure')
 
2172
        self.assertEqual(1, len(result.expectedFailures))
 
2173
        result_content = result.expectedFailures[0][1]
 
2174
        self.assertNotContainsRe(result_content, 'Text attachment: log')
 
2175
        self.assertNotContainsRe(result_content, 'test with expected failure')
 
2176
 
 
2177
    def test_unexpected_success_has_log(self):
 
2178
        content, result = self.run_subunit_stream('test_unexpected_success')
 
2179
        self.assertContainsRe(content, '(?m)^log$')
 
2180
        self.assertContainsRe(content, 'test with unexpected success')
 
2181
        # GZ 2011-05-18: Old versions of subunit treat unexpected success as a
 
2182
        #                success, if a min version check is added remove this
 
2183
        from subunit import TestProtocolClient as _Client
 
2184
        if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
 
2185
            self.expectFailure('subunit treats "unexpectedSuccess"'
 
2186
                               ' as a plain success',
 
2187
                self.assertEqual, 1, len(result.unexpectedSuccesses))
 
2188
        self.assertEqual(1, len(result.unexpectedSuccesses))
 
2189
        test = result.unexpectedSuccesses[0]
 
2190
        # RemotedTestCase doesn't preserve the "details"
 
2191
        ## self.assertTrue('log' in test.getDetails())
 
2192
 
 
2193
    def test_success_has_no_log(self):
 
2194
        content, result = self.run_subunit_stream('test_success')
 
2195
        self.assertEqual(1, result.testsRun)
 
2196
        self.assertNotContainsRe(content, '(?m)^log$')
 
2197
        self.assertNotContainsRe(content, 'this test succeeds')
 
2198
 
 
2199
 
1974
2200
class TestRunBzr(tests.TestCase):
1975
2201
 
1976
2202
    out = ''
2099
2325
        # stdout and stderr of the invoked run_bzr
2100
2326
        current_factory = bzrlib.ui.ui_factory
2101
2327
        self.run_bzr(['foo'])
2102
 
        self.failIf(current_factory is self.factory)
 
2328
        self.assertFalse(current_factory is self.factory)
2103
2329
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2104
2330
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2105
2331
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2262
2488
 
2263
2489
 
2264
2490
class TestStartBzrSubProcess(tests.TestCase):
 
2491
    """Stub test start_bzr_subprocess."""
2265
2492
 
2266
 
    def check_popen_state(self):
2267
 
        """Replace to make assertions when popen is called."""
 
2493
    def _subprocess_log_cleanup(self):
 
2494
        """Inhibits the base version as we don't produce a log file."""
2268
2495
 
2269
2496
    def _popen(self, *args, **kwargs):
2270
 
        """Record the command that is run, so that we can ensure it is correct"""
 
2497
        """Override the base version to record the command that is run.
 
2498
 
 
2499
        From there we can ensure it is correct without spawning a real process.
 
2500
        """
2271
2501
        self.check_popen_state()
2272
2502
        self._popen_args = args
2273
2503
        self._popen_kwargs = kwargs
2274
2504
        raise _DontSpawnProcess()
2275
2505
 
 
2506
    def check_popen_state(self):
 
2507
        """Replace to make assertions when popen is called."""
 
2508
 
2276
2509
    def test_run_bzr_subprocess_no_plugins(self):
2277
2510
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2278
2511
        command = self._popen_args[0]
2282
2515
 
2283
2516
    def test_allow_plugins(self):
2284
2517
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2285
 
            allow_plugins=True)
 
2518
                          allow_plugins=True)
2286
2519
        command = self._popen_args[0]
2287
2520
        self.assertEqual([], command[2:])
2288
2521
 
2289
2522
    def test_set_env(self):
2290
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2523
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2291
2524
        # set in the child
2292
2525
        def check_environment():
2293
2526
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2294
2527
        self.check_popen_state = check_environment
2295
2528
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2296
 
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
 
2529
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
2297
2530
        # not set in theparent
2298
2531
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2299
2532
 
2300
2533
    def test_run_bzr_subprocess_env_del(self):
2301
2534
        """run_bzr_subprocess can remove environment variables too."""
2302
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2535
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2303
2536
        def check_environment():
2304
2537
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2305
2538
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2306
2539
        self.check_popen_state = check_environment
2307
2540
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2308
 
            env_changes={'EXISTANT_ENV_VAR':None})
 
2541
                          env_changes={'EXISTANT_ENV_VAR':None})
2309
2542
        # Still set in parent
2310
2543
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2311
2544
        del os.environ['EXISTANT_ENV_VAR']
2312
2545
 
2313
2546
    def test_env_del_missing(self):
2314
 
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
 
2547
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2315
2548
        def check_environment():
2316
2549
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2317
2550
        self.check_popen_state = check_environment
2318
2551
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2319
 
            env_changes={'NON_EXISTANT_ENV_VAR':None})
 
2552
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
2320
2553
 
2321
2554
    def test_working_dir(self):
2322
2555
        """Test that we can specify the working dir for the child"""
2325
2558
        chdirs = []
2326
2559
        def chdir(path):
2327
2560
            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
 
2561
        self.overrideAttr(os, 'chdir', chdir)
 
2562
        def getcwd():
 
2563
            return 'current'
 
2564
        self.overrideAttr(osutils, 'getcwd', getcwd)
 
2565
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2566
                          working_dir='foo')
2340
2567
        self.assertEqual(['foo', 'current'], chdirs)
2341
2568
 
 
2569
    def test_get_bzr_path_with_cwd_bzrlib(self):
 
2570
        self.get_source_path = lambda: ""
 
2571
        self.overrideAttr(os.path, "isfile", lambda path: True)
 
2572
        self.assertEqual(self.get_bzr_path(), "bzr")
 
2573
 
2342
2574
 
2343
2575
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2344
2576
    """Tests that really need to do things with an external bzr."""
2357
2589
        self.assertEqual('bzr: interrupted\n', result[1])
2358
2590
 
2359
2591
 
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
2592
class TestSelftestFiltering(tests.TestCase):
2435
2593
 
2436
2594
    def setUp(self):
2587
2745
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2588
2746
 
2589
2747
 
2590
 
class TestCheckInventoryShape(tests.TestCaseWithTransport):
 
2748
class TestCheckTreeShape(tests.TestCaseWithTransport):
2591
2749
 
2592
 
    def test_check_inventory_shape(self):
 
2750
    def test_check_tree_shape(self):
2593
2751
        files = ['a', 'b/', 'b/c']
2594
2752
        tree = self.make_branch_and_tree('.')
2595
2753
        self.build_tree(files)
2596
2754
        tree.add(files)
2597
2755
        tree.lock_read()
2598
2756
        try:
2599
 
            self.check_inventory_shape(tree.inventory, files)
 
2757
            self.check_tree_shape(tree, files)
2600
2758
        finally:
2601
2759
            tree.unlock()
2602
2760
 
2934
3092
        tpr.register('bar', 'bBB.aAA.rRR')
2935
3093
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
2936
3094
        self.assertThat(self.get_log(),
2937
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
 
3095
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
 
3096
                           doctest.ELLIPSIS))
2938
3097
 
2939
3098
    def test_get_unknown_prefix(self):
2940
3099
        tpr = self._get_registry()
2960
3119
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
2961
3120
 
2962
3121
 
 
3122
class TestThreadLeakDetection(tests.TestCase):
 
3123
    """Ensure when tests leak threads we detect and report it"""
 
3124
 
 
3125
    class LeakRecordingResult(tests.ExtendedTestResult):
 
3126
        def __init__(self):
 
3127
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
 
3128
            self.leaks = []
 
3129
        def _report_thread_leak(self, test, leaks, alive):
 
3130
            self.leaks.append((test, leaks))
 
3131
 
 
3132
    def test_testcase_without_addCleanups(self):
 
3133
        """Check old TestCase instances don't break with leak detection"""
 
3134
        class Test(unittest.TestCase):
 
3135
            def runTest(self):
 
3136
                pass
 
3137
        result = self.LeakRecordingResult()
 
3138
        test = Test()
 
3139
        result.startTestRun()
 
3140
        test.run(result)
 
3141
        result.stopTestRun()
 
3142
        self.assertEqual(result._tests_leaking_threads_count, 0)
 
3143
        self.assertEqual(result.leaks, [])
 
3144
        
 
3145
    def test_thread_leak(self):
 
3146
        """Ensure a thread that outlives the running of a test is reported
 
3147
 
 
3148
        Uses a thread that blocks on an event, and is started by the inner
 
3149
        test case. As the thread outlives the inner case's run, it should be
 
3150
        detected as a leak, but the event is then set so that the thread can
 
3151
        be safely joined in cleanup so it's not leaked for real.
 
3152
        """
 
3153
        event = threading.Event()
 
3154
        thread = threading.Thread(name="Leaker", target=event.wait)
 
3155
        class Test(tests.TestCase):
 
3156
            def test_leak(self):
 
3157
                thread.start()
 
3158
        result = self.LeakRecordingResult()
 
3159
        test = Test("test_leak")
 
3160
        self.addCleanup(thread.join)
 
3161
        self.addCleanup(event.set)
 
3162
        result.startTestRun()
 
3163
        test.run(result)
 
3164
        result.stopTestRun()
 
3165
        self.assertEqual(result._tests_leaking_threads_count, 1)
 
3166
        self.assertEqual(result._first_thread_leaker_id, test.id())
 
3167
        self.assertEqual(result.leaks, [(test, set([thread]))])
 
3168
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
 
3169
 
 
3170
    def test_multiple_leaks(self):
 
3171
        """Check multiple leaks are blamed on the test cases at fault
 
3172
 
 
3173
        Same concept as the previous test, but has one inner test method that
 
3174
        leaks two threads, and one that doesn't leak at all.
 
3175
        """
 
3176
        event = threading.Event()
 
3177
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
 
3178
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
 
3179
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
 
3180
        class Test(tests.TestCase):
 
3181
            def test_first_leak(self):
 
3182
                thread_b.start()
 
3183
            def test_second_no_leak(self):
 
3184
                pass
 
3185
            def test_third_leak(self):
 
3186
                thread_c.start()
 
3187
                thread_a.start()
 
3188
        result = self.LeakRecordingResult()
 
3189
        first_test = Test("test_first_leak")
 
3190
        third_test = Test("test_third_leak")
 
3191
        self.addCleanup(thread_a.join)
 
3192
        self.addCleanup(thread_b.join)
 
3193
        self.addCleanup(thread_c.join)
 
3194
        self.addCleanup(event.set)
 
3195
        result.startTestRun()
 
3196
        unittest.TestSuite(
 
3197
            [first_test, Test("test_second_no_leak"), third_test]
 
3198
            ).run(result)
 
3199
        result.stopTestRun()
 
3200
        self.assertEqual(result._tests_leaking_threads_count, 2)
 
3201
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
 
3202
        self.assertEqual(result.leaks, [
 
3203
            (first_test, set([thread_b])),
 
3204
            (third_test, set([thread_a, thread_c]))])
 
3205
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
 
3206
 
 
3207
 
 
3208
class TestPostMortemDebugging(tests.TestCase):
 
3209
    """Check post mortem debugging works when tests fail or error"""
 
3210
 
 
3211
    class TracebackRecordingResult(tests.ExtendedTestResult):
 
3212
        def __init__(self):
 
3213
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
 
3214
            self.postcode = None
 
3215
        def _post_mortem(self, tb=None):
 
3216
            """Record the code object at the end of the current traceback"""
 
3217
            tb = tb or sys.exc_info()[2]
 
3218
            if tb is not None:
 
3219
                next = tb.tb_next
 
3220
                while next is not None:
 
3221
                    tb = next
 
3222
                    next = next.tb_next
 
3223
                self.postcode = tb.tb_frame.f_code
 
3224
        def report_error(self, test, err):
 
3225
            pass
 
3226
        def report_failure(self, test, err):
 
3227
            pass
 
3228
 
 
3229
    def test_location_unittest_error(self):
 
3230
        """Needs right post mortem traceback with erroring unittest case"""
 
3231
        class Test(unittest.TestCase):
 
3232
            def runTest(self):
 
3233
                raise RuntimeError
 
3234
        result = self.TracebackRecordingResult()
 
3235
        Test().run(result)
 
3236
        self.assertEqual(result.postcode, Test.runTest.func_code)
 
3237
 
 
3238
    def test_location_unittest_failure(self):
 
3239
        """Needs right post mortem traceback with failing unittest case"""
 
3240
        class Test(unittest.TestCase):
 
3241
            def runTest(self):
 
3242
                raise self.failureException
 
3243
        result = self.TracebackRecordingResult()
 
3244
        Test().run(result)
 
3245
        self.assertEqual(result.postcode, Test.runTest.func_code)
 
3246
 
 
3247
    def test_location_bt_error(self):
 
3248
        """Needs right post mortem traceback with erroring bzrlib.tests case"""
 
3249
        class Test(tests.TestCase):
 
3250
            def test_error(self):
 
3251
                raise RuntimeError
 
3252
        result = self.TracebackRecordingResult()
 
3253
        Test("test_error").run(result)
 
3254
        self.assertEqual(result.postcode, Test.test_error.func_code)
 
3255
 
 
3256
    def test_location_bt_failure(self):
 
3257
        """Needs right post mortem traceback with failing bzrlib.tests case"""
 
3258
        class Test(tests.TestCase):
 
3259
            def test_failure(self):
 
3260
                raise self.failureException
 
3261
        result = self.TracebackRecordingResult()
 
3262
        Test("test_failure").run(result)
 
3263
        self.assertEqual(result.postcode, Test.test_failure.func_code)
 
3264
 
 
3265
    def test_env_var_triggers_post_mortem(self):
 
3266
        """Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
 
3267
        import pdb
 
3268
        result = tests.ExtendedTestResult(StringIO(), 0, 1)
 
3269
        post_mortem_calls = []
 
3270
        self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
 
3271
        self.overrideEnv('BZR_TEST_PDB', None)
 
3272
        result._post_mortem(1)
 
3273
        self.overrideEnv('BZR_TEST_PDB', 'on')
 
3274
        result._post_mortem(2)
 
3275
        self.assertEqual([2], post_mortem_calls)
 
3276
 
 
3277
 
2963
3278
class TestRunSuite(tests.TestCase):
2964
3279
 
2965
3280
    def test_runner_class(self):
2976
3291
                                                self.verbosity)
2977
3292
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2978
3293
        self.assertLength(1, calls)
 
3294
 
 
3295
 
 
3296
class TestEnvironHandling(tests.TestCase):
 
3297
 
 
3298
    def test_overrideEnv_None_called_twice_doesnt_leak(self):
 
3299
        self.assertFalse('MYVAR' in os.environ)
 
3300
        self.overrideEnv('MYVAR', '42')
 
3301
        # We use an embedded test to make sure we fix the _captureVar bug
 
3302
        class Test(tests.TestCase):
 
3303
            def test_me(self):
 
3304
                # The first call save the 42 value
 
3305
                self.overrideEnv('MYVAR', None)
 
3306
                self.assertEquals(None, os.environ.get('MYVAR'))
 
3307
                # Make sure we can call it twice
 
3308
                self.overrideEnv('MYVAR', None)
 
3309
                self.assertEquals(None, os.environ.get('MYVAR'))
 
3310
        output = StringIO()
 
3311
        result = tests.TextTestResult(output, 0, 1)
 
3312
        Test('test_me').run(result)
 
3313
        if not result.wasStrictlySuccessful():
 
3314
            self.fail(output.getvalue())
 
3315
        # We get our value back
 
3316
        self.assertEquals('42', os.environ.get('MYVAR'))
 
3317
 
 
3318
 
 
3319
class TestIsolatedEnv(tests.TestCase):
 
3320
    """Test isolating tests from os.environ.
 
3321
 
 
3322
    Since we use tests that are already isolated from os.environ a bit of care
 
3323
    should be taken when designing the tests to avoid bootstrap side-effects.
 
3324
    The tests start an already clean os.environ which allow doing valid
 
3325
    assertions about which variables are present or not and design tests around
 
3326
    these assertions.
 
3327
    """
 
3328
 
 
3329
    class ScratchMonkey(tests.TestCase):
 
3330
 
 
3331
        def test_me(self):
 
3332
            pass
 
3333
 
 
3334
    def test_basics(self):
 
3335
        # Make sure we know the definition of BZR_HOME: not part of os.environ
 
3336
        # for tests.TestCase.
 
3337
        self.assertTrue('BZR_HOME' in tests.isolated_environ)
 
3338
        self.assertEquals(None, tests.isolated_environ['BZR_HOME'])
 
3339
        # Being part of isolated_environ, BZR_HOME should not appear here
 
3340
        self.assertFalse('BZR_HOME' in os.environ)
 
3341
        # Make sure we know the definition of LINES: part of os.environ for
 
3342
        # tests.TestCase
 
3343
        self.assertTrue('LINES' in tests.isolated_environ)
 
3344
        self.assertEquals('25', tests.isolated_environ['LINES'])
 
3345
        self.assertEquals('25', os.environ['LINES'])
 
3346
 
 
3347
    def test_injecting_unknown_variable(self):
 
3348
        # BZR_HOME is known to be absent from os.environ
 
3349
        test = self.ScratchMonkey('test_me')
 
3350
        tests.override_os_environ(test, {'BZR_HOME': 'foo'})
 
3351
        self.assertEquals('foo', os.environ['BZR_HOME'])
 
3352
        tests.restore_os_environ(test)
 
3353
        self.assertFalse('BZR_HOME' in os.environ)
 
3354
 
 
3355
    def test_injecting_known_variable(self):
 
3356
        test = self.ScratchMonkey('test_me')
 
3357
        # LINES is known to be present in os.environ
 
3358
        tests.override_os_environ(test, {'LINES': '42'})
 
3359
        self.assertEquals('42', os.environ['LINES'])
 
3360
        tests.restore_os_environ(test)
 
3361
        self.assertEquals('25', os.environ['LINES'])
 
3362
 
 
3363
    def test_deleting_variable(self):
 
3364
        test = self.ScratchMonkey('test_me')
 
3365
        # LINES is known to be present in os.environ
 
3366
        tests.override_os_environ(test, {'LINES': None})
 
3367
        self.assertTrue('LINES' not in os.environ)
 
3368
        tests.restore_os_environ(test)
 
3369
        self.assertEquals('25', os.environ['LINES'])
 
3370
 
 
3371
 
 
3372
class TestDocTestSuiteIsolation(tests.TestCase):
 
3373
    """Test that `tests.DocTestSuite` isolates doc tests from os.environ.
 
3374
 
 
3375
    Since tests.TestCase alreay provides an isolation from os.environ, we use
 
3376
    the clean environment as a base for testing. To precisely capture the
 
3377
    isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
 
3378
    compare against.
 
3379
 
 
3380
    We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
 
3381
    not `os.environ` so each test overrides it to suit its needs.
 
3382
 
 
3383
    """
 
3384
 
 
3385
    def get_doctest_suite_for_string(self, klass, string):
 
3386
        class Finder(doctest.DocTestFinder):
 
3387
 
 
3388
            def find(*args, **kwargs):
 
3389
                test = doctest.DocTestParser().get_doctest(
 
3390
                    string, {}, 'foo', 'foo.py', 0)
 
3391
                return [test]
 
3392
 
 
3393
        suite = klass(test_finder=Finder())
 
3394
        return suite
 
3395
 
 
3396
    def run_doctest_suite_for_string(self, klass, string):
 
3397
        suite = self.get_doctest_suite_for_string(klass, string)
 
3398
        output = StringIO()
 
3399
        result = tests.TextTestResult(output, 0, 1)
 
3400
        suite.run(result)
 
3401
        return result, output
 
3402
 
 
3403
    def assertDocTestStringSucceds(self, klass, string):
 
3404
        result, output = self.run_doctest_suite_for_string(klass, string)
 
3405
        if not result.wasStrictlySuccessful():
 
3406
            self.fail(output.getvalue())
 
3407
 
 
3408
    def assertDocTestStringFails(self, klass, string):
 
3409
        result, output = self.run_doctest_suite_for_string(klass, string)
 
3410
        if result.wasStrictlySuccessful():
 
3411
            self.fail(output.getvalue())
 
3412
 
 
3413
    def test_injected_variable(self):
 
3414
        self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
 
3415
        test = """
 
3416
            >>> import os
 
3417
            >>> os.environ['LINES']
 
3418
            '42'
 
3419
            """
 
3420
        # doctest.DocTestSuite fails as it sees '25'
 
3421
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
 
3422
        # tests.DocTestSuite sees '42'
 
3423
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
 
3424
 
 
3425
    def test_deleted_variable(self):
 
3426
        self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
 
3427
        test = """
 
3428
            >>> import os
 
3429
            >>> os.environ.get('LINES')
 
3430
            """
 
3431
        # doctest.DocTestSuite fails as it sees '25'
 
3432
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
 
3433
        # tests.DocTestSuite sees None
 
3434
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
 
3435
 
 
3436
 
 
3437
class TestSelftestExcludePatterns(tests.TestCase):
 
3438
 
 
3439
    def setUp(self):
 
3440
        super(TestSelftestExcludePatterns, self).setUp()
 
3441
        self.overrideAttr(tests, 'test_suite', self.suite_factory)
 
3442
 
 
3443
    def suite_factory(self, keep_only=None, starting_with=None):
 
3444
        """A test suite factory with only a few tests."""
 
3445
        class Test(tests.TestCase):
 
3446
            def id(self):
 
3447
                # We don't need the full class path
 
3448
                return self._testMethodName
 
3449
            def a(self):
 
3450
                pass
 
3451
            def b(self):
 
3452
                pass
 
3453
            def c(self):
 
3454
                pass
 
3455
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
 
3456
 
 
3457
    def assertTestList(self, expected, *selftest_args):
 
3458
        # We rely on setUp installing the right test suite factory so we can
 
3459
        # test at the command level without loading the whole test suite
 
3460
        out, err = self.run_bzr(('selftest', '--list') + selftest_args)
 
3461
        actual = out.splitlines()
 
3462
        self.assertEquals(expected, actual)
 
3463
 
 
3464
    def test_full_list(self):
 
3465
        self.assertTestList(['a', 'b', 'c'])
 
3466
 
 
3467
    def test_single_exclude(self):
 
3468
        self.assertTestList(['b', 'c'], '-x', 'a')
 
3469
 
 
3470
    def test_mutiple_excludes(self):
 
3471
        self.assertTestList(['c'], '-x', 'a', '-x', 'b')
 
3472
 
 
3473
 
 
3474
class TestCounterHooks(tests.TestCase, SelfTestHelper):
 
3475
 
 
3476
    _test_needs_features = [features.subunit]
 
3477
 
 
3478
    def setUp(self):
 
3479
        super(TestCounterHooks, self).setUp()
 
3480
        class Test(tests.TestCase):
 
3481
 
 
3482
            def setUp(self):
 
3483
                super(Test, self).setUp()
 
3484
                self.hooks = hooks.Hooks()
 
3485
                self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
 
3486
                self.install_counter_hook(self.hooks, 'myhook')
 
3487
 
 
3488
            def no_hook(self):
 
3489
                pass
 
3490
 
 
3491
            def run_hook_once(self):
 
3492
                for hook in self.hooks['myhook']:
 
3493
                    hook(self)
 
3494
 
 
3495
        self.test_class = Test
 
3496
 
 
3497
    def assertHookCalls(self, expected_calls, test_name):
 
3498
        test = self.test_class(test_name)
 
3499
        result = unittest.TestResult()
 
3500
        test.run(result)
 
3501
        self.assertTrue(hasattr(test, '_counters'))
 
3502
        self.assertTrue(test._counters.has_key('myhook'))
 
3503
        self.assertEquals(expected_calls, test._counters['myhook'])
 
3504
 
 
3505
    def test_no_hook(self):
 
3506
        self.assertHookCalls(0, 'no_hook')
 
3507
 
 
3508
    def test_run_hook_once(self):
 
3509
        tt = features.testtools
 
3510
        if tt.module.__version__ < (0, 9, 8):
 
3511
            raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
 
3512
        self.assertHookCalls(1, 'run_hook_once')