~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2011-08-17 18:13:57 UTC
  • mfrom: (5268.7.29 transport-segments)
  • Revision ID: pqm@pqm.ubuntu.com-20110817181357-y5q5eth1hk8bl3om
(jelmer) Allow specifying the colocated branch to use in the branch URL,
 and retrieving the branch name using ControlDir._get_selected_branch.
 (Jelmer Vernooij)

Show diffs side-by-side

added added

removed removed

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