← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2010-02-11 06:15:33 UTC
  • mfrom: (5025.1.6 331095-malloc)
  • Revision ID: pqm@pqm.ubuntu.com-20100211061533-5glf4faoutadhql9
(mbp) avoid malloc(0)

Show diffs side-by-side

added added

removed removed

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