~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Jelmer Vernooij
  • Date: 2012-03-21 10:48:58 UTC
  • mto: This revision was merged to the branch mainline in revision 6515.
  • Revision ID: jelmer@vernstok.nl-20120321104858-nho0dkromtrwxfum
Remove some unused imports.

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 gc
 
21
import doctest
21
22
import os
22
23
import signal
23
24
import sys
 
25
import threading
24
26
import time
25
27
import unittest
26
28
import warnings
27
29
 
28
 
from testtools import MultiTestResult
 
30
from testtools import (
 
31
    ExtendedToOriginalDecorator,
 
32
    MultiTestResult,
 
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,
 
46
    controldir,
41
47
    errors,
 
48
    hooks,
42
49
    lockdir,
43
50
    memorytree,
44
51
    osutils,
45
 
    progress,
46
52
    remote,
47
53
    repository,
48
54
    symbol_versioning,
49
55
    tests,
50
56
    transport,
51
57
    workingtree,
 
58
    workingtree_3,
 
59
    workingtree_4,
52
60
    )
53
61
from bzrlib.repofmt import (
54
62
    groupcompress_repo,
55
 
    pack_repo,
56
 
    weaverepo,
57
63
    )
58
64
from bzrlib.symbol_versioning import (
59
65
    deprecated_function,
64
70
    features,
65
71
    test_lsprof,
66
72
    test_server,
67
 
    test_sftp_transport,
68
73
    TestUtil,
69
74
    )
70
 
from bzrlib.trace import note
 
75
from bzrlib.trace import note, mutter
71
76
from bzrlib.transport import memory
72
 
from bzrlib.version import _get_bzr_source_tree
73
77
 
74
78
 
75
79
def _test_ids(test_suite):
77
81
    return [t.id() for t in tests.iter_suite_tests(test_suite)]
78
82
 
79
83
 
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
84
class MetaTestLog(tests.TestCase):
93
85
 
94
86
    def test_logging(self):
100
92
            "text", "plain", {"charset": "utf8"})))
101
93
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
102
94
        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()
 
95
            DocTestMatches(u"...a test message\n", doctest.ELLIPSIS))
113
96
 
114
97
 
115
98
class TestTreeShape(tests.TestCaseInTempDir):
116
99
 
117
100
    def test_unicode_paths(self):
118
 
        self.requireFeature(tests.UnicodeFilename)
 
101
        self.requireFeature(features.UnicodeFilenameFeature)
119
102
 
120
103
        filename = u'hell\u00d8'
121
104
        self.build_tree_contents([(filename, 'contents of hello')])
122
 
        self.failUnlessExists(filename)
 
105
        self.assertPathExists(filename)
 
106
 
 
107
 
 
108
class TestClassesAvailable(tests.TestCase):
 
109
    """As a convenience we expose Test* classes from bzrlib.tests"""
 
110
 
 
111
    def test_test_case(self):
 
112
        from bzrlib.tests import TestCase
 
113
 
 
114
    def test_test_loader(self):
 
115
        from bzrlib.tests import TestLoader
 
116
 
 
117
    def test_test_suite(self):
 
118
        from bzrlib.tests import TestSuite
123
119
 
124
120
 
125
121
class TestTransportScenarios(tests.TestCase):
208
204
    def test_scenarios(self):
209
205
        # check that constructor parameters are passed through to the adapted
210
206
        # test.
211
 
        from bzrlib.tests.per_bzrdir import make_scenarios
 
207
        from bzrlib.tests.per_controldir import make_scenarios
212
208
        vfs_factory = "v"
213
209
        server1 = "a"
214
210
        server2 = "b"
312
308
        from bzrlib.tests.per_interrepository import make_scenarios
313
309
        server1 = "a"
314
310
        server2 = "b"
315
 
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
 
311
        formats = [("C0", "C1", "C2", "C3"), ("D0", "D1", "D2", "D3")]
316
312
        scenarios = make_scenarios(server1, server2, formats)
317
313
        self.assertEqual([
318
314
            ('C0,str,str',
319
315
             {'repository_format': 'C1',
320
316
              'repository_format_to': 'C2',
321
317
              'transport_readonly_server': 'b',
322
 
              'transport_server': 'a'}),
 
318
              'transport_server': 'a',
 
319
              'extra_setup': 'C3'}),
323
320
            ('D0,str,str',
324
321
             {'repository_format': 'D1',
325
322
              'repository_format_to': 'D2',
326
323
              'transport_readonly_server': 'b',
327
 
              'transport_server': 'a'})],
 
324
              'transport_server': 'a',
 
325
              'extra_setup': 'D3'})],
328
326
            scenarios)
329
327
 
330
328
 
336
334
        from bzrlib.tests.per_workingtree import make_scenarios
337
335
        server1 = "a"
338
336
        server2 = "b"
339
 
        formats = [workingtree.WorkingTreeFormat2(),
340
 
                   workingtree.WorkingTreeFormat3(),]
 
337
        formats = [workingtree_4.WorkingTreeFormat4(),
 
338
                   workingtree_3.WorkingTreeFormat3(),]
341
339
        scenarios = make_scenarios(server1, server2, formats)
342
340
        self.assertEqual([
343
 
            ('WorkingTreeFormat2',
 
341
            ('WorkingTreeFormat4',
344
342
             {'bzrdir_format': formats[0]._matchingbzrdir,
345
343
              'transport_readonly_server': 'b',
346
344
              'transport_server': 'a',
373
371
            )
374
372
        server1 = "a"
375
373
        server2 = "b"
376
 
        formats = [workingtree.WorkingTreeFormat2(),
377
 
                   workingtree.WorkingTreeFormat3(),]
 
374
        formats = [workingtree_4.WorkingTreeFormat4(),
 
375
                   workingtree_3.WorkingTreeFormat3(),]
378
376
        scenarios = make_scenarios(server1, server2, formats)
379
377
        self.assertEqual(7, len(scenarios))
380
 
        default_wt_format = workingtree.WorkingTreeFormat4._default_format
381
 
        wt4_format = workingtree.WorkingTreeFormat4()
382
 
        wt5_format = workingtree.WorkingTreeFormat5()
 
378
        default_wt_format = workingtree.format_registry.get_default()
 
379
        wt4_format = workingtree_4.WorkingTreeFormat4()
 
380
        wt5_format = workingtree_4.WorkingTreeFormat5()
383
381
        expected_scenarios = [
384
 
            ('WorkingTreeFormat2',
 
382
            ('WorkingTreeFormat4',
385
383
             {'bzrdir_format': formats[0]._matchingbzrdir,
386
384
              'transport_readonly_server': 'b',
387
385
              'transport_server': 'a',
447
445
        # ones to add.
448
446
        from bzrlib.tests.per_tree import (
449
447
            return_parameter,
450
 
            revision_tree_from_workingtree
451
448
            )
452
449
        from bzrlib.tests.per_intertree import (
453
450
            make_scenarios,
454
451
            )
455
 
        from bzrlib.workingtree import WorkingTreeFormat2, WorkingTreeFormat3
 
452
        from bzrlib.workingtree_3 import WorkingTreeFormat3
 
453
        from bzrlib.workingtree_4 import WorkingTreeFormat4
456
454
        input_test = TestInterTreeScenarios(
457
455
            "test_scenarios")
458
456
        server1 = "a"
459
457
        server2 = "b"
460
 
        format1 = WorkingTreeFormat2()
 
458
        format1 = WorkingTreeFormat4()
461
459
        format2 = WorkingTreeFormat3()
462
460
        formats = [("1", str, format1, format2, "converter1"),
463
461
            ("2", int, format2, format1, "converter2")]
509
507
        self.assertRaises(AssertionError, self.assertEqualStat,
510
508
            os.lstat("foo"), os.lstat("longname"))
511
509
 
 
510
    def test_failUnlessExists(self):
 
511
        """Deprecated failUnlessExists and failIfExists"""
 
512
        self.applyDeprecated(
 
513
            deprecated_in((2, 4)),
 
514
            self.failUnlessExists, '.')
 
515
        self.build_tree(['foo/', 'foo/bar'])
 
516
        self.applyDeprecated(
 
517
            deprecated_in((2, 4)),
 
518
            self.failUnlessExists, 'foo/bar')
 
519
        self.applyDeprecated(
 
520
            deprecated_in((2, 4)),
 
521
            self.failIfExists, 'foo/foo')
 
522
 
 
523
    def test_assertPathExists(self):
 
524
        self.assertPathExists('.')
 
525
        self.build_tree(['foo/', 'foo/bar'])
 
526
        self.assertPathExists('foo/bar')
 
527
        self.assertPathDoesNotExist('foo/foo')
 
528
 
512
529
 
513
530
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
514
531
 
548
565
        tree = self.make_branch_and_memory_tree('dir')
549
566
        # Guard against regression into MemoryTransport leaking
550
567
        # files to disk instead of keeping them in memory.
551
 
        self.failIf(osutils.lexists('dir'))
 
568
        self.assertFalse(osutils.lexists('dir'))
552
569
        self.assertIsInstance(tree, memorytree.MemoryTree)
553
570
 
554
571
    def test_make_branch_and_memory_tree_with_format(self):
555
572
        """make_branch_and_memory_tree should accept a format option."""
556
573
        format = bzrdir.BzrDirMetaFormat1()
557
 
        format.repository_format = weaverepo.RepositoryFormat7()
 
574
        format.repository_format = repository.format_registry.get_default()
558
575
        tree = self.make_branch_and_memory_tree('dir', format=format)
559
576
        # Guard against regression into MemoryTransport leaking
560
577
        # files to disk instead of keeping them in memory.
561
 
        self.failIf(osutils.lexists('dir'))
 
578
        self.assertFalse(osutils.lexists('dir'))
562
579
        self.assertIsInstance(tree, memorytree.MemoryTree)
563
580
        self.assertEqual(format.repository_format.__class__,
564
581
            tree.branch.repository._format.__class__)
568
585
        self.assertIsInstance(builder, branchbuilder.BranchBuilder)
569
586
        # Guard against regression into MemoryTransport leaking
570
587
        # files to disk instead of keeping them in memory.
571
 
        self.failIf(osutils.lexists('dir'))
 
588
        self.assertFalse(osutils.lexists('dir'))
572
589
 
573
590
    def test_make_branch_builder_with_format(self):
574
591
        # Use a repo layout that doesn't conform to a 'named' layout, to ensure
575
592
        # that the format objects are used.
576
593
        format = bzrdir.BzrDirMetaFormat1()
577
 
        repo_format = weaverepo.RepositoryFormat7()
 
594
        repo_format = repository.format_registry.get_default()
578
595
        format.repository_format = repo_format
579
596
        builder = self.make_branch_builder('dir', format=format)
580
597
        the_branch = builder.get_branch()
581
598
        # Guard against regression into MemoryTransport leaking
582
599
        # files to disk instead of keeping them in memory.
583
 
        self.failIf(osutils.lexists('dir'))
 
600
        self.assertFalse(osutils.lexists('dir'))
584
601
        self.assertEqual(format.repository_format.__class__,
585
602
                         the_branch.repository._format.__class__)
586
603
        self.assertEqual(repo_format.get_format_string(),
592
609
        the_branch = builder.get_branch()
593
610
        # Guard against regression into MemoryTransport leaking
594
611
        # files to disk instead of keeping them in memory.
595
 
        self.failIf(osutils.lexists('dir'))
596
 
        dir_format = bzrdir.format_registry.make_bzrdir('knit')
 
612
        self.assertFalse(osutils.lexists('dir'))
 
613
        dir_format = controldir.format_registry.make_bzrdir('knit')
597
614
        self.assertEqual(dir_format.repository_format.__class__,
598
615
                         the_branch.repository._format.__class__)
599
616
        self.assertEqual('Bazaar-NG Knit Repository Format 1',
603
620
    def test_dangling_locks_cause_failures(self):
604
621
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
605
622
            def test_function(self):
606
 
                t = self.get_transport('.')
 
623
                t = self.get_transport_from_path('.')
607
624
                l = lockdir.LockDir(t, 'lock')
608
625
                l.create()
609
626
                l.attempt_lock()
610
627
        test = TestDanglingLock('test_function')
611
628
        result = test.run()
 
629
        total_failures = result.errors + result.failures
612
630
        if self._lock_check_thorough:
613
 
            self.assertEqual(1, len(result.errors))
 
631
            self.assertEqual(1, len(total_failures))
614
632
        else:
615
633
            # When _lock_check_thorough is disabled, then we don't trigger a
616
634
            # failure
617
 
            self.assertEqual(0, len(result.errors))
 
635
            self.assertEqual(0, len(total_failures))
618
636
 
619
637
 
620
638
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
621
639
    """Tests for the convenience functions TestCaseWithTransport introduces."""
622
640
 
623
641
    def test_get_readonly_url_none(self):
624
 
        from bzrlib.transport import get_transport
625
642
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
626
643
        self.vfs_transport_factory = memory.MemoryServer
627
644
        self.transport_readonly_server = None
629
646
        # for the server
630
647
        url = self.get_readonly_url()
631
648
        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))
 
649
        t = transport.get_transport_from_url(url)
 
650
        t2 = transport.get_transport_from_url(url2)
 
651
        self.assertIsInstance(t, ReadonlyTransportDecorator)
 
652
        self.assertIsInstance(t2, ReadonlyTransportDecorator)
636
653
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
637
654
 
638
655
    def test_get_readonly_url_http(self):
639
656
        from bzrlib.tests.http_server import HttpServer
640
 
        from bzrlib.transport import get_transport
641
657
        from bzrlib.transport.http import HttpTransportBase
642
658
        self.transport_server = test_server.LocalURLServer
643
659
        self.transport_readonly_server = HttpServer
645
661
        url = self.get_readonly_url()
646
662
        url2 = self.get_readonly_url('foo/bar')
647
663
        # 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))
 
664
        t = transport.get_transport_from_url(url)
 
665
        t2 = transport.get_transport_from_url(url2)
 
666
        self.assertIsInstance(t, HttpTransportBase)
 
667
        self.assertIsInstance(t2, HttpTransportBase)
652
668
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
653
669
 
654
670
    def test_is_directory(self):
662
678
    def test_make_branch_builder(self):
663
679
        builder = self.make_branch_builder('dir')
664
680
        rev_id = builder.build_commit()
665
 
        self.failUnlessExists('dir')
666
 
        a_dir = bzrdir.BzrDir.open('dir')
 
681
        self.assertPathExists('dir')
 
682
        a_dir = controldir.ControlDir.open('dir')
667
683
        self.assertRaises(errors.NoWorkingTree, a_dir.open_workingtree)
668
684
        a_branch = a_dir.open_branch()
669
685
        builder_branch = builder.get_branch()
684
700
        self.assertIsInstance(result_bzrdir.transport,
685
701
                              memory.MemoryTransport)
686
702
        # should not be on disk, should only be in memory
687
 
        self.failIfExists('subdir')
 
703
        self.assertPathDoesNotExist('subdir')
688
704
 
689
705
 
690
706
class TestChrootedTest(tests.ChrootedTestCase):
691
707
 
692
708
    def test_root_is_root(self):
693
 
        from bzrlib.transport import get_transport
694
 
        t = get_transport(self.get_readonly_url())
 
709
        t = transport.get_transport_from_url(self.get_readonly_url())
695
710
        url = t.base
696
711
        self.assertEqual(url, t.clone('..').base)
697
712
 
699
714
class TestProfileResult(tests.TestCase):
700
715
 
701
716
    def test_profiles_tests(self):
702
 
        self.requireFeature(test_lsprof.LSProfFeature)
703
 
        terminal = testtools.tests.helpers.ExtendedTestResult()
 
717
        self.requireFeature(features.lsprof_feature)
 
718
        terminal = testtools.testresult.doubles.ExtendedTestResult()
704
719
        result = tests.ProfileResult(terminal)
705
720
        class Sample(tests.TestCase):
706
721
            def a(self):
723
738
                descriptions=0,
724
739
                verbosity=1,
725
740
                )
726
 
        capture = testtools.tests.helpers.ExtendedTestResult()
 
741
        capture = testtools.testresult.doubles.ExtendedTestResult()
727
742
        test_case.run(MultiTestResult(result, capture))
728
743
        run_case = capture._events[0][1]
729
744
        timed_string = result._testTimeString(run_case)
750
765
        self.check_timing(ShortDelayTestCase('test_short_delay'),
751
766
                          r"^ +[0-9]+ms$")
752
767
 
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
768
    def _time_hello_world_encoding(self):
794
769
        """Profile two sleep calls
795
770
 
800
775
 
801
776
    def test_lsprofiling(self):
802
777
        """Verbose test result prints lsprof statistics from test cases."""
803
 
        self.requireFeature(test_lsprof.LSProfFeature)
 
778
        self.requireFeature(features.lsprof_feature)
804
779
        result_stream = StringIO()
805
780
        result = bzrlib.tests.VerboseTestResult(
806
 
            unittest._WritelnDecorator(result_stream),
 
781
            result_stream,
807
782
            descriptions=0,
808
783
            verbosity=2,
809
784
            )
835
810
        self.assertContainsRe(output,
836
811
            r"LSProf output for <type 'unicode'>\(\('world',\), {'errors': 'replace'}\)\n")
837
812
 
 
813
    def test_uses_time_from_testtools(self):
 
814
        """Test case timings in verbose results should use testtools times"""
 
815
        import datetime
 
816
        class TimeAddedVerboseTestResult(tests.VerboseTestResult):
 
817
            def startTest(self, test):
 
818
                self.time(datetime.datetime.utcfromtimestamp(1.145))
 
819
                super(TimeAddedVerboseTestResult, self).startTest(test)
 
820
            def addSuccess(self, test):
 
821
                self.time(datetime.datetime.utcfromtimestamp(51.147))
 
822
                super(TimeAddedVerboseTestResult, self).addSuccess(test)
 
823
            def report_tests_starting(self): pass
 
824
        sio = StringIO()
 
825
        self.get_passing_test().run(TimeAddedVerboseTestResult(sio, 0, 2))
 
826
        self.assertEndsWith(sio.getvalue(), "OK    50002ms\n")
 
827
 
838
828
    def test_known_failure(self):
839
 
        """A KnownFailure being raised should trigger several result actions."""
 
829
        """Using knownFailure should trigger several result actions."""
840
830
        class InstrumentedTestResult(tests.ExtendedTestResult):
841
831
            def stopTestRun(self): pass
842
 
            def startTests(self): pass
843
 
            def report_test_start(self, test): pass
 
832
            def report_tests_starting(self): pass
844
833
            def report_known_failure(self, test, err=None, details=None):
845
834
                self._call = test, 'known failure'
846
835
        result = InstrumentedTestResult(None, None, None, None)
847
836
        class Test(tests.TestCase):
848
837
            def test_function(self):
849
 
                raise tests.KnownFailure('failed!')
 
838
                self.knownFailure('failed!')
850
839
        test = Test("test_function")
851
840
        test.run(result)
852
841
        # it should invoke 'report_known_failure'.
864
853
        # verbose test output formatting
865
854
        result_stream = StringIO()
866
855
        result = bzrlib.tests.VerboseTestResult(
867
 
            unittest._WritelnDecorator(result_stream),
 
856
            result_stream,
868
857
            descriptions=0,
869
858
            verbosity=2,
870
859
            )
871
 
        test = self.get_passing_test()
872
 
        result.startTest(test)
873
 
        prefix = len(result_stream.getvalue())
874
 
        # the err parameter has the shape:
875
 
        # (class, exception object, traceback)
876
 
        # KnownFailures dont get their tracebacks shown though, so we
877
 
        # can skip that.
878
 
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
879
 
        result.report_known_failure(test, err)
880
 
        output = result_stream.getvalue()[prefix:]
881
 
        lines = output.splitlines()
882
 
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
883
 
        self.assertEqual(lines[1], '    foo')
884
 
        self.assertEqual(2, len(lines))
 
860
        _get_test("test_xfail").run(result)
 
861
        self.assertContainsRe(result_stream.getvalue(),
 
862
            "\n\\S+\\.test_xfail\\s+XFAIL\\s+\\d+ms\n"
 
863
            "\\s*(?:Text attachment: )?reason"
 
864
            "(?:\n-+\n|: {{{)"
 
865
            "this_fails"
 
866
            "(?:\n-+\n|}}}\n)")
885
867
 
886
868
    def get_passing_test(self):
887
869
        """Return a test object that can't be run usefully."""
893
875
        """Test the behaviour of invoking addNotSupported."""
894
876
        class InstrumentedTestResult(tests.ExtendedTestResult):
895
877
            def stopTestRun(self): pass
896
 
            def startTests(self): pass
897
 
            def report_test_start(self, test): pass
 
878
            def report_tests_starting(self): pass
898
879
            def report_unsupported(self, test, feature):
899
880
                self._call = test, feature
900
881
        result = InstrumentedTestResult(None, None, None, None)
901
882
        test = SampleTestCase('_test_pass')
902
 
        feature = tests.Feature()
 
883
        feature = features.Feature()
903
884
        result.startTest(test)
904
885
        result.addNotSupported(test, feature)
905
886
        # it should invoke 'report_unsupported'.
919
900
        # verbose test output formatting
920
901
        result_stream = StringIO()
921
902
        result = bzrlib.tests.VerboseTestResult(
922
 
            unittest._WritelnDecorator(result_stream),
 
903
            result_stream,
923
904
            descriptions=0,
924
905
            verbosity=2,
925
906
            )
926
907
        test = self.get_passing_test()
927
 
        feature = tests.Feature()
 
908
        feature = features.Feature()
928
909
        result.startTest(test)
929
910
        prefix = len(result_stream.getvalue())
930
911
        result.report_unsupported(test, feature)
939
920
        """An UnavailableFeature being raised should invoke addNotSupported."""
940
921
        class InstrumentedTestResult(tests.ExtendedTestResult):
941
922
            def stopTestRun(self): pass
942
 
            def startTests(self): pass
943
 
            def report_test_start(self, test): pass
 
923
            def report_tests_starting(self): pass
944
924
            def addNotSupported(self, test, feature):
945
925
                self._call = test, feature
946
926
        result = InstrumentedTestResult(None, None, None, None)
947
 
        feature = tests.Feature()
 
927
        feature = features.Feature()
948
928
        class Test(tests.TestCase):
949
929
            def test_function(self):
950
930
                raise tests.UnavailableFeature(feature)
969
949
    def test_strict_with_known_failure(self):
970
950
        result = bzrlib.tests.TextTestResult(self._log_file, descriptions=0,
971
951
                                             verbosity=1)
972
 
        test = self.get_passing_test()
973
 
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
974
 
        result.addExpectedFailure(test, err)
 
952
        test = _get_test("test_xfail")
 
953
        test.run(result)
975
954
        self.assertFalse(result.wasStrictlySuccessful())
976
955
        self.assertEqual(None, result._extractBenchmarkTime(test))
977
956
 
988
967
        class InstrumentedTestResult(tests.ExtendedTestResult):
989
968
            calls = 0
990
969
            def startTests(self): self.calls += 1
991
 
            def report_test_start(self, test): pass
992
970
        result = InstrumentedTestResult(None, None, None, None)
993
971
        def test_function():
994
972
            pass
996
974
        test.run(result)
997
975
        self.assertEquals(1, result.calls)
998
976
 
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()
 
977
    def test_startTests_only_once(self):
 
978
        """With multiple tests startTests should still only be called once"""
 
979
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
980
            calls = 0
 
981
            def startTests(self): self.calls += 1
 
982
        result = InstrumentedTestResult(None, None, None, None)
 
983
        suite = unittest.TestSuite([
 
984
            unittest.FunctionTestCase(lambda: None),
 
985
            unittest.FunctionTestCase(lambda: None)])
 
986
        suite.run(result)
 
987
        self.assertEquals(1, result.calls)
 
988
        self.assertEquals(2, result.count)
1007
989
 
1008
990
 
1009
991
class TestRunner(tests.TestCase):
1022
1004
        because of our use of global state.
1023
1005
        """
1024
1006
        old_root = tests.TestCaseInTempDir.TEST_ROOT
1025
 
        old_leak = tests.TestCase._first_thread_leaker_id
1026
1007
        try:
1027
1008
            tests.TestCaseInTempDir.TEST_ROOT = None
1028
 
            tests.TestCase._first_thread_leaker_id = None
1029
1009
            return testrunner.run(test)
1030
1010
        finally:
1031
1011
            tests.TestCaseInTempDir.TEST_ROOT = old_root
1032
 
            tests.TestCase._first_thread_leaker_id = old_leak
1033
1012
 
1034
1013
    def test_known_failure_failed_run(self):
1035
1014
        # run a test that generates a known failure which should be printed in
1040
1019
        test = unittest.TestSuite()
1041
1020
        test.addTest(Test("known_failure_test"))
1042
1021
        def failing_test():
1043
 
            self.fail('foo')
 
1022
            raise AssertionError('foo')
1044
1023
        test.addTest(unittest.FunctionTestCase(failing_test))
1045
1024
        stream = StringIO()
1046
1025
        runner = tests.TextTestRunner(stream=stream)
1054
1033
            '^----------------------------------------------------------------------\n'
1055
1034
            'Traceback \\(most recent call last\\):\n'
1056
1035
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1057
 
            '    self.fail\\(\'foo\'\\)\n'
 
1036
            '    raise AssertionError\\(\'foo\'\\)\n'
1058
1037
            '.*'
1059
1038
            '^----------------------------------------------------------------------\n'
1060
1039
            '.*'
1066
1045
        # the final output.
1067
1046
        class Test(tests.TestCase):
1068
1047
            def known_failure_test(self):
1069
 
                self.expectFailure('failed', self.assertTrue, False)
 
1048
                self.knownFailure("Never works...")
1070
1049
        test = Test("known_failure_test")
1071
1050
        stream = StringIO()
1072
1051
        runner = tests.TextTestRunner(stream=stream)
1078
1057
            '\n'
1079
1058
            'OK \\(known_failures=1\\)\n')
1080
1059
 
 
1060
    def test_unexpected_success_bad(self):
 
1061
        class Test(tests.TestCase):
 
1062
            def test_truth(self):
 
1063
                self.expectFailure("No absolute truth", self.assertTrue, True)
 
1064
        runner = tests.TextTestRunner(stream=StringIO())
 
1065
        result = self.run_test_runner(runner, Test("test_truth"))
 
1066
        self.assertContainsRe(runner.stream.getvalue(),
 
1067
            "=+\n"
 
1068
            "FAIL: \\S+\.test_truth\n"
 
1069
            "-+\n"
 
1070
            "(?:.*\n)*"
 
1071
            "\\s*(?:Text attachment: )?reason"
 
1072
            "(?:\n-+\n|: {{{)"
 
1073
            "No absolute truth"
 
1074
            "(?:\n-+\n|}}}\n)"
 
1075
            "(?:.*\n)*"
 
1076
            "-+\n"
 
1077
            "Ran 1 test in .*\n"
 
1078
            "\n"
 
1079
            "FAILED \\(failures=1\\)\n\\Z")
 
1080
 
1081
1081
    def test_result_decorator(self):
1082
1082
        # decorate results
1083
1083
        calls = []
1084
 
        class LoggingDecorator(tests.ForwardingResult):
 
1084
        class LoggingDecorator(ExtendedToOriginalDecorator):
1085
1085
            def startTest(self, test):
1086
 
                tests.ForwardingResult.startTest(self, test)
 
1086
                ExtendedToOriginalDecorator.startTest(self, test)
1087
1087
                calls.append('start')
1088
1088
        test = unittest.FunctionTestCase(lambda:None)
1089
1089
        stream = StringIO()
1167
1167
 
1168
1168
    def test_unsupported_features_listed(self):
1169
1169
        """When unsupported features are encountered they are detailed."""
1170
 
        class Feature1(tests.Feature):
 
1170
        class Feature1(features.Feature):
1171
1171
            def _probe(self): return False
1172
 
        class Feature2(tests.Feature):
 
1172
        class Feature2(features.Feature):
1173
1173
            def _probe(self): return False
1174
1174
        # create sample tests
1175
1175
        test1 = SampleTestCase('_test_pass')
1190
1190
            ],
1191
1191
            lines[-3:])
1192
1192
 
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)
 
1193
    def test_verbose_test_count(self):
 
1194
        """A verbose test run reports the right test count at the start"""
 
1195
        suite = TestUtil.TestSuite([
 
1196
            unittest.FunctionTestCase(lambda:None),
 
1197
            unittest.FunctionTestCase(lambda:None)])
 
1198
        self.assertEqual(suite.countTestCases(), 2)
 
1199
        stream = StringIO()
 
1200
        runner = tests.TextTestRunner(stream=stream, verbosity=2)
 
1201
        # Need to use the CountingDecorator as that's what sets num_tests
 
1202
        result = self.run_test_runner(runner, tests.CountingDecorator(suite))
 
1203
        self.assertStartsWith(stream.getvalue(), "running 2 tests")
1215
1204
 
1216
1205
    def test_startTestRun(self):
1217
1206
        """run should call result.startTestRun()"""
1218
1207
        calls = []
1219
 
        class LoggingDecorator(tests.ForwardingResult):
 
1208
        class LoggingDecorator(ExtendedToOriginalDecorator):
1220
1209
            def startTestRun(self):
1221
 
                tests.ForwardingResult.startTestRun(self)
 
1210
                ExtendedToOriginalDecorator.startTestRun(self)
1222
1211
                calls.append('startTestRun')
1223
1212
        test = unittest.FunctionTestCase(lambda:None)
1224
1213
        stream = StringIO()
1230
1219
    def test_stopTestRun(self):
1231
1220
        """run should call result.stopTestRun()"""
1232
1221
        calls = []
1233
 
        class LoggingDecorator(tests.ForwardingResult):
 
1222
        class LoggingDecorator(ExtendedToOriginalDecorator):
1234
1223
            def stopTestRun(self):
1235
 
                tests.ForwardingResult.stopTestRun(self)
 
1224
                ExtendedToOriginalDecorator.stopTestRun(self)
1236
1225
                calls.append('stopTestRun')
1237
1226
        test = unittest.FunctionTestCase(lambda:None)
1238
1227
        stream = StringIO()
1241
1230
        result = self.run_test_runner(runner, test)
1242
1231
        self.assertLength(1, calls)
1243
1232
 
 
1233
    def test_unicode_test_output_on_ascii_stream(self):
 
1234
        """Showing results should always succeed even on an ascii console"""
 
1235
        class FailureWithUnicode(tests.TestCase):
 
1236
            def test_log_unicode(self):
 
1237
                self.log(u"\u2606")
 
1238
                self.fail("Now print that log!")
 
1239
        out = StringIO()
 
1240
        self.overrideAttr(osutils, "get_terminal_encoding",
 
1241
            lambda trace=False: "ascii")
 
1242
        result = self.run_test_runner(tests.TextTestRunner(stream=out),
 
1243
            FailureWithUnicode("test_log_unicode"))
 
1244
        self.assertContainsRe(out.getvalue(),
 
1245
            "(?:Text attachment: )?log"
 
1246
            "(?:\n-+\n|: {{{)"
 
1247
            "\d+\.\d+  \\\\u2606"
 
1248
            "(?:\n-+\n|}}}\n)")
 
1249
 
1244
1250
 
1245
1251
class SampleTestCase(tests.TestCase):
1246
1252
 
1421
1427
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1422
1428
        output_stream = StringIO()
1423
1429
        result = bzrlib.tests.VerboseTestResult(
1424
 
            unittest._WritelnDecorator(output_stream),
 
1430
            output_stream,
1425
1431
            descriptions=0,
1426
1432
            verbosity=2)
1427
1433
        sample_test.run(result)
1434
1440
        # Note this test won't fail with hooks that the core library doesn't
1435
1441
        # use - but it trigger with a plugin that adds hooks, so its still a
1436
1442
        # useful warning in that case.
1437
 
        self.assertEqual(bzrlib.branch.BranchHooks(),
1438
 
            bzrlib.branch.Branch.hooks)
1439
 
        self.assertEqual(bzrlib.smart.server.SmartServerHooks(),
 
1443
        self.assertEqual(bzrlib.branch.BranchHooks(), bzrlib.branch.Branch.hooks)
 
1444
        self.assertEqual(
 
1445
            bzrlib.smart.server.SmartServerHooks(),
1440
1446
            bzrlib.smart.server.SmartTCPServer.hooks)
1441
 
        self.assertEqual(bzrlib.commands.CommandHooks(),
1442
 
            bzrlib.commands.Command.hooks)
 
1447
        self.assertEqual(
 
1448
            bzrlib.commands.CommandHooks(), bzrlib.commands.Command.hooks)
1443
1449
 
1444
1450
    def test__gather_lsprof_in_benchmarks(self):
1445
1451
        """When _gather_lsprof_in_benchmarks is on, accumulate profile data.
1446
1452
 
1447
1453
        Each self.time() call is individually and separately profiled.
1448
1454
        """
1449
 
        self.requireFeature(test_lsprof.LSProfFeature)
 
1455
        self.requireFeature(features.lsprof_feature)
1450
1456
        # overrides the class member with an instance member so no cleanup
1451
1457
        # needed.
1452
1458
        self._gather_lsprof_in_benchmarks = True
1471
1477
        transport_server = memory.MemoryServer()
1472
1478
        transport_server.start_server()
1473
1479
        self.addCleanup(transport_server.stop_server)
1474
 
        t = transport.get_transport(transport_server.get_url())
1475
 
        bzrdir.BzrDir.create(t.base)
 
1480
        t = transport.get_transport_from_url(transport_server.get_url())
 
1481
        controldir.ControlDir.create(t.base)
1476
1482
        self.assertRaises(errors.BzrError,
1477
 
            bzrdir.BzrDir.open_from_transport, t)
 
1483
            controldir.ControlDir.open_from_transport, t)
1478
1484
        # But if we declare this as safe, we can open the bzrdir.
1479
1485
        self.permit_url(t.base)
1480
1486
        self._bzr_selftest_roots.append(t.base)
1481
 
        bzrdir.BzrDir.open_from_transport(t)
 
1487
        controldir.ControlDir.open_from_transport(t)
1482
1488
 
1483
1489
    def test_requireFeature_available(self):
1484
1490
        """self.requireFeature(available) is a no-op."""
1485
 
        class Available(tests.Feature):
 
1491
        class Available(features.Feature):
1486
1492
            def _probe(self):return True
1487
1493
        feature = Available()
1488
1494
        self.requireFeature(feature)
1489
1495
 
1490
1496
    def test_requireFeature_unavailable(self):
1491
1497
        """self.requireFeature(unavailable) raises UnavailableFeature."""
1492
 
        class Unavailable(tests.Feature):
 
1498
        class Unavailable(features.Feature):
1493
1499
            def _probe(self):return False
1494
1500
        feature = Unavailable()
1495
1501
        self.assertRaises(tests.UnavailableFeature,
1654
1660
        test.run(unittest.TestResult())
1655
1661
        self.assertEqual('original', obj.test_attr)
1656
1662
 
 
1663
    def test_recordCalls(self):
 
1664
        from bzrlib.tests import test_selftest
 
1665
        calls = self.recordCalls(
 
1666
            test_selftest, '_add_numbers')
 
1667
        self.assertEqual(test_selftest._add_numbers(2, 10),
 
1668
            12)
 
1669
        self.assertEquals(calls, [((2, 10), {})])
 
1670
 
 
1671
 
 
1672
def _add_numbers(a, b):
 
1673
    return a + b
 
1674
 
 
1675
 
 
1676
class _MissingFeature(features.Feature):
 
1677
    def _probe(self):
 
1678
        return False
 
1679
missing_feature = _MissingFeature()
 
1680
 
 
1681
 
 
1682
def _get_test(name):
 
1683
    """Get an instance of a specific example test.
 
1684
 
 
1685
    We protect this in a function so that they don't auto-run in the test
 
1686
    suite.
 
1687
    """
 
1688
 
 
1689
    class ExampleTests(tests.TestCase):
 
1690
 
 
1691
        def test_fail(self):
 
1692
            mutter('this was a failing test')
 
1693
            self.fail('this test will fail')
 
1694
 
 
1695
        def test_error(self):
 
1696
            mutter('this test errored')
 
1697
            raise RuntimeError('gotcha')
 
1698
 
 
1699
        def test_missing_feature(self):
 
1700
            mutter('missing the feature')
 
1701
            self.requireFeature(missing_feature)
 
1702
 
 
1703
        def test_skip(self):
 
1704
            mutter('this test will be skipped')
 
1705
            raise tests.TestSkipped('reason')
 
1706
 
 
1707
        def test_success(self):
 
1708
            mutter('this test succeeds')
 
1709
 
 
1710
        def test_xfail(self):
 
1711
            mutter('test with expected failure')
 
1712
            self.knownFailure('this_fails')
 
1713
 
 
1714
        def test_unexpected_success(self):
 
1715
            mutter('test with unexpected success')
 
1716
            self.expectFailure('should_fail', lambda: None)
 
1717
 
 
1718
    return ExampleTests(name)
 
1719
 
 
1720
 
 
1721
class TestTestCaseLogDetails(tests.TestCase):
 
1722
 
 
1723
    def _run_test(self, test_name):
 
1724
        test = _get_test(test_name)
 
1725
        result = testtools.TestResult()
 
1726
        test.run(result)
 
1727
        return result
 
1728
 
 
1729
    def test_fail_has_log(self):
 
1730
        result = self._run_test('test_fail')
 
1731
        self.assertEqual(1, len(result.failures))
 
1732
        result_content = result.failures[0][1]
 
1733
        self.assertContainsRe(result_content,
 
1734
            '(?m)^(?:Text attachment: )?log(?:$|: )')
 
1735
        self.assertContainsRe(result_content, 'this was a failing test')
 
1736
 
 
1737
    def test_error_has_log(self):
 
1738
        result = self._run_test('test_error')
 
1739
        self.assertEqual(1, len(result.errors))
 
1740
        result_content = result.errors[0][1]
 
1741
        self.assertContainsRe(result_content,
 
1742
            '(?m)^(?:Text attachment: )?log(?:$|: )')
 
1743
        self.assertContainsRe(result_content, 'this test errored')
 
1744
 
 
1745
    def test_skip_has_no_log(self):
 
1746
        result = self._run_test('test_skip')
 
1747
        self.assertEqual(['reason'], result.skip_reasons.keys())
 
1748
        skips = result.skip_reasons['reason']
 
1749
        self.assertEqual(1, len(skips))
 
1750
        test = skips[0]
 
1751
        self.assertFalse('log' in test.getDetails())
 
1752
 
 
1753
    def test_missing_feature_has_no_log(self):
 
1754
        # testtools doesn't know about addNotSupported, so it just gets
 
1755
        # considered as a skip
 
1756
        result = self._run_test('test_missing_feature')
 
1757
        self.assertEqual([missing_feature], result.skip_reasons.keys())
 
1758
        skips = result.skip_reasons[missing_feature]
 
1759
        self.assertEqual(1, len(skips))
 
1760
        test = skips[0]
 
1761
        self.assertFalse('log' in test.getDetails())
 
1762
 
 
1763
    def test_xfail_has_no_log(self):
 
1764
        result = self._run_test('test_xfail')
 
1765
        self.assertEqual(1, len(result.expectedFailures))
 
1766
        result_content = result.expectedFailures[0][1]
 
1767
        self.assertNotContainsRe(result_content,
 
1768
            '(?m)^(?:Text attachment: )?log(?:$|: )')
 
1769
        self.assertNotContainsRe(result_content, 'test with expected failure')
 
1770
 
 
1771
    def test_unexpected_success_has_log(self):
 
1772
        result = self._run_test('test_unexpected_success')
 
1773
        self.assertEqual(1, len(result.unexpectedSuccesses))
 
1774
        # Inconsistency, unexpectedSuccesses is a list of tests,
 
1775
        # expectedFailures is a list of reasons?
 
1776
        test = result.unexpectedSuccesses[0]
 
1777
        details = test.getDetails()
 
1778
        self.assertTrue('log' in details)
 
1779
 
 
1780
 
 
1781
class TestTestCloning(tests.TestCase):
 
1782
    """Tests that test cloning of TestCases (as used by multiply_tests)."""
 
1783
 
 
1784
    def test_cloned_testcase_does_not_share_details(self):
 
1785
        """A TestCase cloned with clone_test does not share mutable attributes
 
1786
        such as details or cleanups.
 
1787
        """
 
1788
        class Test(tests.TestCase):
 
1789
            def test_foo(self):
 
1790
                self.addDetail('foo', Content('text/plain', lambda: 'foo'))
 
1791
        orig_test = Test('test_foo')
 
1792
        cloned_test = tests.clone_test(orig_test, orig_test.id() + '(cloned)')
 
1793
        orig_test.run(unittest.TestResult())
 
1794
        self.assertEqual('foo', orig_test.getDetails()['foo'].iter_bytes())
 
1795
        self.assertEqual(None, cloned_test.getDetails().get('foo'))
 
1796
 
 
1797
    def test_double_apply_scenario_preserves_first_scenario(self):
 
1798
        """Applying two levels of scenarios to a test preserves the attributes
 
1799
        added by both scenarios.
 
1800
        """
 
1801
        class Test(tests.TestCase):
 
1802
            def test_foo(self):
 
1803
                pass
 
1804
        test = Test('test_foo')
 
1805
        scenarios_x = [('x=1', {'x': 1}), ('x=2', {'x': 2})]
 
1806
        scenarios_y = [('y=1', {'y': 1}), ('y=2', {'y': 2})]
 
1807
        suite = tests.multiply_tests(test, scenarios_x, unittest.TestSuite())
 
1808
        suite = tests.multiply_tests(suite, scenarios_y, unittest.TestSuite())
 
1809
        all_tests = list(tests.iter_suite_tests(suite))
 
1810
        self.assertLength(4, all_tests)
 
1811
        all_xys = sorted((t.x, t.y) for t in all_tests)
 
1812
        self.assertEqual([(1, 1), (1, 2), (2, 1), (2, 2)], all_xys)
 
1813
 
1657
1814
 
1658
1815
# NB: Don't delete this; it's not actually from 0.11!
1659
1816
@deprecated_function(deprecated_in((0, 11, 0)))
1787
1944
    def test_make_branch_and_tree_with_format(self):
1788
1945
        # we should be able to supply a format to make_branch_and_tree
1789
1946
        self.make_branch_and_tree('a', format=bzrlib.bzrdir.BzrDirMetaFormat1())
1790
 
        self.make_branch_and_tree('b', format=bzrlib.bzrdir.BzrDirFormat6())
1791
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('a')._format,
 
1947
        self.assertIsInstance(bzrlib.controldir.ControlDir.open('a')._format,
1792
1948
                              bzrlib.bzrdir.BzrDirMetaFormat1)
1793
 
        self.assertIsInstance(bzrlib.bzrdir.BzrDir.open('b')._format,
1794
 
                              bzrlib.bzrdir.BzrDirFormat6)
1795
1949
 
1796
1950
    def test_make_branch_and_memory_tree(self):
1797
1951
        # we should be able to get a new branch and a mutable tree from
1814
1968
                tree.branch.repository.bzrdir.root_transport)
1815
1969
 
1816
1970
 
1817
 
class SelfTestHelper:
 
1971
class SelfTestHelper(object):
1818
1972
 
1819
1973
    def run_selftest(self, **kwargs):
1820
1974
        """Run selftest returning its output."""
1874
2028
        self.assertLength(2, output.readlines())
1875
2029
 
1876
2030
    def test_lsprof_tests(self):
1877
 
        self.requireFeature(test_lsprof.LSProfFeature)
1878
 
        calls = []
 
2031
        self.requireFeature(features.lsprof_feature)
 
2032
        results = []
1879
2033
        class Test(object):
1880
2034
            def __call__(test, result):
1881
2035
                test.run(result)
1882
2036
            def run(test, result):
1883
 
                self.assertIsInstance(result, tests.ForwardingResult)
1884
 
                calls.append("called")
 
2037
                results.append(result)
1885
2038
            def countTestCases(self):
1886
2039
                return 1
1887
2040
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
1888
 
        self.assertLength(1, calls)
 
2041
        self.assertLength(1, results)
 
2042
        self.assertIsInstance(results.pop(), ExtendedToOriginalDecorator)
1889
2043
 
1890
2044
    def test_random(self):
1891
2045
        # test randomising by listing a number of tests.
1971
2125
            load_list='missing file name', list_only=True)
1972
2126
 
1973
2127
 
 
2128
class TestSubunitLogDetails(tests.TestCase, SelfTestHelper):
 
2129
 
 
2130
    _test_needs_features = [features.subunit]
 
2131
 
 
2132
    def run_subunit_stream(self, test_name):
 
2133
        from subunit import ProtocolTestCase
 
2134
        def factory():
 
2135
            return TestUtil.TestSuite([_get_test(test_name)])
 
2136
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
 
2137
            test_suite_factory=factory)
 
2138
        test = ProtocolTestCase(stream)
 
2139
        result = testtools.TestResult()
 
2140
        test.run(result)
 
2141
        content = stream.getvalue()
 
2142
        return content, result
 
2143
 
 
2144
    def test_fail_has_log(self):
 
2145
        content, result = self.run_subunit_stream('test_fail')
 
2146
        self.assertEqual(1, len(result.failures))
 
2147
        self.assertContainsRe(content, '(?m)^log$')
 
2148
        self.assertContainsRe(content, 'this test will fail')
 
2149
 
 
2150
    def test_error_has_log(self):
 
2151
        content, result = self.run_subunit_stream('test_error')
 
2152
        self.assertContainsRe(content, '(?m)^log$')
 
2153
        self.assertContainsRe(content, 'this test errored')
 
2154
 
 
2155
    def test_skip_has_no_log(self):
 
2156
        content, result = self.run_subunit_stream('test_skip')
 
2157
        self.assertNotContainsRe(content, '(?m)^log$')
 
2158
        self.assertNotContainsRe(content, 'this test will be skipped')
 
2159
        self.assertEqual(['reason'], result.skip_reasons.keys())
 
2160
        skips = result.skip_reasons['reason']
 
2161
        self.assertEqual(1, len(skips))
 
2162
        test = skips[0]
 
2163
        # RemotedTestCase doesn't preserve the "details"
 
2164
        ## self.assertFalse('log' in test.getDetails())
 
2165
 
 
2166
    def test_missing_feature_has_no_log(self):
 
2167
        content, result = self.run_subunit_stream('test_missing_feature')
 
2168
        self.assertNotContainsRe(content, '(?m)^log$')
 
2169
        self.assertNotContainsRe(content, 'missing the feature')
 
2170
        self.assertEqual(['_MissingFeature\n'], result.skip_reasons.keys())
 
2171
        skips = result.skip_reasons['_MissingFeature\n']
 
2172
        self.assertEqual(1, len(skips))
 
2173
        test = skips[0]
 
2174
        # RemotedTestCase doesn't preserve the "details"
 
2175
        ## self.assertFalse('log' in test.getDetails())
 
2176
 
 
2177
    def test_xfail_has_no_log(self):
 
2178
        content, result = self.run_subunit_stream('test_xfail')
 
2179
        self.assertNotContainsRe(content, '(?m)^log$')
 
2180
        self.assertNotContainsRe(content, 'test with expected failure')
 
2181
        self.assertEqual(1, len(result.expectedFailures))
 
2182
        result_content = result.expectedFailures[0][1]
 
2183
        self.assertNotContainsRe(result_content,
 
2184
            '(?m)^(?:Text attachment: )?log(?:$|: )')
 
2185
        self.assertNotContainsRe(result_content, 'test with expected failure')
 
2186
 
 
2187
    def test_unexpected_success_has_log(self):
 
2188
        content, result = self.run_subunit_stream('test_unexpected_success')
 
2189
        self.assertContainsRe(content, '(?m)^log$')
 
2190
        self.assertContainsRe(content, 'test with unexpected success')
 
2191
        # GZ 2011-05-18: Old versions of subunit treat unexpected success as a
 
2192
        #                success, if a min version check is added remove this
 
2193
        from subunit import TestProtocolClient as _Client
 
2194
        if _Client.addUnexpectedSuccess.im_func is _Client.addSuccess.im_func:
 
2195
            self.expectFailure('subunit treats "unexpectedSuccess"'
 
2196
                               ' as a plain success',
 
2197
                self.assertEqual, 1, len(result.unexpectedSuccesses))
 
2198
        self.assertEqual(1, len(result.unexpectedSuccesses))
 
2199
        test = result.unexpectedSuccesses[0]
 
2200
        # RemotedTestCase doesn't preserve the "details"
 
2201
        ## self.assertTrue('log' in test.getDetails())
 
2202
 
 
2203
    def test_success_has_no_log(self):
 
2204
        content, result = self.run_subunit_stream('test_success')
 
2205
        self.assertEqual(1, result.testsRun)
 
2206
        self.assertNotContainsRe(content, '(?m)^log$')
 
2207
        self.assertNotContainsRe(content, 'this test succeeds')
 
2208
 
 
2209
 
1974
2210
class TestRunBzr(tests.TestCase):
1975
2211
 
1976
2212
    out = ''
2099
2335
        # stdout and stderr of the invoked run_bzr
2100
2336
        current_factory = bzrlib.ui.ui_factory
2101
2337
        self.run_bzr(['foo'])
2102
 
        self.failIf(current_factory is self.factory)
 
2338
        self.assertFalse(current_factory is self.factory)
2103
2339
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2104
2340
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2105
2341
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2262
2498
 
2263
2499
 
2264
2500
class TestStartBzrSubProcess(tests.TestCase):
 
2501
    """Stub test start_bzr_subprocess."""
2265
2502
 
2266
 
    def check_popen_state(self):
2267
 
        """Replace to make assertions when popen is called."""
 
2503
    def _subprocess_log_cleanup(self):
 
2504
        """Inhibits the base version as we don't produce a log file."""
2268
2505
 
2269
2506
    def _popen(self, *args, **kwargs):
2270
 
        """Record the command that is run, so that we can ensure it is correct"""
 
2507
        """Override the base version to record the command that is run.
 
2508
 
 
2509
        From there we can ensure it is correct without spawning a real process.
 
2510
        """
2271
2511
        self.check_popen_state()
2272
2512
        self._popen_args = args
2273
2513
        self._popen_kwargs = kwargs
2274
2514
        raise _DontSpawnProcess()
2275
2515
 
 
2516
    def check_popen_state(self):
 
2517
        """Replace to make assertions when popen is called."""
 
2518
 
2276
2519
    def test_run_bzr_subprocess_no_plugins(self):
2277
2520
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2278
2521
        command = self._popen_args[0]
2282
2525
 
2283
2526
    def test_allow_plugins(self):
2284
2527
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2285
 
            allow_plugins=True)
 
2528
                          allow_plugins=True)
2286
2529
        command = self._popen_args[0]
2287
2530
        self.assertEqual([], command[2:])
2288
2531
 
2289
2532
    def test_set_env(self):
2290
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2533
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2291
2534
        # set in the child
2292
2535
        def check_environment():
2293
2536
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2294
2537
        self.check_popen_state = check_environment
2295
2538
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2296
 
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
 
2539
                          env_changes={'EXISTANT_ENV_VAR':'set variable'})
2297
2540
        # not set in theparent
2298
2541
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2299
2542
 
2300
2543
    def test_run_bzr_subprocess_env_del(self):
2301
2544
        """run_bzr_subprocess can remove environment variables too."""
2302
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2545
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2303
2546
        def check_environment():
2304
2547
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2305
2548
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2306
2549
        self.check_popen_state = check_environment
2307
2550
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2308
 
            env_changes={'EXISTANT_ENV_VAR':None})
 
2551
                          env_changes={'EXISTANT_ENV_VAR':None})
2309
2552
        # Still set in parent
2310
2553
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2311
2554
        del os.environ['EXISTANT_ENV_VAR']
2312
2555
 
2313
2556
    def test_env_del_missing(self):
2314
 
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
 
2557
        self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2315
2558
        def check_environment():
2316
2559
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2317
2560
        self.check_popen_state = check_environment
2318
2561
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2319
 
            env_changes={'NON_EXISTANT_ENV_VAR':None})
 
2562
                          env_changes={'NON_EXISTANT_ENV_VAR':None})
2320
2563
 
2321
2564
    def test_working_dir(self):
2322
2565
        """Test that we can specify the working dir for the child"""
2325
2568
        chdirs = []
2326
2569
        def chdir(path):
2327
2570
            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
 
2571
        self.overrideAttr(os, 'chdir', chdir)
 
2572
        def getcwd():
 
2573
            return 'current'
 
2574
        self.overrideAttr(osutils, 'getcwd', getcwd)
 
2575
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2576
                          working_dir='foo')
2340
2577
        self.assertEqual(['foo', 'current'], chdirs)
2341
2578
 
 
2579
    def test_get_bzr_path_with_cwd_bzrlib(self):
 
2580
        self.get_source_path = lambda: ""
 
2581
        self.overrideAttr(os.path, "isfile", lambda path: True)
 
2582
        self.assertEqual(self.get_bzr_path(), "bzr")
 
2583
 
2342
2584
 
2343
2585
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2344
2586
    """Tests that really need to do things with an external bzr."""
2357
2599
        self.assertEqual('bzr: interrupted\n', result[1])
2358
2600
 
2359
2601
 
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
2602
class TestSelftestFiltering(tests.TestCase):
2435
2603
 
2436
2604
    def setUp(self):
2587
2755
        self.assertEqual(remaining_names, _test_ids(split_suite[1]))
2588
2756
 
2589
2757
 
2590
 
class TestCheckInventoryShape(tests.TestCaseWithTransport):
 
2758
class TestCheckTreeShape(tests.TestCaseWithTransport):
2591
2759
 
2592
 
    def test_check_inventory_shape(self):
 
2760
    def test_check_tree_shape(self):
2593
2761
        files = ['a', 'b/', 'b/c']
2594
2762
        tree = self.make_branch_and_tree('.')
2595
2763
        self.build_tree(files)
2596
2764
        tree.add(files)
2597
2765
        tree.lock_read()
2598
2766
        try:
2599
 
            self.check_inventory_shape(tree.inventory, files)
 
2767
            self.check_tree_shape(tree, files)
2600
2768
        finally:
2601
2769
            tree.unlock()
2602
2770
 
2934
3102
        tpr.register('bar', 'bBB.aAA.rRR')
2935
3103
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
2936
3104
        self.assertThat(self.get_log(),
2937
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
 
3105
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR",
 
3106
                           doctest.ELLIPSIS))
2938
3107
 
2939
3108
    def test_get_unknown_prefix(self):
2940
3109
        tpr = self._get_registry()
2960
3129
        self.assertEquals('bzrlib.plugins', tpr.resolve_alias('bp'))
2961
3130
 
2962
3131
 
 
3132
class TestThreadLeakDetection(tests.TestCase):
 
3133
    """Ensure when tests leak threads we detect and report it"""
 
3134
 
 
3135
    class LeakRecordingResult(tests.ExtendedTestResult):
 
3136
        def __init__(self):
 
3137
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
 
3138
            self.leaks = []
 
3139
        def _report_thread_leak(self, test, leaks, alive):
 
3140
            self.leaks.append((test, leaks))
 
3141
 
 
3142
    def test_testcase_without_addCleanups(self):
 
3143
        """Check old TestCase instances don't break with leak detection"""
 
3144
        class Test(unittest.TestCase):
 
3145
            def runTest(self):
 
3146
                pass
 
3147
        result = self.LeakRecordingResult()
 
3148
        test = Test()
 
3149
        result.startTestRun()
 
3150
        test.run(result)
 
3151
        result.stopTestRun()
 
3152
        self.assertEqual(result._tests_leaking_threads_count, 0)
 
3153
        self.assertEqual(result.leaks, [])
 
3154
        
 
3155
    def test_thread_leak(self):
 
3156
        """Ensure a thread that outlives the running of a test is reported
 
3157
 
 
3158
        Uses a thread that blocks on an event, and is started by the inner
 
3159
        test case. As the thread outlives the inner case's run, it should be
 
3160
        detected as a leak, but the event is then set so that the thread can
 
3161
        be safely joined in cleanup so it's not leaked for real.
 
3162
        """
 
3163
        event = threading.Event()
 
3164
        thread = threading.Thread(name="Leaker", target=event.wait)
 
3165
        class Test(tests.TestCase):
 
3166
            def test_leak(self):
 
3167
                thread.start()
 
3168
        result = self.LeakRecordingResult()
 
3169
        test = Test("test_leak")
 
3170
        self.addCleanup(thread.join)
 
3171
        self.addCleanup(event.set)
 
3172
        result.startTestRun()
 
3173
        test.run(result)
 
3174
        result.stopTestRun()
 
3175
        self.assertEqual(result._tests_leaking_threads_count, 1)
 
3176
        self.assertEqual(result._first_thread_leaker_id, test.id())
 
3177
        self.assertEqual(result.leaks, [(test, set([thread]))])
 
3178
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
 
3179
 
 
3180
    def test_multiple_leaks(self):
 
3181
        """Check multiple leaks are blamed on the test cases at fault
 
3182
 
 
3183
        Same concept as the previous test, but has one inner test method that
 
3184
        leaks two threads, and one that doesn't leak at all.
 
3185
        """
 
3186
        event = threading.Event()
 
3187
        thread_a = threading.Thread(name="LeakerA", target=event.wait)
 
3188
        thread_b = threading.Thread(name="LeakerB", target=event.wait)
 
3189
        thread_c = threading.Thread(name="LeakerC", target=event.wait)
 
3190
        class Test(tests.TestCase):
 
3191
            def test_first_leak(self):
 
3192
                thread_b.start()
 
3193
            def test_second_no_leak(self):
 
3194
                pass
 
3195
            def test_third_leak(self):
 
3196
                thread_c.start()
 
3197
                thread_a.start()
 
3198
        result = self.LeakRecordingResult()
 
3199
        first_test = Test("test_first_leak")
 
3200
        third_test = Test("test_third_leak")
 
3201
        self.addCleanup(thread_a.join)
 
3202
        self.addCleanup(thread_b.join)
 
3203
        self.addCleanup(thread_c.join)
 
3204
        self.addCleanup(event.set)
 
3205
        result.startTestRun()
 
3206
        unittest.TestSuite(
 
3207
            [first_test, Test("test_second_no_leak"), third_test]
 
3208
            ).run(result)
 
3209
        result.stopTestRun()
 
3210
        self.assertEqual(result._tests_leaking_threads_count, 2)
 
3211
        self.assertEqual(result._first_thread_leaker_id, first_test.id())
 
3212
        self.assertEqual(result.leaks, [
 
3213
            (first_test, set([thread_b])),
 
3214
            (third_test, set([thread_a, thread_c]))])
 
3215
        self.assertContainsString(result.stream.getvalue(), "leaking threads")
 
3216
 
 
3217
 
 
3218
class TestPostMortemDebugging(tests.TestCase):
 
3219
    """Check post mortem debugging works when tests fail or error"""
 
3220
 
 
3221
    class TracebackRecordingResult(tests.ExtendedTestResult):
 
3222
        def __init__(self):
 
3223
            tests.ExtendedTestResult.__init__(self, StringIO(), 0, 1)
 
3224
            self.postcode = None
 
3225
        def _post_mortem(self, tb=None):
 
3226
            """Record the code object at the end of the current traceback"""
 
3227
            tb = tb or sys.exc_info()[2]
 
3228
            if tb is not None:
 
3229
                next = tb.tb_next
 
3230
                while next is not None:
 
3231
                    tb = next
 
3232
                    next = next.tb_next
 
3233
                self.postcode = tb.tb_frame.f_code
 
3234
        def report_error(self, test, err):
 
3235
            pass
 
3236
        def report_failure(self, test, err):
 
3237
            pass
 
3238
 
 
3239
    def test_location_unittest_error(self):
 
3240
        """Needs right post mortem traceback with erroring unittest case"""
 
3241
        class Test(unittest.TestCase):
 
3242
            def runTest(self):
 
3243
                raise RuntimeError
 
3244
        result = self.TracebackRecordingResult()
 
3245
        Test().run(result)
 
3246
        self.assertEqual(result.postcode, Test.runTest.func_code)
 
3247
 
 
3248
    def test_location_unittest_failure(self):
 
3249
        """Needs right post mortem traceback with failing unittest case"""
 
3250
        class Test(unittest.TestCase):
 
3251
            def runTest(self):
 
3252
                raise self.failureException
 
3253
        result = self.TracebackRecordingResult()
 
3254
        Test().run(result)
 
3255
        self.assertEqual(result.postcode, Test.runTest.func_code)
 
3256
 
 
3257
    def test_location_bt_error(self):
 
3258
        """Needs right post mortem traceback with erroring bzrlib.tests case"""
 
3259
        class Test(tests.TestCase):
 
3260
            def test_error(self):
 
3261
                raise RuntimeError
 
3262
        result = self.TracebackRecordingResult()
 
3263
        Test("test_error").run(result)
 
3264
        self.assertEqual(result.postcode, Test.test_error.func_code)
 
3265
 
 
3266
    def test_location_bt_failure(self):
 
3267
        """Needs right post mortem traceback with failing bzrlib.tests case"""
 
3268
        class Test(tests.TestCase):
 
3269
            def test_failure(self):
 
3270
                raise self.failureException
 
3271
        result = self.TracebackRecordingResult()
 
3272
        Test("test_failure").run(result)
 
3273
        self.assertEqual(result.postcode, Test.test_failure.func_code)
 
3274
 
 
3275
    def test_env_var_triggers_post_mortem(self):
 
3276
        """Check pdb.post_mortem is called iff BZR_TEST_PDB is set"""
 
3277
        import pdb
 
3278
        result = tests.ExtendedTestResult(StringIO(), 0, 1)
 
3279
        post_mortem_calls = []
 
3280
        self.overrideAttr(pdb, "post_mortem", post_mortem_calls.append)
 
3281
        self.overrideEnv('BZR_TEST_PDB', None)
 
3282
        result._post_mortem(1)
 
3283
        self.overrideEnv('BZR_TEST_PDB', 'on')
 
3284
        result._post_mortem(2)
 
3285
        self.assertEqual([2], post_mortem_calls)
 
3286
 
 
3287
 
2963
3288
class TestRunSuite(tests.TestCase):
2964
3289
 
2965
3290
    def test_runner_class(self):
2976
3301
                                                self.verbosity)
2977
3302
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2978
3303
        self.assertLength(1, calls)
 
3304
 
 
3305
 
 
3306
class _Selftest(object):
 
3307
    """Mixin for tests needing full selftest output"""
 
3308
 
 
3309
    def _inject_stream_into_subunit(self, stream):
 
3310
        """To be overridden by subclasses that run tests out of process"""
 
3311
 
 
3312
    def _run_selftest(self, **kwargs):
 
3313
        sio = StringIO()
 
3314
        self._inject_stream_into_subunit(sio)
 
3315
        tests.selftest(stream=sio, stop_on_failure=False, **kwargs)
 
3316
        return sio.getvalue()
 
3317
 
 
3318
 
 
3319
class _ForkedSelftest(_Selftest):
 
3320
    """Mixin for tests needing full selftest output with forked children"""
 
3321
 
 
3322
    _test_needs_features = [features.subunit]
 
3323
 
 
3324
    def _inject_stream_into_subunit(self, stream):
 
3325
        """Monkey-patch subunit so the extra output goes to stream not stdout
 
3326
 
 
3327
        Some APIs need rewriting so this kind of bogus hackery can be replaced
 
3328
        by passing the stream param from run_tests down into ProtocolTestCase.
 
3329
        """
 
3330
        from subunit import ProtocolTestCase
 
3331
        _original_init = ProtocolTestCase.__init__
 
3332
        def _init_with_passthrough(self, *args, **kwargs):
 
3333
            _original_init(self, *args, **kwargs)
 
3334
            self._passthrough = stream
 
3335
        self.overrideAttr(ProtocolTestCase, "__init__", _init_with_passthrough)
 
3336
 
 
3337
    def _run_selftest(self, **kwargs):
 
3338
        # GZ 2011-05-26: Add a PosixSystem feature so this check can go away
 
3339
        if getattr(os, "fork", None) is None:
 
3340
            raise tests.TestNotApplicable("Platform doesn't support forking")
 
3341
        # Make sure the fork code is actually invoked by claiming two cores
 
3342
        self.overrideAttr(osutils, "local_concurrency", lambda: 2)
 
3343
        kwargs.setdefault("suite_decorators", []).append(tests.fork_decorator)
 
3344
        return super(_ForkedSelftest, self)._run_selftest(**kwargs)
 
3345
 
 
3346
 
 
3347
class TestParallelFork(_ForkedSelftest, tests.TestCase):
 
3348
    """Check operation of --parallel=fork selftest option"""
 
3349
 
 
3350
    def test_error_in_child_during_fork(self):
 
3351
        """Error in a forked child during test setup should get reported"""
 
3352
        class Test(tests.TestCase):
 
3353
            def testMethod(self):
 
3354
                pass
 
3355
        # We don't care what, just break something that a child will run
 
3356
        self.overrideAttr(tests, "workaround_zealous_crypto_random", None)
 
3357
        out = self._run_selftest(test_suite_factory=Test)
 
3358
        # Lines from the tracebacks of the two child processes may be mixed
 
3359
        # together due to the way subunit parses and forwards the streams,
 
3360
        # so permit extra lines between each part of the error output.
 
3361
        self.assertContainsRe(out,
 
3362
            "Traceback.*:\n"
 
3363
            "(?:.*\n)*"
 
3364
            ".+ in fork_for_tests\n"
 
3365
            "(?:.*\n)*"
 
3366
            "\s*workaround_zealous_crypto_random\(\)\n"
 
3367
            "(?:.*\n)*"
 
3368
            "TypeError:")
 
3369
 
 
3370
 
 
3371
class TestUncollectedWarnings(_Selftest, tests.TestCase):
 
3372
    """Check a test case still alive after being run emits a warning"""
 
3373
 
 
3374
    class Test(tests.TestCase):
 
3375
        def test_pass(self):
 
3376
            pass
 
3377
        def test_self_ref(self):
 
3378
            self.also_self = self.test_self_ref
 
3379
        def test_skip(self):
 
3380
            self.skip("Don't need")
 
3381
 
 
3382
    def _get_suite(self):
 
3383
        return TestUtil.TestSuite([
 
3384
            self.Test("test_pass"),
 
3385
            self.Test("test_self_ref"),
 
3386
            self.Test("test_skip"),
 
3387
            ])
 
3388
 
 
3389
    def _run_selftest_with_suite(self, **kwargs):
 
3390
        old_flags = tests.selftest_debug_flags
 
3391
        tests.selftest_debug_flags = old_flags.union(["uncollected_cases"])
 
3392
        gc_on = gc.isenabled()
 
3393
        if gc_on:
 
3394
            gc.disable()
 
3395
        try:
 
3396
            output = self._run_selftest(test_suite_factory=self._get_suite,
 
3397
                **kwargs)
 
3398
        finally:
 
3399
            if gc_on:
 
3400
                gc.enable()
 
3401
            tests.selftest_debug_flags = old_flags
 
3402
        self.assertNotContainsRe(output, "Uncollected test case.*test_pass")
 
3403
        self.assertContainsRe(output, "Uncollected test case.*test_self_ref")
 
3404
        return output
 
3405
 
 
3406
    def test_testsuite(self):
 
3407
        self._run_selftest_with_suite()
 
3408
 
 
3409
    def test_pattern(self):
 
3410
        out = self._run_selftest_with_suite(pattern="test_(?:pass|self_ref)$")
 
3411
        self.assertNotContainsRe(out, "test_skip")
 
3412
 
 
3413
    def test_exclude_pattern(self):
 
3414
        out = self._run_selftest_with_suite(exclude_pattern="test_skip$")
 
3415
        self.assertNotContainsRe(out, "test_skip")
 
3416
 
 
3417
    def test_random_seed(self):
 
3418
        self._run_selftest_with_suite(random_seed="now")
 
3419
 
 
3420
    def test_matching_tests_first(self):
 
3421
        self._run_selftest_with_suite(matching_tests_first=True,
 
3422
            pattern="test_self_ref$")
 
3423
 
 
3424
    def test_starting_with_and_exclude(self):
 
3425
        out = self._run_selftest_with_suite(starting_with=["bt."],
 
3426
            exclude_pattern="test_skip$")
 
3427
        self.assertNotContainsRe(out, "test_skip")
 
3428
 
 
3429
    def test_additonal_decorator(self):
 
3430
        out = self._run_selftest_with_suite(
 
3431
            suite_decorators=[tests.TestDecorator])
 
3432
 
 
3433
 
 
3434
class TestUncollectedWarningsSubunit(TestUncollectedWarnings):
 
3435
    """Check warnings from tests staying alive are emitted with subunit"""
 
3436
 
 
3437
    _test_needs_features = [features.subunit]
 
3438
 
 
3439
    def _run_selftest_with_suite(self, **kwargs):
 
3440
        return TestUncollectedWarnings._run_selftest_with_suite(self,
 
3441
            runner_class=tests.SubUnitBzrRunner, **kwargs)
 
3442
 
 
3443
 
 
3444
class TestUncollectedWarningsForked(_ForkedSelftest, TestUncollectedWarnings):
 
3445
    """Check warnings from tests staying alive are emitted when forking"""
 
3446
 
 
3447
 
 
3448
class TestEnvironHandling(tests.TestCase):
 
3449
 
 
3450
    def test_overrideEnv_None_called_twice_doesnt_leak(self):
 
3451
        self.assertFalse('MYVAR' in os.environ)
 
3452
        self.overrideEnv('MYVAR', '42')
 
3453
        # We use an embedded test to make sure we fix the _captureVar bug
 
3454
        class Test(tests.TestCase):
 
3455
            def test_me(self):
 
3456
                # The first call save the 42 value
 
3457
                self.overrideEnv('MYVAR', None)
 
3458
                self.assertEquals(None, os.environ.get('MYVAR'))
 
3459
                # Make sure we can call it twice
 
3460
                self.overrideEnv('MYVAR', None)
 
3461
                self.assertEquals(None, os.environ.get('MYVAR'))
 
3462
        output = StringIO()
 
3463
        result = tests.TextTestResult(output, 0, 1)
 
3464
        Test('test_me').run(result)
 
3465
        if not result.wasStrictlySuccessful():
 
3466
            self.fail(output.getvalue())
 
3467
        # We get our value back
 
3468
        self.assertEquals('42', os.environ.get('MYVAR'))
 
3469
 
 
3470
 
 
3471
class TestIsolatedEnv(tests.TestCase):
 
3472
    """Test isolating tests from os.environ.
 
3473
 
 
3474
    Since we use tests that are already isolated from os.environ a bit of care
 
3475
    should be taken when designing the tests to avoid bootstrap side-effects.
 
3476
    The tests start an already clean os.environ which allow doing valid
 
3477
    assertions about which variables are present or not and design tests around
 
3478
    these assertions.
 
3479
    """
 
3480
 
 
3481
    class ScratchMonkey(tests.TestCase):
 
3482
 
 
3483
        def test_me(self):
 
3484
            pass
 
3485
 
 
3486
    def test_basics(self):
 
3487
        # Make sure we know the definition of BZR_HOME: not part of os.environ
 
3488
        # for tests.TestCase.
 
3489
        self.assertTrue('BZR_HOME' in tests.isolated_environ)
 
3490
        self.assertEquals(None, tests.isolated_environ['BZR_HOME'])
 
3491
        # Being part of isolated_environ, BZR_HOME should not appear here
 
3492
        self.assertFalse('BZR_HOME' in os.environ)
 
3493
        # Make sure we know the definition of LINES: part of os.environ for
 
3494
        # tests.TestCase
 
3495
        self.assertTrue('LINES' in tests.isolated_environ)
 
3496
        self.assertEquals('25', tests.isolated_environ['LINES'])
 
3497
        self.assertEquals('25', os.environ['LINES'])
 
3498
 
 
3499
    def test_injecting_unknown_variable(self):
 
3500
        # BZR_HOME is known to be absent from os.environ
 
3501
        test = self.ScratchMonkey('test_me')
 
3502
        tests.override_os_environ(test, {'BZR_HOME': 'foo'})
 
3503
        self.assertEquals('foo', os.environ['BZR_HOME'])
 
3504
        tests.restore_os_environ(test)
 
3505
        self.assertFalse('BZR_HOME' in os.environ)
 
3506
 
 
3507
    def test_injecting_known_variable(self):
 
3508
        test = self.ScratchMonkey('test_me')
 
3509
        # LINES is known to be present in os.environ
 
3510
        tests.override_os_environ(test, {'LINES': '42'})
 
3511
        self.assertEquals('42', os.environ['LINES'])
 
3512
        tests.restore_os_environ(test)
 
3513
        self.assertEquals('25', os.environ['LINES'])
 
3514
 
 
3515
    def test_deleting_variable(self):
 
3516
        test = self.ScratchMonkey('test_me')
 
3517
        # LINES is known to be present in os.environ
 
3518
        tests.override_os_environ(test, {'LINES': None})
 
3519
        self.assertTrue('LINES' not in os.environ)
 
3520
        tests.restore_os_environ(test)
 
3521
        self.assertEquals('25', os.environ['LINES'])
 
3522
 
 
3523
 
 
3524
class TestDocTestSuiteIsolation(tests.TestCase):
 
3525
    """Test that `tests.DocTestSuite` isolates doc tests from os.environ.
 
3526
 
 
3527
    Since tests.TestCase alreay provides an isolation from os.environ, we use
 
3528
    the clean environment as a base for testing. To precisely capture the
 
3529
    isolation provided by tests.DocTestSuite, we use doctest.DocTestSuite to
 
3530
    compare against.
 
3531
 
 
3532
    We want to make sure `tests.DocTestSuite` respect `tests.isolated_environ`,
 
3533
    not `os.environ` so each test overrides it to suit its needs.
 
3534
 
 
3535
    """
 
3536
 
 
3537
    def get_doctest_suite_for_string(self, klass, string):
 
3538
        class Finder(doctest.DocTestFinder):
 
3539
 
 
3540
            def find(*args, **kwargs):
 
3541
                test = doctest.DocTestParser().get_doctest(
 
3542
                    string, {}, 'foo', 'foo.py', 0)
 
3543
                return [test]
 
3544
 
 
3545
        suite = klass(test_finder=Finder())
 
3546
        return suite
 
3547
 
 
3548
    def run_doctest_suite_for_string(self, klass, string):
 
3549
        suite = self.get_doctest_suite_for_string(klass, string)
 
3550
        output = StringIO()
 
3551
        result = tests.TextTestResult(output, 0, 1)
 
3552
        suite.run(result)
 
3553
        return result, output
 
3554
 
 
3555
    def assertDocTestStringSucceds(self, klass, string):
 
3556
        result, output = self.run_doctest_suite_for_string(klass, string)
 
3557
        if not result.wasStrictlySuccessful():
 
3558
            self.fail(output.getvalue())
 
3559
 
 
3560
    def assertDocTestStringFails(self, klass, string):
 
3561
        result, output = self.run_doctest_suite_for_string(klass, string)
 
3562
        if result.wasStrictlySuccessful():
 
3563
            self.fail(output.getvalue())
 
3564
 
 
3565
    def test_injected_variable(self):
 
3566
        self.overrideAttr(tests, 'isolated_environ', {'LINES': '42'})
 
3567
        test = """
 
3568
            >>> import os
 
3569
            >>> os.environ['LINES']
 
3570
            '42'
 
3571
            """
 
3572
        # doctest.DocTestSuite fails as it sees '25'
 
3573
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
 
3574
        # tests.DocTestSuite sees '42'
 
3575
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
 
3576
 
 
3577
    def test_deleted_variable(self):
 
3578
        self.overrideAttr(tests, 'isolated_environ', {'LINES': None})
 
3579
        test = """
 
3580
            >>> import os
 
3581
            >>> os.environ.get('LINES')
 
3582
            """
 
3583
        # doctest.DocTestSuite fails as it sees '25'
 
3584
        self.assertDocTestStringFails(doctest.DocTestSuite, test)
 
3585
        # tests.DocTestSuite sees None
 
3586
        self.assertDocTestStringSucceds(tests.IsolatedDocTestSuite, test)
 
3587
 
 
3588
 
 
3589
class TestSelftestExcludePatterns(tests.TestCase):
 
3590
 
 
3591
    def setUp(self):
 
3592
        super(TestSelftestExcludePatterns, self).setUp()
 
3593
        self.overrideAttr(tests, 'test_suite', self.suite_factory)
 
3594
 
 
3595
    def suite_factory(self, keep_only=None, starting_with=None):
 
3596
        """A test suite factory with only a few tests."""
 
3597
        class Test(tests.TestCase):
 
3598
            def id(self):
 
3599
                # We don't need the full class path
 
3600
                return self._testMethodName
 
3601
            def a(self):
 
3602
                pass
 
3603
            def b(self):
 
3604
                pass
 
3605
            def c(self):
 
3606
                pass
 
3607
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
 
3608
 
 
3609
    def assertTestList(self, expected, *selftest_args):
 
3610
        # We rely on setUp installing the right test suite factory so we can
 
3611
        # test at the command level without loading the whole test suite
 
3612
        out, err = self.run_bzr(('selftest', '--list') + selftest_args)
 
3613
        actual = out.splitlines()
 
3614
        self.assertEquals(expected, actual)
 
3615
 
 
3616
    def test_full_list(self):
 
3617
        self.assertTestList(['a', 'b', 'c'])
 
3618
 
 
3619
    def test_single_exclude(self):
 
3620
        self.assertTestList(['b', 'c'], '-x', 'a')
 
3621
 
 
3622
    def test_mutiple_excludes(self):
 
3623
        self.assertTestList(['c'], '-x', 'a', '-x', 'b')
 
3624
 
 
3625
 
 
3626
class TestCounterHooks(tests.TestCase, SelfTestHelper):
 
3627
 
 
3628
    _test_needs_features = [features.subunit]
 
3629
 
 
3630
    def setUp(self):
 
3631
        super(TestCounterHooks, self).setUp()
 
3632
        class Test(tests.TestCase):
 
3633
 
 
3634
            def setUp(self):
 
3635
                super(Test, self).setUp()
 
3636
                self.hooks = hooks.Hooks()
 
3637
                self.hooks.add_hook('myhook', 'Foo bar blah', (2,4))
 
3638
                self.install_counter_hook(self.hooks, 'myhook')
 
3639
 
 
3640
            def no_hook(self):
 
3641
                pass
 
3642
 
 
3643
            def run_hook_once(self):
 
3644
                for hook in self.hooks['myhook']:
 
3645
                    hook(self)
 
3646
 
 
3647
        self.test_class = Test
 
3648
 
 
3649
    def assertHookCalls(self, expected_calls, test_name):
 
3650
        test = self.test_class(test_name)
 
3651
        result = unittest.TestResult()
 
3652
        test.run(result)
 
3653
        self.assertTrue(hasattr(test, '_counters'))
 
3654
        self.assertTrue(test._counters.has_key('myhook'))
 
3655
        self.assertEquals(expected_calls, test._counters['myhook'])
 
3656
 
 
3657
    def test_no_hook(self):
 
3658
        self.assertHookCalls(0, 'no_hook')
 
3659
 
 
3660
    def test_run_hook_once(self):
 
3661
        tt = features.testtools
 
3662
        if tt.module.__version__ < (0, 9, 8):
 
3663
            raise tests.TestSkipped('testtools-0.9.8 required for addDetail')
 
3664
        self.assertHookCalls(1, 'run_hook_once')