~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: 2009-08-13 14:20:03 UTC
  • mfrom: (4599.2.4 windows-installer)
  • Revision ID: pqm@pqm.ubuntu.com-20090813142003-3x748ymw3avzmme7
(jam) Updates to the windows installers.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005-2010 Canonical Ltd
 
1
# Copyright (C) 2005, 2006, 2007, 2008, 2009 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
21
20
import os
22
 
import signal
23
21
import sys
24
22
import time
25
23
import unittest
26
24
import warnings
27
25
 
28
 
from testtools import MultiTestResult
29
 
from testtools.content_type import ContentType
30
 
from testtools.matchers import (
31
 
    DocTestMatches,
32
 
    Equals,
33
 
    )
34
 
import testtools.tests.helpers
35
 
 
36
26
import bzrlib
37
27
from bzrlib import (
38
28
    branchbuilder,
47
37
    repository,
48
38
    symbol_versioning,
49
39
    tests,
50
 
    transport,
51
40
    workingtree,
52
41
    )
53
42
from bzrlib.repofmt import (
54
 
    groupcompress_repo,
55
43
    pack_repo,
56
44
    weaverepo,
57
45
    )
61
49
    deprecated_method,
62
50
    )
63
51
from bzrlib.tests import (
64
 
    features,
65
52
    test_lsprof,
66
 
    test_server,
67
53
    test_sftp_transport,
68
54
    TestUtil,
69
55
    )
70
56
from bzrlib.trace import note
71
 
from bzrlib.transport import memory
 
57
from bzrlib.transport.memory import MemoryServer, MemoryTransport
72
58
from bzrlib.version import _get_bzr_source_tree
73
59
 
74
60
 
88
74
                          TestUtil._load_module_by_name,
89
75
                          'bzrlib.no-name-yet')
90
76
 
91
 
 
92
77
class MetaTestLog(tests.TestCase):
93
78
 
94
79
    def test_logging(self):
95
80
        """Test logs are captured when a test fails."""
96
81
        self.log('a test message')
97
 
        details = self.getDetails()
98
 
        log = details['log']
99
 
        self.assertThat(log.content_type, Equals(ContentType(
100
 
            "text", "plain", {"charset": "utf8"})))
101
 
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
102
 
        self.assertThat(self.get_log(),
103
 
            DocTestMatches(u"...a test message\n", ELLIPSIS))
 
82
        self._log_file.flush()
 
83
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
84
                              'a test message\n')
104
85
 
105
86
 
106
87
class TestUnicodeFilename(tests.TestCase):
143
124
        self.assertEqual(sample_permutation,
144
125
                         get_transport_test_permutations(MockModule()))
145
126
 
146
 
    def test_scenarios_include_all_modules(self):
 
127
    def test_scenarios_invlude_all_modules(self):
147
128
        # this checks that the scenario generator returns as many permutations
148
129
        # as there are in all the registered transport modules - we assume if
149
130
        # this matches its probably doing the right thing especially in
208
189
    def test_scenarios(self):
209
190
        # check that constructor parameters are passed through to the adapted
210
191
        # test.
211
 
        from bzrlib.tests.per_controldir import make_scenarios
 
192
        from bzrlib.tests.per_bzrdir import make_scenarios
212
193
        vfs_factory = "v"
213
194
        server1 = "a"
214
195
        server2 = "b"
234
215
        from bzrlib.tests.per_repository import formats_to_scenarios
235
216
        formats = [("(c)", remote.RemoteRepositoryFormat()),
236
217
                   ("(d)", repository.format_registry.get(
237
 
                    'Bazaar repository format 2a (needs bzr 1.16 or later)\n'))]
 
218
                        'Bazaar pack repository format 1 (needs bzr 0.92)\n'))]
238
219
        no_vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
239
220
            None)
240
221
        vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
241
222
            vfs_transport_factory="vfs")
242
223
        # no_vfs generate scenarios without vfs_transport_factory
243
 
        expected = [
 
224
        self.assertEqual([
244
225
            ('RemoteRepositoryFormat(c)',
245
226
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
246
227
              'repository_format': remote.RemoteRepositoryFormat(),
247
228
              'transport_readonly_server': 'readonly',
248
229
              'transport_server': 'server'}),
249
 
            ('RepositoryFormat2a(d)',
 
230
            ('RepositoryFormatKnitPack1(d)',
250
231
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
251
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
232
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
252
233
              'transport_readonly_server': 'readonly',
253
 
              'transport_server': 'server'})]
254
 
        self.assertEqual(expected, no_vfs_scenarios)
 
234
              'transport_server': 'server'})],
 
235
            no_vfs_scenarios)
255
236
        self.assertEqual([
256
237
            ('RemoteRepositoryFormat(c)',
257
238
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
259
240
              'transport_readonly_server': 'readonly',
260
241
              'transport_server': 'server',
261
242
              'vfs_transport_factory': 'vfs'}),
262
 
            ('RepositoryFormat2a(d)',
 
243
            ('RepositoryFormatKnitPack1(d)',
263
244
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
264
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
245
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
265
246
              'transport_readonly_server': 'readonly',
266
247
              'transport_server': 'server',
267
248
              'vfs_transport_factory': 'vfs'})],
312
293
        from bzrlib.tests.per_interrepository import make_scenarios
313
294
        server1 = "a"
314
295
        server2 = "b"
315
 
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
 
296
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
316
297
        scenarios = make_scenarios(server1, server2, formats)
317
298
        self.assertEqual([
318
 
            ('C0,str,str',
319
 
             {'repository_format': 'C1',
 
299
            ('str,str,str',
 
300
             {'interrepo_class': str,
 
301
              'repository_format': 'C1',
320
302
              'repository_format_to': 'C2',
321
303
              'transport_readonly_server': 'b',
322
304
              'transport_server': 'a'}),
323
 
            ('D0,str,str',
324
 
             {'repository_format': 'D1',
 
305
            ('int,str,str',
 
306
             {'interrepo_class': int,
 
307
              'repository_format': 'D1',
325
308
              'repository_format_to': 'D2',
326
309
              'transport_readonly_server': 'b',
327
310
              'transport_server': 'a'})],
505
488
        self.assertEqualStat(real, fake)
506
489
 
507
490
    def test_assertEqualStat_notequal(self):
508
 
        self.build_tree(["foo", "longname"])
 
491
        self.build_tree(["foo", "bar"])
509
492
        self.assertRaises(AssertionError, self.assertEqualStat,
510
 
            os.lstat("foo"), os.lstat("longname"))
 
493
            os.lstat("foo"), os.lstat("bar"))
511
494
 
512
495
 
513
496
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
531
514
        cwd = osutils.getcwd()
532
515
        self.assertIsSameRealPath(self.test_dir, cwd)
533
516
 
534
 
    def test_BZR_HOME_and_HOME_are_bytestrings(self):
535
 
        """The $BZR_HOME and $HOME environment variables should not be unicode.
536
 
 
537
 
        See https://bugs.launchpad.net/bzr/+bug/464174
538
 
        """
539
 
        self.assertIsInstance(os.environ['BZR_HOME'], str)
540
 
        self.assertIsInstance(os.environ['HOME'], str)
541
 
 
542
517
    def test_make_branch_and_memory_tree(self):
543
518
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
544
519
 
600
575
                         self.get_transport().get_bytes(
601
576
                            'dir/.bzr/repository/format'))
602
577
 
 
578
    def test_safety_net(self):
 
579
        """No test should modify the safety .bzr directory.
 
580
 
 
581
        We just test that the _check_safety_net private method raises
 
582
        AssertionError, it's easier than building a test suite with the same
 
583
        test.
 
584
        """
 
585
        # Oops, a commit in the current directory (i.e. without local .bzr
 
586
        # directory) will crawl up the hierarchy to find a .bzr directory.
 
587
        self.run_bzr(['commit', '-mfoo', '--unchanged'])
 
588
        # But we have a safety net in place.
 
589
        self.assertRaises(AssertionError, self._check_safety_net)
 
590
 
603
591
    def test_dangling_locks_cause_failures(self):
604
592
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
605
593
            def test_function(self):
609
597
                l.attempt_lock()
610
598
        test = TestDanglingLock('test_function')
611
599
        result = test.run()
612
 
        total_failures = result.errors + result.failures
613
 
        if self._lock_check_thorough:
614
 
            self.assertLength(1, total_failures)
615
 
        else:
616
 
            # When _lock_check_thorough is disabled, then we don't trigger a
617
 
            # failure
618
 
            self.assertLength(0, total_failures)
 
600
        self.assertEqual(1, len(result.errors))
619
601
 
620
602
 
621
603
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
622
604
    """Tests for the convenience functions TestCaseWithTransport introduces."""
623
605
 
624
606
    def test_get_readonly_url_none(self):
 
607
        from bzrlib.transport import get_transport
 
608
        from bzrlib.transport.memory import MemoryServer
625
609
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
626
 
        self.vfs_transport_factory = memory.MemoryServer
 
610
        self.vfs_transport_factory = MemoryServer
627
611
        self.transport_readonly_server = None
628
612
        # calling get_readonly_transport() constructs a decorator on the url
629
613
        # for the server
630
614
        url = self.get_readonly_url()
631
615
        url2 = self.get_readonly_url('foo/bar')
632
 
        t = transport.get_transport(url)
633
 
        t2 = transport.get_transport(url2)
 
616
        t = get_transport(url)
 
617
        t2 = get_transport(url2)
634
618
        self.failUnless(isinstance(t, ReadonlyTransportDecorator))
635
619
        self.failUnless(isinstance(t2, ReadonlyTransportDecorator))
636
620
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
637
621
 
638
622
    def test_get_readonly_url_http(self):
639
623
        from bzrlib.tests.http_server import HttpServer
 
624
        from bzrlib.transport import get_transport
 
625
        from bzrlib.transport.local import LocalURLServer
640
626
        from bzrlib.transport.http import HttpTransportBase
641
 
        self.transport_server = test_server.LocalURLServer
 
627
        self.transport_server = LocalURLServer
642
628
        self.transport_readonly_server = HttpServer
643
629
        # calling get_readonly_transport() gives us a HTTP server instance.
644
630
        url = self.get_readonly_url()
645
631
        url2 = self.get_readonly_url('foo/bar')
646
632
        # the transport returned may be any HttpTransportBase subclass
647
 
        t = transport.get_transport(url)
648
 
        t2 = transport.get_transport(url2)
 
633
        t = get_transport(url)
 
634
        t2 = get_transport(url2)
649
635
        self.failUnless(isinstance(t, HttpTransportBase))
650
636
        self.failUnless(isinstance(t2, HttpTransportBase))
651
637
        self.assertEqual(t2.base[:-1], t.abspath('foo/bar'))
675
661
 
676
662
    def setUp(self):
677
663
        super(TestTestCaseTransports, self).setUp()
678
 
        self.vfs_transport_factory = memory.MemoryServer
 
664
        self.vfs_transport_factory = MemoryServer
679
665
 
680
666
    def test_make_bzrdir_preserves_transport(self):
681
667
        t = self.get_transport()
682
668
        result_bzrdir = self.make_bzrdir('subdir')
683
669
        self.assertIsInstance(result_bzrdir.transport,
684
 
                              memory.MemoryTransport)
 
670
                              MemoryTransport)
685
671
        # should not be on disk, should only be in memory
686
672
        self.failIfExists('subdir')
687
673
 
689
675
class TestChrootedTest(tests.ChrootedTestCase):
690
676
 
691
677
    def test_root_is_root(self):
692
 
        t = transport.get_transport(self.get_readonly_url())
 
678
        from bzrlib.transport import get_transport
 
679
        t = get_transport(self.get_readonly_url())
693
680
        url = t.base
694
681
        self.assertEqual(url, t.clone('..').base)
695
682
 
696
683
 
697
 
class TestProfileResult(tests.TestCase):
698
 
 
699
 
    def test_profiles_tests(self):
700
 
        self.requireFeature(test_lsprof.LSProfFeature)
701
 
        terminal = testtools.tests.helpers.ExtendedTestResult()
702
 
        result = tests.ProfileResult(terminal)
703
 
        class Sample(tests.TestCase):
704
 
            def a(self):
705
 
                self.sample_function()
706
 
            def sample_function(self):
707
 
                pass
708
 
        test = Sample("a")
709
 
        test.run(result)
710
 
        case = terminal._events[0][1]
711
 
        self.assertLength(1, case._benchcalls)
712
 
        # We must be able to unpack it as the test reporting code wants
713
 
        (_, _, _), stats = case._benchcalls[0]
714
 
        self.assertTrue(callable(stats.pprint))
715
 
 
716
 
 
717
684
class TestTestResult(tests.TestCase):
718
685
 
719
686
    def check_timing(self, test_case, expected_re):
721
688
                descriptions=0,
722
689
                verbosity=1,
723
690
                )
724
 
        capture = testtools.tests.helpers.ExtendedTestResult()
725
 
        test_case.run(MultiTestResult(result, capture))
726
 
        run_case = capture._events[0][1]
727
 
        timed_string = result._testTimeString(run_case)
 
691
        test_case.run(result)
 
692
        timed_string = result._testTimeString(test_case)
728
693
        self.assertContainsRe(timed_string, expected_re)
729
694
 
730
695
    def test_test_reporting(self):
748
713
        self.check_timing(ShortDelayTestCase('test_short_delay'),
749
714
                          r"^ +[0-9]+ms$")
750
715
 
751
 
    def _patch_get_bzr_source_tree(self):
752
 
        # Reading from the actual source tree breaks isolation, but we don't
753
 
        # want to assume that thats *all* that would happen.
754
 
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', lambda: None)
755
 
 
756
716
    def test_assigned_benchmark_file_stores_date(self):
757
 
        self._patch_get_bzr_source_tree()
758
717
        output = StringIO()
759
718
        result = bzrlib.tests.TextTestResult(self._log_file,
760
719
                                        descriptions=0,
768
727
        self.assertContainsRe(output_string, "--date [0-9.]+")
769
728
 
770
729
    def test_benchhistory_records_test_times(self):
771
 
        self._patch_get_bzr_source_tree()
772
730
        result_stream = StringIO()
773
731
        result = bzrlib.tests.TextTestResult(
774
732
            self._log_file,
801
759
        self.requireFeature(test_lsprof.LSProfFeature)
802
760
        result_stream = StringIO()
803
761
        result = bzrlib.tests.VerboseTestResult(
804
 
            result_stream,
 
762
            unittest._WritelnDecorator(result_stream),
805
763
            descriptions=0,
806
764
            verbosity=2,
807
765
            )
836
794
    def test_known_failure(self):
837
795
        """A KnownFailure being raised should trigger several result actions."""
838
796
        class InstrumentedTestResult(tests.ExtendedTestResult):
839
 
            def stopTestRun(self): pass
 
797
            def done(self): pass
840
798
            def startTests(self): pass
841
799
            def report_test_start(self, test): pass
842
 
            def report_known_failure(self, test, err=None, details=None):
843
 
                self._call = test, 'known failure'
 
800
            def report_known_failure(self, test, err):
 
801
                self._call = test, err
844
802
        result = InstrumentedTestResult(None, None, None, None)
845
 
        class Test(tests.TestCase):
846
 
            def test_function(self):
847
 
                raise tests.KnownFailure('failed!')
848
 
        test = Test("test_function")
 
803
        def test_function():
 
804
            raise tests.KnownFailure('failed!')
 
805
        test = unittest.FunctionTestCase(test_function)
849
806
        test.run(result)
850
807
        # it should invoke 'report_known_failure'.
851
808
        self.assertEqual(2, len(result._call))
852
 
        self.assertEqual(test.id(), result._call[0].id())
853
 
        self.assertEqual('known failure', result._call[1])
 
809
        self.assertEqual(test, result._call[0])
 
810
        self.assertEqual(tests.KnownFailure, result._call[1][0])
 
811
        self.assertIsInstance(result._call[1][1], tests.KnownFailure)
854
812
        # we dont introspec the traceback, if the rest is ok, it would be
855
813
        # exceptional for it not to be.
856
814
        # it should update the known_failure_count on the object.
862
820
        # verbose test output formatting
863
821
        result_stream = StringIO()
864
822
        result = bzrlib.tests.VerboseTestResult(
865
 
            result_stream,
 
823
            unittest._WritelnDecorator(result_stream),
866
824
            descriptions=0,
867
825
            verbosity=2,
868
826
            )
878
836
        output = result_stream.getvalue()[prefix:]
879
837
        lines = output.splitlines()
880
838
        self.assertContainsRe(lines[0], r'XFAIL *\d+ms$')
881
 
        if sys.version_info > (2, 7):
882
 
            self.expectFailure("_ExpectedFailure on 2.7 loses the message",
883
 
                self.assertNotEqual, lines[1], '    ')
884
839
        self.assertEqual(lines[1], '    foo')
885
840
        self.assertEqual(2, len(lines))
886
841
 
893
848
    def test_add_not_supported(self):
894
849
        """Test the behaviour of invoking addNotSupported."""
895
850
        class InstrumentedTestResult(tests.ExtendedTestResult):
896
 
            def stopTestRun(self): pass
 
851
            def done(self): pass
897
852
            def startTests(self): pass
898
853
            def report_test_start(self, test): pass
899
854
            def report_unsupported(self, test, feature):
920
875
        # verbose test output formatting
921
876
        result_stream = StringIO()
922
877
        result = bzrlib.tests.VerboseTestResult(
923
 
            result_stream,
 
878
            unittest._WritelnDecorator(result_stream),
924
879
            descriptions=0,
925
880
            verbosity=2,
926
881
            )
931
886
        result.report_unsupported(test, feature)
932
887
        output = result_stream.getvalue()[prefix:]
933
888
        lines = output.splitlines()
934
 
        # We don't check for the final '0ms' since it may fail on slow hosts
935
 
        self.assertStartsWith(lines[0], 'NODEP')
936
 
        self.assertEqual(lines[1],
937
 
                         "    The feature 'Feature' is not available.")
 
889
        self.assertEqual(lines, ['NODEP        0ms',
 
890
                                 "    The feature 'Feature' is not available."])
938
891
 
939
892
    def test_unavailable_exception(self):
940
893
        """An UnavailableFeature being raised should invoke addNotSupported."""
941
894
        class InstrumentedTestResult(tests.ExtendedTestResult):
942
 
            def stopTestRun(self): pass
 
895
            def done(self): pass
943
896
            def startTests(self): pass
944
897
            def report_test_start(self, test): pass
945
898
            def addNotSupported(self, test, feature):
946
899
                self._call = test, feature
947
900
        result = InstrumentedTestResult(None, None, None, None)
948
901
        feature = tests.Feature()
949
 
        class Test(tests.TestCase):
950
 
            def test_function(self):
951
 
                raise tests.UnavailableFeature(feature)
952
 
        test = Test("test_function")
 
902
        def test_function():
 
903
            raise tests.UnavailableFeature(feature)
 
904
        test = unittest.FunctionTestCase(test_function)
953
905
        test.run(result)
954
906
        # it should invoke 'addNotSupported'.
955
907
        self.assertEqual(2, len(result._call))
956
 
        self.assertEqual(test.id(), result._call[0].id())
 
908
        self.assertEqual(test, result._call[0])
957
909
        self.assertEqual(feature, result._call[1])
958
910
        # and not count as an error
959
911
        self.assertEqual(0, result.error_count)
972
924
                                             verbosity=1)
973
925
        test = self.get_passing_test()
974
926
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
975
 
        result.addExpectedFailure(test, err)
 
927
        result._addKnownFailure(test, err)
976
928
        self.assertFalse(result.wasStrictlySuccessful())
977
929
        self.assertEqual(None, result._extractBenchmarkTime(test))
978
930
 
1023
975
        because of our use of global state.
1024
976
        """
1025
977
        old_root = tests.TestCaseInTempDir.TEST_ROOT
1026
 
        old_leak = tests.TestCase._first_thread_leaker_id
1027
978
        try:
1028
979
            tests.TestCaseInTempDir.TEST_ROOT = None
1029
 
            tests.TestCase._first_thread_leaker_id = None
1030
980
            return testrunner.run(test)
1031
981
        finally:
1032
982
            tests.TestCaseInTempDir.TEST_ROOT = old_root
1033
 
            tests.TestCase._first_thread_leaker_id = old_leak
1034
983
 
1035
984
    def test_known_failure_failed_run(self):
1036
985
        # run a test that generates a known failure which should be printed in
1037
986
        # the final output when real failures occur.
1038
 
        class Test(tests.TestCase):
1039
 
            def known_failure_test(self):
1040
 
                self.expectFailure('failed', self.assertTrue, False)
 
987
        def known_failure_test():
 
988
            raise tests.KnownFailure('failed')
1041
989
        test = unittest.TestSuite()
1042
 
        test.addTest(Test("known_failure_test"))
 
990
        test.addTest(unittest.FunctionTestCase(known_failure_test))
1043
991
        def failing_test():
1044
 
            self.fail('foo')
 
992
            raise AssertionError('foo')
1045
993
        test.addTest(unittest.FunctionTestCase(failing_test))
1046
994
        stream = StringIO()
1047
995
        runner = tests.TextTestRunner(stream=stream)
1048
996
        result = self.run_test_runner(runner, test)
1049
997
        lines = stream.getvalue().splitlines()
1050
 
        self.assertContainsRe(stream.getvalue(),
1051
 
            '(?sm)^bzr selftest.*$'
1052
 
            '.*'
1053
 
            '^======================================================================\n'
1054
 
            '^FAIL: failing_test\n'
1055
 
            '^----------------------------------------------------------------------\n'
1056
 
            'Traceback \\(most recent call last\\):\n'
1057
 
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
1058
 
            '    self.fail\\(\'foo\'\\)\n'
1059
 
            '.*'
1060
 
            '^----------------------------------------------------------------------\n'
1061
 
            '.*'
1062
 
            'FAILED \\(failures=1, known_failure_count=1\\)'
1063
 
            )
 
998
        self.assertEqual([
 
999
            '',
 
1000
            '======================================================================',
 
1001
            'FAIL: unittest.FunctionTestCase (failing_test)',
 
1002
            '----------------------------------------------------------------------',
 
1003
            'Traceback (most recent call last):',
 
1004
            '    raise AssertionError(\'foo\')',
 
1005
            'AssertionError: foo',
 
1006
            '',
 
1007
            '----------------------------------------------------------------------',
 
1008
            '',
 
1009
            'FAILED (failures=1, known_failure_count=1)'],
 
1010
            lines[3:8] + lines[9:13] + lines[14:])
1064
1011
 
1065
1012
    def test_known_failure_ok_run(self):
1066
 
        # run a test that generates a known failure which should be printed in
1067
 
        # the final output.
1068
 
        class Test(tests.TestCase):
1069
 
            def known_failure_test(self):
1070
 
                self.expectFailure('failed', self.assertTrue, False)
1071
 
        test = Test("known_failure_test")
 
1013
        # run a test that generates a known failure which should be printed in the final output.
 
1014
        def known_failure_test():
 
1015
            raise tests.KnownFailure('failed')
 
1016
        test = unittest.FunctionTestCase(known_failure_test)
1072
1017
        stream = StringIO()
1073
1018
        runner = tests.TextTestRunner(stream=stream)
1074
1019
        result = self.run_test_runner(runner, test)
1079
1024
            '\n'
1080
1025
            'OK \\(known_failures=1\\)\n')
1081
1026
 
1082
 
    def test_result_decorator(self):
1083
 
        # decorate results
1084
 
        calls = []
1085
 
        class LoggingDecorator(tests.ForwardingResult):
1086
 
            def startTest(self, test):
1087
 
                tests.ForwardingResult.startTest(self, test)
1088
 
                calls.append('start')
1089
 
        test = unittest.FunctionTestCase(lambda:None)
1090
 
        stream = StringIO()
1091
 
        runner = tests.TextTestRunner(stream=stream,
1092
 
            result_decorators=[LoggingDecorator])
1093
 
        result = self.run_test_runner(runner, test)
1094
 
        self.assertLength(1, calls)
1095
 
 
1096
1027
    def test_skipped_test(self):
1097
1028
        # run a test that is skipped, and check the suite as a whole still
1098
1029
        # succeeds.
1151
1082
 
1152
1083
    def test_not_applicable(self):
1153
1084
        # run a test that is skipped because it's not applicable
1154
 
        class Test(tests.TestCase):
1155
 
            def not_applicable_test(self):
1156
 
                raise tests.TestNotApplicable('this test never runs')
 
1085
        def not_applicable_test():
 
1086
            raise tests.TestNotApplicable('this test never runs')
1157
1087
        out = StringIO()
1158
1088
        runner = tests.TextTestRunner(stream=out, verbosity=2)
1159
 
        test = Test("not_applicable_test")
 
1089
        test = unittest.FunctionTestCase(not_applicable_test)
1160
1090
        result = self.run_test_runner(runner, test)
1161
1091
        self._log_file.write(out.getvalue())
1162
1092
        self.assertTrue(result.wasSuccessful())
1166
1096
        self.assertContainsRe(out.getvalue(),
1167
1097
                r'(?m)^    this test never runs')
1168
1098
 
 
1099
    def test_not_applicable_demo(self):
 
1100
        # just so you can see it in the test output
 
1101
        raise tests.TestNotApplicable('this test is just a demonstation')
 
1102
 
1169
1103
    def test_unsupported_features_listed(self):
1170
1104
        """When unsupported features are encountered they are detailed."""
1171
1105
        class Feature1(tests.Feature):
1191
1125
            ],
1192
1126
            lines[-3:])
1193
1127
 
1194
 
    def _patch_get_bzr_source_tree(self):
1195
 
        # Reading from the actual source tree breaks isolation, but we don't
1196
 
        # want to assume that thats *all* that would happen.
1197
 
        self._get_source_tree_calls = []
1198
 
        def new_get():
1199
 
            self._get_source_tree_calls.append("called")
1200
 
            return None
1201
 
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree',  new_get)
1202
 
 
1203
1128
    def test_bench_history(self):
1204
 
        # tests that the running the benchmark passes bench_history into
1205
 
        # the test result object. We can tell that happens if
1206
 
        # _get_bzr_source_tree is called.
1207
 
        self._patch_get_bzr_source_tree()
 
1129
        # tests that the running the benchmark produces a history file
 
1130
        # containing a timestamp and the revision id of the bzrlib source which
 
1131
        # was tested.
 
1132
        workingtree = _get_bzr_source_tree()
1208
1133
        test = TestRunner('dummy_test')
1209
1134
        output = StringIO()
1210
1135
        runner = tests.TextTestRunner(stream=self._log_file,
1212
1137
        result = self.run_test_runner(runner, test)
1213
1138
        output_string = output.getvalue()
1214
1139
        self.assertContainsRe(output_string, "--date [0-9.]+")
1215
 
        self.assertLength(1, self._get_source_tree_calls)
1216
 
 
1217
 
    def test_startTestRun(self):
1218
 
        """run should call result.startTestRun()"""
1219
 
        calls = []
1220
 
        class LoggingDecorator(tests.ForwardingResult):
1221
 
            def startTestRun(self):
1222
 
                tests.ForwardingResult.startTestRun(self)
1223
 
                calls.append('startTestRun')
1224
 
        test = unittest.FunctionTestCase(lambda:None)
1225
 
        stream = StringIO()
1226
 
        runner = tests.TextTestRunner(stream=stream,
1227
 
            result_decorators=[LoggingDecorator])
1228
 
        result = self.run_test_runner(runner, test)
1229
 
        self.assertLength(1, calls)
1230
 
 
1231
 
    def test_stopTestRun(self):
1232
 
        """run should call result.stopTestRun()"""
1233
 
        calls = []
1234
 
        class LoggingDecorator(tests.ForwardingResult):
1235
 
            def stopTestRun(self):
1236
 
                tests.ForwardingResult.stopTestRun(self)
1237
 
                calls.append('stopTestRun')
1238
 
        test = unittest.FunctionTestCase(lambda:None)
1239
 
        stream = StringIO()
1240
 
        runner = tests.TextTestRunner(stream=stream,
1241
 
            result_decorators=[LoggingDecorator])
1242
 
        result = self.run_test_runner(runner, test)
1243
 
        self.assertLength(1, calls)
 
1140
        if workingtree is not None:
 
1141
            revision_id = workingtree.get_parent_ids()[0]
 
1142
            self.assertEndsWith(output_string.rstrip(), revision_id)
 
1143
 
 
1144
    def assertLogDeleted(self, test):
 
1145
        log = test._get_log()
 
1146
        self.assertEqual("DELETED log file to reduce memory footprint", log)
 
1147
        self.assertEqual('', test._log_contents)
 
1148
        self.assertIs(None, test._log_file_name)
 
1149
 
 
1150
    def test_success_log_deleted(self):
 
1151
        """Successful tests have their log deleted"""
 
1152
 
 
1153
        class LogTester(tests.TestCase):
 
1154
 
 
1155
            def test_success(self):
 
1156
                self.log('this will be removed\n')
 
1157
 
 
1158
        sio = StringIO()
 
1159
        runner = tests.TextTestRunner(stream=sio)
 
1160
        test = LogTester('test_success')
 
1161
        result = self.run_test_runner(runner, test)
 
1162
 
 
1163
        self.assertLogDeleted(test)
 
1164
 
 
1165
    def test_skipped_log_deleted(self):
 
1166
        """Skipped tests have their log deleted"""
 
1167
 
 
1168
        class LogTester(tests.TestCase):
 
1169
 
 
1170
            def test_skipped(self):
 
1171
                self.log('this will be removed\n')
 
1172
                raise tests.TestSkipped()
 
1173
 
 
1174
        sio = StringIO()
 
1175
        runner = tests.TextTestRunner(stream=sio)
 
1176
        test = LogTester('test_skipped')
 
1177
        result = self.run_test_runner(runner, test)
 
1178
 
 
1179
        self.assertLogDeleted(test)
 
1180
 
 
1181
    def test_not_aplicable_log_deleted(self):
 
1182
        """Not applicable tests have their log deleted"""
 
1183
 
 
1184
        class LogTester(tests.TestCase):
 
1185
 
 
1186
            def test_not_applicable(self):
 
1187
                self.log('this will be removed\n')
 
1188
                raise tests.TestNotApplicable()
 
1189
 
 
1190
        sio = StringIO()
 
1191
        runner = tests.TextTestRunner(stream=sio)
 
1192
        test = LogTester('test_not_applicable')
 
1193
        result = self.run_test_runner(runner, test)
 
1194
 
 
1195
        self.assertLogDeleted(test)
 
1196
 
 
1197
    def test_known_failure_log_deleted(self):
 
1198
        """Know failure tests have their log deleted"""
 
1199
 
 
1200
        class LogTester(tests.TestCase):
 
1201
 
 
1202
            def test_known_failure(self):
 
1203
                self.log('this will be removed\n')
 
1204
                raise tests.KnownFailure()
 
1205
 
 
1206
        sio = StringIO()
 
1207
        runner = tests.TextTestRunner(stream=sio)
 
1208
        test = LogTester('test_known_failure')
 
1209
        result = self.run_test_runner(runner, test)
 
1210
 
 
1211
        self.assertLogDeleted(test)
 
1212
 
 
1213
    def test_fail_log_kept(self):
 
1214
        """Failed tests have their log kept"""
 
1215
 
 
1216
        class LogTester(tests.TestCase):
 
1217
 
 
1218
            def test_fail(self):
 
1219
                self.log('this will be kept\n')
 
1220
                self.fail('this test fails')
 
1221
 
 
1222
        sio = StringIO()
 
1223
        runner = tests.TextTestRunner(stream=sio)
 
1224
        test = LogTester('test_fail')
 
1225
        result = self.run_test_runner(runner, test)
 
1226
 
 
1227
        text = sio.getvalue()
 
1228
        self.assertContainsRe(text, 'this will be kept')
 
1229
        self.assertContainsRe(text, 'this test fails')
 
1230
 
 
1231
        log = test._get_log()
 
1232
        self.assertContainsRe(log, 'this will be kept')
 
1233
        self.assertEqual(log, test._log_contents)
 
1234
 
 
1235
    def test_error_log_kept(self):
 
1236
        """Tests with errors have their log kept"""
 
1237
 
 
1238
        class LogTester(tests.TestCase):
 
1239
 
 
1240
            def test_error(self):
 
1241
                self.log('this will be kept\n')
 
1242
                raise ValueError('random exception raised')
 
1243
 
 
1244
        sio = StringIO()
 
1245
        runner = tests.TextTestRunner(stream=sio)
 
1246
        test = LogTester('test_error')
 
1247
        result = self.run_test_runner(runner, test)
 
1248
 
 
1249
        text = sio.getvalue()
 
1250
        self.assertContainsRe(text, 'this will be kept')
 
1251
        self.assertContainsRe(text, 'random exception raised')
 
1252
 
 
1253
        log = test._get_log()
 
1254
        self.assertContainsRe(log, 'this will be kept')
 
1255
        self.assertEqual(log, test._log_contents)
1244
1256
 
1245
1257
 
1246
1258
class SampleTestCase(tests.TestCase):
1251
1263
class _TestException(Exception):
1252
1264
    pass
1253
1265
 
1254
 
 
1255
1266
class TestTestCase(tests.TestCase):
1256
1267
    """Tests that test the core bzrlib TestCase."""
1257
1268
 
1306
1317
        # we could set something and run a test that will check
1307
1318
        # it gets santised, but this is probably sufficient for now:
1308
1319
        # if someone runs the test with -Dsomething it will error.
1309
 
        flags = set()
1310
 
        if self._lock_check_thorough:
1311
 
            flags.add('strict_locks')
1312
 
        self.assertEqual(flags, bzrlib.debug.debug_flags)
 
1320
        self.assertEqual(set(), bzrlib.debug.debug_flags)
1313
1321
 
1314
1322
    def change_selftest_debug_flags(self, new_flags):
1315
 
        self.overrideAttr(tests, 'selftest_debug_flags', set(new_flags))
 
1323
        orig_selftest_flags = tests.selftest_debug_flags
 
1324
        self.addCleanup(self._restore_selftest_debug_flags, orig_selftest_flags)
 
1325
        tests.selftest_debug_flags = set(new_flags)
 
1326
 
 
1327
    def _restore_selftest_debug_flags(self, flags):
 
1328
        tests.selftest_debug_flags = flags
1316
1329
 
1317
1330
    def test_allow_debug_flag(self):
1318
1331
        """The -Eallow_debug flag prevents bzrlib.debug.debug_flags from being
1325
1338
                self.flags = set(bzrlib.debug.debug_flags)
1326
1339
        test = TestThatRecordsFlags('test_foo')
1327
1340
        test.run(self.make_test_result())
1328
 
        flags = set(['a-flag'])
1329
 
        if 'disable_lock_checks' not in tests.selftest_debug_flags:
1330
 
            flags.add('strict_locks')
1331
 
        self.assertEqual(flags, self.flags)
1332
 
 
1333
 
    def test_disable_lock_checks(self):
1334
 
        """The -Edisable_lock_checks flag disables thorough checks."""
1335
 
        class TestThatRecordsFlags(tests.TestCase):
1336
 
            def test_foo(nested_self):
1337
 
                self.flags = set(bzrlib.debug.debug_flags)
1338
 
                self.test_lock_check_thorough = nested_self._lock_check_thorough
1339
 
        self.change_selftest_debug_flags(set())
1340
 
        test = TestThatRecordsFlags('test_foo')
1341
 
        test.run(self.make_test_result())
1342
 
        # By default we do strict lock checking and thorough lock/unlock
1343
 
        # tracking.
1344
 
        self.assertTrue(self.test_lock_check_thorough)
1345
 
        self.assertEqual(set(['strict_locks']), self.flags)
1346
 
        # Now set the disable_lock_checks flag, and show that this changed.
1347
 
        self.change_selftest_debug_flags(set(['disable_lock_checks']))
1348
 
        test = TestThatRecordsFlags('test_foo')
1349
 
        test.run(self.make_test_result())
1350
 
        self.assertFalse(self.test_lock_check_thorough)
1351
 
        self.assertEqual(set(), self.flags)
1352
 
 
1353
 
    def test_this_fails_strict_lock_check(self):
1354
 
        class TestThatRecordsFlags(tests.TestCase):
1355
 
            def test_foo(nested_self):
1356
 
                self.flags1 = set(bzrlib.debug.debug_flags)
1357
 
                self.thisFailsStrictLockCheck()
1358
 
                self.flags2 = set(bzrlib.debug.debug_flags)
1359
 
        # Make sure lock checking is active
1360
 
        self.change_selftest_debug_flags(set())
1361
 
        test = TestThatRecordsFlags('test_foo')
1362
 
        test.run(self.make_test_result())
1363
 
        self.assertEqual(set(['strict_locks']), self.flags1)
1364
 
        self.assertEqual(set(), self.flags2)
 
1341
        self.assertEqual(set(['a-flag']), self.flags)
1365
1342
 
1366
1343
    def test_debug_flags_restored(self):
1367
1344
        """The bzrlib debug flags should be restored to their original state
1378
1355
        self.assertEqual(set(['original-state']), bzrlib.debug.debug_flags)
1379
1356
 
1380
1357
    def make_test_result(self):
1381
 
        """Get a test result that writes to the test log file."""
1382
1358
        return tests.TextTestResult(self._log_file, descriptions=0, verbosity=1)
1383
1359
 
1384
1360
    def inner_test(self):
1392
1368
        result = self.make_test_result()
1393
1369
        self.inner_test.run(result)
1394
1370
        note("outer finish")
1395
 
        self.addCleanup(osutils.delete_any, self._log_file_name)
1396
1371
 
1397
1372
    def test_trace_nesting(self):
1398
1373
        # this tests that each test case nests its trace facility correctly.
1422
1397
        sample_test = TestTestCase("method_that_times_a_bit_twice")
1423
1398
        output_stream = StringIO()
1424
1399
        result = bzrlib.tests.VerboseTestResult(
1425
 
            output_stream,
 
1400
            unittest._WritelnDecorator(output_stream),
1426
1401
            descriptions=0,
1427
1402
            verbosity=2)
1428
1403
        sample_test.run(result)
1458
1433
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1459
1434
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1460
1435
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
1461
 
        del self._benchcalls[:]
1462
1436
 
1463
1437
    def test_knownFailure(self):
1464
1438
        """Self.knownFailure() should raise a KnownFailure exception."""
1465
1439
        self.assertRaises(tests.KnownFailure, self.knownFailure, "A Failure")
1466
1440
 
1467
 
    def test_open_bzrdir_safe_roots(self):
1468
 
        # even a memory transport should fail to open when its url isn't 
1469
 
        # permitted.
1470
 
        # Manually set one up (TestCase doesn't and shouldn't provide magic
1471
 
        # machinery)
1472
 
        transport_server = memory.MemoryServer()
1473
 
        transport_server.start_server()
1474
 
        self.addCleanup(transport_server.stop_server)
1475
 
        t = transport.get_transport(transport_server.get_url())
1476
 
        bzrdir.BzrDir.create(t.base)
1477
 
        self.assertRaises(errors.BzrError,
1478
 
            bzrdir.BzrDir.open_from_transport, t)
1479
 
        # But if we declare this as safe, we can open the bzrdir.
1480
 
        self.permit_url(t.base)
1481
 
        self._bzr_selftest_roots.append(t.base)
1482
 
        bzrdir.BzrDir.open_from_transport(t)
1483
 
 
1484
1441
    def test_requireFeature_available(self):
1485
1442
        """self.requireFeature(available) is a no-op."""
1486
1443
        class Available(tests.Feature):
1530
1487
        """Test disabled tests behaviour with support aware results."""
1531
1488
        test = SampleTestCase('_test_pass')
1532
1489
        class DisabledFeature(object):
1533
 
            def __eq__(self, other):
1534
 
                return isinstance(other, DisabledFeature)
1535
1490
            def available(self):
1536
1491
                return False
1537
1492
        the_feature = DisabledFeature()
1548
1503
                self.calls.append(('addNotSupported', test, feature))
1549
1504
        result = InstrumentedTestResult()
1550
1505
        test.run(result)
1551
 
        case = result.calls[0][1]
1552
1506
        self.assertEqual([
1553
 
            ('startTest', case),
1554
 
            ('addNotSupported', case, the_feature),
1555
 
            ('stopTest', case),
 
1507
            ('startTest', test),
 
1508
            ('addNotSupported', test, the_feature),
 
1509
            ('stopTest', test),
1556
1510
            ],
1557
1511
            result.calls)
1558
1512
 
1559
 
    def test_start_server_registers_url(self):
1560
 
        transport_server = memory.MemoryServer()
1561
 
        # A little strict, but unlikely to be changed soon.
1562
 
        self.assertEqual([], self._bzr_selftest_roots)
1563
 
        self.start_server(transport_server)
1564
 
        self.assertSubset([transport_server.get_url()],
1565
 
            self._bzr_selftest_roots)
1566
 
 
1567
1513
    def test_assert_list_raises_on_generator(self):
1568
1514
        def generator_which_will_raise():
1569
1515
            # This will not raise until after the first yield
1619
1565
        self.assertRaises(AssertionError,
1620
1566
            self.assertListRaises, _TestException, success_generator)
1621
1567
 
1622
 
    def test_overrideAttr_without_value(self):
1623
 
        self.test_attr = 'original' # Define a test attribute
1624
 
        obj = self # Make 'obj' visible to the embedded test
1625
 
        class Test(tests.TestCase):
1626
 
 
1627
 
            def setUp(self):
1628
 
                tests.TestCase.setUp(self)
1629
 
                self.orig = self.overrideAttr(obj, 'test_attr')
1630
 
 
1631
 
            def test_value(self):
1632
 
                self.assertEqual('original', self.orig)
1633
 
                self.assertEqual('original', obj.test_attr)
1634
 
                obj.test_attr = 'modified'
1635
 
                self.assertEqual('modified', obj.test_attr)
1636
 
 
1637
 
        test = Test('test_value')
1638
 
        test.run(unittest.TestResult())
1639
 
        self.assertEqual('original', obj.test_attr)
1640
 
 
1641
 
    def test_overrideAttr_with_value(self):
1642
 
        self.test_attr = 'original' # Define a test attribute
1643
 
        obj = self # Make 'obj' visible to the embedded test
1644
 
        class Test(tests.TestCase):
1645
 
 
1646
 
            def setUp(self):
1647
 
                tests.TestCase.setUp(self)
1648
 
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
1649
 
 
1650
 
            def test_value(self):
1651
 
                self.assertEqual('original', self.orig)
1652
 
                self.assertEqual('modified', obj.test_attr)
1653
 
 
1654
 
        test = Test('test_value')
1655
 
        test.run(unittest.TestResult())
1656
 
        self.assertEqual('original', obj.test_attr)
1657
 
 
1658
1568
 
1659
1569
# NB: Don't delete this; it's not actually from 0.11!
1660
1570
@deprecated_function(deprecated_in((0, 11, 0)))
1703
1613
        self.assertEndsWith('foo', 'oo')
1704
1614
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1705
1615
 
1706
 
    def test_assertEqualDiff(self):
1707
 
        e = self.assertRaises(AssertionError,
1708
 
                              self.assertEqualDiff, '', '\n')
1709
 
        self.assertEquals(str(e),
1710
 
                          # Don't blink ! The '+' applies to the second string
1711
 
                          'first string is missing a final newline.\n+ \n')
1712
 
        e = self.assertRaises(AssertionError,
1713
 
                              self.assertEqualDiff, '\n', '')
1714
 
        self.assertEquals(str(e),
1715
 
                          # Don't blink ! The '-' applies to the second string
1716
 
                          'second string is missing a final newline.\n- \n')
1717
 
 
1718
 
 
1719
 
class TestDeprecations(tests.TestCase):
1720
 
 
1721
1616
    def test_applyDeprecated_not_deprecated(self):
1722
1617
        sample_object = ApplyDeprecatedHelper()
1723
1618
        # calling an undeprecated callable raises an assertion
1800
1695
        tree = self.make_branch_and_memory_tree('a')
1801
1696
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1802
1697
 
1803
 
    def test_make_tree_for_local_vfs_backed_transport(self):
1804
 
        # make_branch_and_tree has to use local branch and repositories
1805
 
        # when the vfs transport and local disk are colocated, even if
1806
 
        # a different transport is in use for url generation.
1807
 
        self.transport_server = test_server.FakeVFATServer
1808
 
        self.assertFalse(self.get_url('t1').startswith('file://'))
 
1698
 
 
1699
class TestSFTPMakeBranchAndTree(test_sftp_transport.TestCaseWithSFTPServer):
 
1700
 
 
1701
    def test_make_tree_for_sftp_branch(self):
 
1702
        """Transports backed by local directories create local trees."""
 
1703
 
1809
1704
        tree = self.make_branch_and_tree('t1')
1810
1705
        base = tree.bzrdir.root_transport.base
1811
 
        self.assertStartsWith(base, 'file://')
 
1706
        self.failIf(base.startswith('sftp'),
 
1707
                'base %r is on sftp but should be local' % base)
1812
1708
        self.assertEquals(tree.bzrdir.root_transport,
1813
1709
                tree.branch.bzrdir.root_transport)
1814
1710
        self.assertEquals(tree.bzrdir.root_transport,
1815
1711
                tree.branch.repository.bzrdir.root_transport)
1816
1712
 
1817
1713
 
1818
 
class SelfTestHelper:
1819
 
 
1820
 
    def run_selftest(self, **kwargs):
1821
 
        """Run selftest returning its output."""
1822
 
        output = StringIO()
1823
 
        old_transport = bzrlib.tests.default_transport
1824
 
        old_root = tests.TestCaseWithMemoryTransport.TEST_ROOT
1825
 
        tests.TestCaseWithMemoryTransport.TEST_ROOT = None
1826
 
        try:
1827
 
            self.assertEqual(True, tests.selftest(stream=output, **kwargs))
1828
 
        finally:
1829
 
            bzrlib.tests.default_transport = old_transport
1830
 
            tests.TestCaseWithMemoryTransport.TEST_ROOT = old_root
1831
 
        output.seek(0)
1832
 
        return output
1833
 
 
1834
 
 
1835
 
class TestSelftest(tests.TestCase, SelfTestHelper):
 
1714
class TestSelftest(tests.TestCase):
1836
1715
    """Tests of bzrlib.tests.selftest."""
1837
1716
 
1838
1717
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
1846
1725
            test_suite_factory=factory)
1847
1726
        self.assertEqual([True], factory_called)
1848
1727
 
1849
 
    def factory(self):
1850
 
        """A test suite factory."""
1851
 
        class Test(tests.TestCase):
1852
 
            def a(self):
1853
 
                pass
1854
 
            def b(self):
1855
 
                pass
1856
 
            def c(self):
1857
 
                pass
1858
 
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
1859
 
 
1860
 
    def test_list_only(self):
1861
 
        output = self.run_selftest(test_suite_factory=self.factory,
1862
 
            list_only=True)
1863
 
        self.assertEqual(3, len(output.readlines()))
1864
 
 
1865
 
    def test_list_only_filtered(self):
1866
 
        output = self.run_selftest(test_suite_factory=self.factory,
1867
 
            list_only=True, pattern="Test.b")
1868
 
        self.assertEndsWith(output.getvalue(), "Test.b\n")
1869
 
        self.assertLength(1, output.readlines())
1870
 
 
1871
 
    def test_list_only_excludes(self):
1872
 
        output = self.run_selftest(test_suite_factory=self.factory,
1873
 
            list_only=True, exclude_pattern="Test.b")
1874
 
        self.assertNotContainsRe("Test.b", output.getvalue())
1875
 
        self.assertLength(2, output.readlines())
1876
 
 
1877
 
    def test_lsprof_tests(self):
1878
 
        self.requireFeature(test_lsprof.LSProfFeature)
1879
 
        calls = []
1880
 
        class Test(object):
1881
 
            def __call__(test, result):
1882
 
                test.run(result)
1883
 
            def run(test, result):
1884
 
                self.assertIsInstance(result, tests.ForwardingResult)
1885
 
                calls.append("called")
1886
 
            def countTestCases(self):
1887
 
                return 1
1888
 
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
1889
 
        self.assertLength(1, calls)
1890
 
 
1891
 
    def test_random(self):
1892
 
        # test randomising by listing a number of tests.
1893
 
        output_123 = self.run_selftest(test_suite_factory=self.factory,
1894
 
            list_only=True, random_seed="123")
1895
 
        output_234 = self.run_selftest(test_suite_factory=self.factory,
1896
 
            list_only=True, random_seed="234")
1897
 
        self.assertNotEqual(output_123, output_234)
1898
 
        # "Randominzing test order..\n\n
1899
 
        self.assertLength(5, output_123.readlines())
1900
 
        self.assertLength(5, output_234.readlines())
1901
 
 
1902
 
    def test_random_reuse_is_same_order(self):
1903
 
        # test randomising by listing a number of tests.
1904
 
        expected = self.run_selftest(test_suite_factory=self.factory,
1905
 
            list_only=True, random_seed="123")
1906
 
        repeated = self.run_selftest(test_suite_factory=self.factory,
1907
 
            list_only=True, random_seed="123")
1908
 
        self.assertEqual(expected.getvalue(), repeated.getvalue())
1909
 
 
1910
 
    def test_runner_class(self):
1911
 
        self.requireFeature(features.subunit)
1912
 
        from subunit import ProtocolTestCase
1913
 
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
1914
 
            test_suite_factory=self.factory)
1915
 
        test = ProtocolTestCase(stream)
1916
 
        result = unittest.TestResult()
1917
 
        test.run(result)
1918
 
        self.assertEqual(3, result.testsRun)
1919
 
 
1920
 
    def test_starting_with_single_argument(self):
1921
 
        output = self.run_selftest(test_suite_factory=self.factory,
1922
 
            starting_with=['bzrlib.tests.test_selftest.Test.a'],
1923
 
            list_only=True)
1924
 
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n',
1925
 
            output.getvalue())
1926
 
 
1927
 
    def test_starting_with_multiple_argument(self):
1928
 
        output = self.run_selftest(test_suite_factory=self.factory,
1929
 
            starting_with=['bzrlib.tests.test_selftest.Test.a',
1930
 
                'bzrlib.tests.test_selftest.Test.b'],
1931
 
            list_only=True)
1932
 
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n'
1933
 
            'bzrlib.tests.test_selftest.Test.b\n',
1934
 
            output.getvalue())
1935
 
 
1936
 
    def check_transport_set(self, transport_server):
1937
 
        captured_transport = []
1938
 
        def seen_transport(a_transport):
1939
 
            captured_transport.append(a_transport)
1940
 
        class Capture(tests.TestCase):
1941
 
            def a(self):
1942
 
                seen_transport(bzrlib.tests.default_transport)
1943
 
        def factory():
1944
 
            return TestUtil.TestSuite([Capture("a")])
1945
 
        self.run_selftest(transport=transport_server, test_suite_factory=factory)
1946
 
        self.assertEqual(transport_server, captured_transport[0])
1947
 
 
1948
 
    def test_transport_sftp(self):
1949
 
        self.requireFeature(features.paramiko)
1950
 
        from bzrlib.tests import stub_sftp
1951
 
        self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
1952
 
 
1953
 
    def test_transport_memory(self):
1954
 
        self.check_transport_set(memory.MemoryServer)
1955
 
 
1956
 
 
1957
 
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
1958
 
    # Does IO: reads test.list
1959
 
 
1960
 
    def test_load_list(self):
1961
 
        # Provide a list with one test - this test.
1962
 
        test_id_line = '%s\n' % self.id()
1963
 
        self.build_tree_contents([('test.list', test_id_line)])
1964
 
        # And generate a list of the tests in  the suite.
1965
 
        stream = self.run_selftest(load_list='test.list', list_only=True)
1966
 
        self.assertEqual(test_id_line, stream.getvalue())
1967
 
 
1968
 
    def test_load_unknown(self):
1969
 
        # Provide a list with one test - this test.
1970
 
        # And generate a list of the tests in  the suite.
1971
 
        err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
1972
 
            load_list='missing file name', list_only=True)
1973
 
 
1974
 
 
1975
 
class TestRunBzr(tests.TestCase):
1976
 
 
1977
 
    out = ''
1978
 
    err = ''
1979
 
 
1980
 
    def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
1981
 
                         working_dir=None):
1982
 
        """Override _run_bzr_core to test how it is invoked by run_bzr.
1983
 
 
1984
 
        Attempts to run bzr from inside this class don't actually run it.
1985
 
 
1986
 
        We test how run_bzr actually invokes bzr in another location.  Here we
1987
 
        only need to test that it passes the right parameters to run_bzr.
1988
 
        """
1989
 
        self.argv = list(argv)
1990
 
        self.retcode = retcode
1991
 
        self.encoding = encoding
1992
 
        self.stdin = stdin
1993
 
        self.working_dir = working_dir
1994
 
        return self.retcode, self.out, self.err
1995
 
 
1996
 
    def test_run_bzr_error(self):
1997
 
        self.out = "It sure does!\n"
1998
 
        out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
1999
 
        self.assertEqual(['rocks'], self.argv)
2000
 
        self.assertEqual(34, self.retcode)
2001
 
        self.assertEqual('It sure does!\n', out)
2002
 
        self.assertEquals(out, self.out)
2003
 
        self.assertEqual('', err)
2004
 
        self.assertEquals(err, self.err)
2005
 
 
2006
 
    def test_run_bzr_error_regexes(self):
2007
 
        self.out = ''
2008
 
        self.err = "bzr: ERROR: foobarbaz is not versioned"
2009
 
        out, err = self.run_bzr_error(
2010
 
            ["bzr: ERROR: foobarbaz is not versioned"],
2011
 
            ['file-id', 'foobarbaz'])
2012
 
 
2013
 
    def test_encoding(self):
2014
 
        """Test that run_bzr passes encoding to _run_bzr_core"""
2015
 
        self.run_bzr('foo bar')
2016
 
        self.assertEqual(None, self.encoding)
2017
 
        self.assertEqual(['foo', 'bar'], self.argv)
2018
 
 
2019
 
        self.run_bzr('foo bar', encoding='baz')
2020
 
        self.assertEqual('baz', self.encoding)
2021
 
        self.assertEqual(['foo', 'bar'], self.argv)
2022
 
 
2023
 
    def test_retcode(self):
2024
 
        """Test that run_bzr passes retcode to _run_bzr_core"""
2025
 
        # Default is retcode == 0
2026
 
        self.run_bzr('foo bar')
2027
 
        self.assertEqual(0, self.retcode)
2028
 
        self.assertEqual(['foo', 'bar'], self.argv)
2029
 
 
2030
 
        self.run_bzr('foo bar', retcode=1)
2031
 
        self.assertEqual(1, self.retcode)
2032
 
        self.assertEqual(['foo', 'bar'], self.argv)
2033
 
 
2034
 
        self.run_bzr('foo bar', retcode=None)
2035
 
        self.assertEqual(None, self.retcode)
2036
 
        self.assertEqual(['foo', 'bar'], self.argv)
2037
 
 
2038
 
        self.run_bzr(['foo', 'bar'], retcode=3)
2039
 
        self.assertEqual(3, self.retcode)
2040
 
        self.assertEqual(['foo', 'bar'], self.argv)
2041
 
 
2042
 
    def test_stdin(self):
2043
 
        # test that the stdin keyword to run_bzr is passed through to
2044
 
        # _run_bzr_core as-is. We do this by overriding
2045
 
        # _run_bzr_core in this class, and then calling run_bzr,
2046
 
        # which is a convenience function for _run_bzr_core, so
2047
 
        # should invoke it.
2048
 
        self.run_bzr('foo bar', stdin='gam')
2049
 
        self.assertEqual('gam', self.stdin)
2050
 
        self.assertEqual(['foo', 'bar'], self.argv)
2051
 
 
2052
 
        self.run_bzr('foo bar', stdin='zippy')
2053
 
        self.assertEqual('zippy', self.stdin)
2054
 
        self.assertEqual(['foo', 'bar'], self.argv)
2055
 
 
2056
 
    def test_working_dir(self):
2057
 
        """Test that run_bzr passes working_dir to _run_bzr_core"""
2058
 
        self.run_bzr('foo bar')
2059
 
        self.assertEqual(None, self.working_dir)
2060
 
        self.assertEqual(['foo', 'bar'], self.argv)
2061
 
 
2062
 
        self.run_bzr('foo bar', working_dir='baz')
2063
 
        self.assertEqual('baz', self.working_dir)
2064
 
        self.assertEqual(['foo', 'bar'], self.argv)
2065
 
 
2066
 
    def test_reject_extra_keyword_arguments(self):
2067
 
        self.assertRaises(TypeError, self.run_bzr, "foo bar",
2068
 
                          error_regex=['error message'])
2069
 
 
2070
 
 
2071
 
class TestRunBzrCaptured(tests.TestCaseWithTransport):
2072
 
    # Does IO when testing the working_dir parameter.
2073
 
 
2074
 
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
2075
 
                         a_callable=None, *args, **kwargs):
2076
 
        self.stdin = stdin
2077
 
        self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
2078
 
        self.factory = bzrlib.ui.ui_factory
2079
 
        self.working_dir = osutils.getcwd()
2080
 
        stdout.write('foo\n')
2081
 
        stderr.write('bar\n')
2082
 
        return 0
2083
 
 
2084
 
    def test_stdin(self):
2085
 
        # test that the stdin keyword to _run_bzr_core is passed through to
2086
 
        # apply_redirected as a StringIO. We do this by overriding
2087
 
        # apply_redirected in this class, and then calling _run_bzr_core,
2088
 
        # which calls apply_redirected.
2089
 
        self.run_bzr(['foo', 'bar'], stdin='gam')
2090
 
        self.assertEqual('gam', self.stdin.read())
2091
 
        self.assertTrue(self.stdin is self.factory_stdin)
2092
 
        self.run_bzr(['foo', 'bar'], stdin='zippy')
2093
 
        self.assertEqual('zippy', self.stdin.read())
2094
 
        self.assertTrue(self.stdin is self.factory_stdin)
2095
 
 
2096
 
    def test_ui_factory(self):
2097
 
        # each invocation of self.run_bzr should get its
2098
 
        # own UI factory, which is an instance of TestUIFactory,
2099
 
        # with stdin, stdout and stderr attached to the stdin,
2100
 
        # stdout and stderr of the invoked run_bzr
2101
 
        current_factory = bzrlib.ui.ui_factory
2102
 
        self.run_bzr(['foo'])
2103
 
        self.failIf(current_factory is self.factory)
2104
 
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2105
 
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2106
 
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2107
 
        self.assertEqual('bar\n', self.factory.stderr.getvalue())
2108
 
        self.assertIsInstance(self.factory, tests.TestUIFactory)
2109
 
 
2110
 
    def test_working_dir(self):
2111
 
        self.build_tree(['one/', 'two/'])
2112
 
        cwd = osutils.getcwd()
2113
 
 
2114
 
        # Default is to work in the current directory
2115
 
        self.run_bzr(['foo', 'bar'])
2116
 
        self.assertEqual(cwd, self.working_dir)
2117
 
 
2118
 
        self.run_bzr(['foo', 'bar'], working_dir=None)
2119
 
        self.assertEqual(cwd, self.working_dir)
2120
 
 
2121
 
        # The function should be run in the alternative directory
2122
 
        # but afterwards the current working dir shouldn't be changed
2123
 
        self.run_bzr(['foo', 'bar'], working_dir='one')
2124
 
        self.assertNotEqual(cwd, self.working_dir)
2125
 
        self.assertEndsWith(self.working_dir, 'one')
2126
 
        self.assertEqual(cwd, osutils.getcwd())
2127
 
 
2128
 
        self.run_bzr(['foo', 'bar'], working_dir='two')
2129
 
        self.assertNotEqual(cwd, self.working_dir)
2130
 
        self.assertEndsWith(self.working_dir, 'two')
2131
 
        self.assertEqual(cwd, osutils.getcwd())
2132
 
 
2133
 
 
2134
 
class StubProcess(object):
2135
 
    """A stub process for testing run_bzr_subprocess."""
2136
 
    
2137
 
    def __init__(self, out="", err="", retcode=0):
2138
 
        self.out = out
2139
 
        self.err = err
2140
 
        self.returncode = retcode
2141
 
 
2142
 
    def communicate(self):
2143
 
        return self.out, self.err
2144
 
 
2145
 
 
2146
 
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
2147
 
    """Base class for tests testing how we might run bzr."""
2148
 
 
2149
 
    def setUp(self):
2150
 
        tests.TestCaseWithTransport.setUp(self)
2151
 
        self.subprocess_calls = []
2152
 
 
2153
 
    def start_bzr_subprocess(self, process_args, env_changes=None,
2154
 
                             skip_if_plan_to_signal=False,
2155
 
                             working_dir=None,
2156
 
                             allow_plugins=False):
2157
 
        """capture what run_bzr_subprocess tries to do."""
2158
 
        self.subprocess_calls.append({'process_args':process_args,
2159
 
            'env_changes':env_changes,
2160
 
            'skip_if_plan_to_signal':skip_if_plan_to_signal,
2161
 
            'working_dir':working_dir, 'allow_plugins':allow_plugins})
2162
 
        return self.next_subprocess
2163
 
 
2164
 
 
2165
 
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
2166
 
 
2167
 
    def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
2168
 
        """Run run_bzr_subprocess with args and kwargs using a stubbed process.
2169
 
 
2170
 
        Inside TestRunBzrSubprocessCommands we use a stub start_bzr_subprocess
2171
 
        that will return static results. This assertion method populates those
2172
 
        results and also checks the arguments run_bzr_subprocess generates.
2173
 
        """
2174
 
        self.next_subprocess = process
2175
 
        try:
2176
 
            result = self.run_bzr_subprocess(*args, **kwargs)
2177
 
        except:
2178
 
            self.next_subprocess = None
2179
 
            for key, expected in expected_args.iteritems():
2180
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2181
 
            raise
 
1728
 
 
1729
class TestKnownFailure(tests.TestCase):
 
1730
 
 
1731
    def test_known_failure(self):
 
1732
        """Check that KnownFailure is defined appropriately."""
 
1733
        # a KnownFailure is an assertion error for compatability with unaware
 
1734
        # runners.
 
1735
        self.assertIsInstance(tests.KnownFailure(""), AssertionError)
 
1736
 
 
1737
    def test_expect_failure(self):
 
1738
        try:
 
1739
            self.expectFailure("Doomed to failure", self.assertTrue, False)
 
1740
        except tests.KnownFailure, e:
 
1741
            self.assertEqual('Doomed to failure', e.args[0])
 
1742
        try:
 
1743
            self.expectFailure("Doomed to failure", self.assertTrue, True)
 
1744
        except AssertionError, e:
 
1745
            self.assertEqual('Unexpected success.  Should have failed:'
 
1746
                             ' Doomed to failure', e.args[0])
2182
1747
        else:
2183
 
            self.next_subprocess = None
2184
 
            for key, expected in expected_args.iteritems():
2185
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2186
 
            return result
2187
 
 
2188
 
    def test_run_bzr_subprocess(self):
2189
 
        """The run_bzr_helper_external command behaves nicely."""
2190
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2191
 
            StubProcess(), '--version')
2192
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2193
 
            StubProcess(), ['--version'])
2194
 
        # retcode=None disables retcode checking
2195
 
        result = self.assertRunBzrSubprocess({},
2196
 
            StubProcess(retcode=3), '--version', retcode=None)
2197
 
        result = self.assertRunBzrSubprocess({},
2198
 
            StubProcess(out="is free software"), '--version')
2199
 
        self.assertContainsRe(result[0], 'is free software')
2200
 
        # Running a subcommand that is missing errors
2201
 
        self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
2202
 
            {'process_args':['--versionn']}, StubProcess(retcode=3),
2203
 
            '--versionn')
2204
 
        # Unless it is told to expect the error from the subprocess
2205
 
        result = self.assertRunBzrSubprocess({},
2206
 
            StubProcess(retcode=3), '--versionn', retcode=3)
2207
 
        # Or to ignore retcode checking
2208
 
        result = self.assertRunBzrSubprocess({},
2209
 
            StubProcess(err="unknown command", retcode=3), '--versionn',
2210
 
            retcode=None)
2211
 
        self.assertContainsRe(result[1], 'unknown command')
2212
 
 
2213
 
    def test_env_change_passes_through(self):
2214
 
        self.assertRunBzrSubprocess(
2215
 
            {'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
2216
 
            StubProcess(), '',
2217
 
            env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
2218
 
 
2219
 
    def test_no_working_dir_passed_as_None(self):
2220
 
        self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
2221
 
 
2222
 
    def test_no_working_dir_passed_through(self):
2223
 
        self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
2224
 
            working_dir='dir')
2225
 
 
2226
 
    def test_run_bzr_subprocess_no_plugins(self):
2227
 
        self.assertRunBzrSubprocess({'allow_plugins': False},
2228
 
            StubProcess(), '')
2229
 
 
2230
 
    def test_allow_plugins(self):
2231
 
        self.assertRunBzrSubprocess({'allow_plugins': True},
2232
 
            StubProcess(), '', allow_plugins=True)
2233
 
 
2234
 
 
2235
 
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2236
 
 
2237
 
    def test_finish_bzr_subprocess_with_error(self):
2238
 
        """finish_bzr_subprocess allows specification of the desired exit code.
2239
 
        """
2240
 
        process = StubProcess(err="unknown command", retcode=3)
2241
 
        result = self.finish_bzr_subprocess(process, retcode=3)
2242
 
        self.assertEqual('', result[0])
2243
 
        self.assertContainsRe(result[1], 'unknown command')
2244
 
 
2245
 
    def test_finish_bzr_subprocess_ignoring_retcode(self):
2246
 
        """finish_bzr_subprocess allows the exit code to be ignored."""
2247
 
        process = StubProcess(err="unknown command", retcode=3)
2248
 
        result = self.finish_bzr_subprocess(process, retcode=None)
2249
 
        self.assertEqual('', result[0])
2250
 
        self.assertContainsRe(result[1], 'unknown command')
2251
 
 
2252
 
    def test_finish_subprocess_with_unexpected_retcode(self):
2253
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2254
 
        not the expected one.
2255
 
        """
2256
 
        process = StubProcess(err="unknown command", retcode=3)
2257
 
        self.assertRaises(self.failureException, self.finish_bzr_subprocess,
2258
 
                          process)
2259
 
 
2260
 
 
2261
 
class _DontSpawnProcess(Exception):
2262
 
    """A simple exception which just allows us to skip unnecessary steps"""
2263
 
 
2264
 
 
2265
 
class TestStartBzrSubProcess(tests.TestCase):
2266
 
 
2267
 
    def check_popen_state(self):
2268
 
        """Replace to make assertions when popen is called."""
2269
 
 
2270
 
    def _popen(self, *args, **kwargs):
2271
 
        """Record the command that is run, so that we can ensure it is correct"""
2272
 
        self.check_popen_state()
2273
 
        self._popen_args = args
2274
 
        self._popen_kwargs = kwargs
2275
 
        raise _DontSpawnProcess()
2276
 
 
2277
 
    def test_run_bzr_subprocess_no_plugins(self):
2278
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2279
 
        command = self._popen_args[0]
2280
 
        self.assertEqual(sys.executable, command[0])
2281
 
        self.assertEqual(self.get_bzr_path(), command[1])
2282
 
        self.assertEqual(['--no-plugins'], command[2:])
2283
 
 
2284
 
    def test_allow_plugins(self):
2285
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2286
 
            allow_plugins=True)
2287
 
        command = self._popen_args[0]
2288
 
        self.assertEqual([], command[2:])
2289
 
 
2290
 
    def test_set_env(self):
2291
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2292
 
        # set in the child
2293
 
        def check_environment():
2294
 
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2295
 
        self.check_popen_state = check_environment
2296
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2297
 
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
2298
 
        # not set in theparent
2299
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2300
 
 
2301
 
    def test_run_bzr_subprocess_env_del(self):
2302
 
        """run_bzr_subprocess can remove environment variables too."""
2303
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2304
 
        def check_environment():
2305
 
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2306
 
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2307
 
        self.check_popen_state = check_environment
2308
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2309
 
            env_changes={'EXISTANT_ENV_VAR':None})
2310
 
        # Still set in parent
2311
 
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2312
 
        del os.environ['EXISTANT_ENV_VAR']
2313
 
 
2314
 
    def test_env_del_missing(self):
2315
 
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2316
 
        def check_environment():
2317
 
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2318
 
        self.check_popen_state = check_environment
2319
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2320
 
            env_changes={'NON_EXISTANT_ENV_VAR':None})
2321
 
 
2322
 
    def test_working_dir(self):
2323
 
        """Test that we can specify the working dir for the child"""
2324
 
        orig_getcwd = osutils.getcwd
2325
 
        orig_chdir = os.chdir
2326
 
        chdirs = []
2327
 
        def chdir(path):
2328
 
            chdirs.append(path)
2329
 
        os.chdir = chdir
2330
 
        try:
2331
 
            def getcwd():
2332
 
                return 'current'
2333
 
            osutils.getcwd = getcwd
2334
 
            try:
2335
 
                self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2336
 
                    working_dir='foo')
2337
 
            finally:
2338
 
                osutils.getcwd = orig_getcwd
2339
 
        finally:
2340
 
            os.chdir = orig_chdir
2341
 
        self.assertEqual(['foo', 'current'], chdirs)
2342
 
 
2343
 
    def test_get_bzr_path_with_cwd_bzrlib(self):
2344
 
        self.get_source_path = lambda: ""
2345
 
        self.overrideAttr(os.path, "isfile", lambda path: True)
2346
 
        self.assertEqual(self.get_bzr_path(), "bzr")
2347
 
 
2348
 
 
2349
 
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2350
 
    """Tests that really need to do things with an external bzr."""
2351
 
 
2352
 
    def test_start_and_stop_bzr_subprocess_send_signal(self):
2353
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2354
 
        not the expected one.
2355
 
        """
2356
 
        self.disable_missing_extensions_warning()
2357
 
        process = self.start_bzr_subprocess(['wait-until-signalled'],
2358
 
                                            skip_if_plan_to_signal=True)
2359
 
        self.assertEqual('running\n', process.stdout.readline())
2360
 
        result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
2361
 
                                            retcode=3)
2362
 
        self.assertEqual('', result[0])
2363
 
        self.assertEqual('bzr: interrupted\n', result[1])
 
1748
            self.fail('Assertion not raised')
2364
1749
 
2365
1750
 
2366
1751
class TestFeature(tests.TestCase):
2404
1789
        self.assertIs(feature, exception.args[0])
2405
1790
 
2406
1791
 
2407
 
simple_thunk_feature = tests._CompatabilityThunkFeature(
2408
 
    deprecated_in((2, 1, 0)),
2409
 
    'bzrlib.tests.test_selftest',
2410
 
    'simple_thunk_feature','UnicodeFilename',
2411
 
    replacement_module='bzrlib.tests'
2412
 
    )
2413
 
 
2414
 
class Test_CompatibilityFeature(tests.TestCase):
2415
 
 
2416
 
    def test_does_thunk(self):
2417
 
        res = self.callDeprecated(
2418
 
            ['bzrlib.tests.test_selftest.simple_thunk_feature was deprecated'
2419
 
             ' in version 2.1.0. Use bzrlib.tests.UnicodeFilename instead.'],
2420
 
            simple_thunk_feature.available)
2421
 
        self.assertEqual(tests.UnicodeFilename.available(), res)
2422
 
 
2423
 
 
2424
 
class TestModuleAvailableFeature(tests.TestCase):
2425
 
 
2426
 
    def test_available_module(self):
2427
 
        feature = tests.ModuleAvailableFeature('bzrlib.tests')
2428
 
        self.assertEqual('bzrlib.tests', feature.module_name)
2429
 
        self.assertEqual('bzrlib.tests', str(feature))
2430
 
        self.assertTrue(feature.available())
2431
 
        self.assertIs(tests, feature.module)
2432
 
 
2433
 
    def test_unavailable_module(self):
2434
 
        feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
2435
 
        self.assertEqual('bzrlib.no_such_module_exists', str(feature))
2436
 
        self.assertFalse(feature.available())
2437
 
        self.assertIs(None, feature.module)
2438
 
 
2439
 
 
2440
1792
class TestSelftestFiltering(tests.TestCase):
2441
1793
 
2442
1794
    def setUp(self):
2443
1795
        tests.TestCase.setUp(self)
2444
1796
        self.suite = TestUtil.TestSuite()
2445
1797
        self.loader = TestUtil.TestLoader()
2446
 
        self.suite.addTest(self.loader.loadTestsFromModule(
2447
 
            sys.modules['bzrlib.tests.test_selftest']))
 
1798
        self.suite.addTest(self.loader.loadTestsFromModuleNames([
 
1799
            'bzrlib.tests.test_selftest']))
2448
1800
        self.all_names = _test_ids(self.suite)
2449
1801
 
2450
1802
    def test_condition_id_re(self):
2621
1973
        # the test framework
2622
1974
        self.assertEquals('always fails', str(e))
2623
1975
        # check that there's no traceback in the test log
2624
 
        self.assertNotContainsRe(self.get_log(), r'Traceback')
 
1976
        self.assertNotContainsRe(self._get_log(keep_log_file=True),
 
1977
            r'Traceback')
2625
1978
 
2626
1979
    def test_run_bzr_user_error_caught(self):
2627
1980
        # Running bzr in blackbox mode, normal/expected/user errors should be
2628
1981
        # caught in the regular way and turned into an error message plus exit
2629
1982
        # code.
2630
 
        transport_server = memory.MemoryServer()
2631
 
        transport_server.start_server()
2632
 
        self.addCleanup(transport_server.stop_server)
2633
 
        url = transport_server.get_url()
2634
 
        self.permit_url(url)
2635
 
        out, err = self.run_bzr(["log", "%s/nonexistantpath" % url], retcode=3)
 
1983
        out, err = self.run_bzr(["log", "/nonexistantpath"], retcode=3)
2636
1984
        self.assertEqual(out, '')
2637
1985
        self.assertContainsRe(err,
2638
1986
            'bzr: ERROR: Not a branch: ".*nonexistantpath/".\n')
2764
2112
 
2765
2113
class TestTestSuite(tests.TestCase):
2766
2114
 
2767
 
    def test__test_suite_testmod_names(self):
2768
 
        # Test that a plausible list of test module names are returned
2769
 
        # by _test_suite_testmod_names.
2770
 
        test_list = tests._test_suite_testmod_names()
2771
 
        self.assertSubset([
2772
 
            'bzrlib.tests.blackbox',
2773
 
            'bzrlib.tests.per_transport',
2774
 
            'bzrlib.tests.test_selftest',
2775
 
            ],
2776
 
            test_list)
2777
 
 
2778
 
    def test__test_suite_modules_to_doctest(self):
2779
 
        # Test that a plausible list of modules to doctest is returned
2780
 
        # by _test_suite_modules_to_doctest.
2781
 
        test_list = tests._test_suite_modules_to_doctest()
2782
 
        if __doc__ is None:
2783
 
            # When docstrings are stripped, there are no modules to doctest
2784
 
            self.assertEqual([], test_list)
2785
 
            return
2786
 
        self.assertSubset([
2787
 
            'bzrlib.timestamp',
2788
 
            ],
2789
 
            test_list)
2790
 
 
2791
2115
    def test_test_suite(self):
2792
 
        # test_suite() loads the entire test suite to operate. To avoid this
2793
 
        # overhead, and yet still be confident that things are happening,
2794
 
        # we temporarily replace two functions used by test_suite with 
2795
 
        # test doubles that supply a few sample tests to load, and check they
2796
 
        # are loaded.
2797
 
        calls = []
2798
 
        def testmod_names():
2799
 
            calls.append("testmod_names")
2800
 
            return [
2801
 
                'bzrlib.tests.blackbox.test_branch',
2802
 
                'bzrlib.tests.per_transport',
2803
 
                'bzrlib.tests.test_selftest',
2804
 
                ]
2805
 
        self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
2806
 
        def doctests():
2807
 
            calls.append("modules_to_doctest")
2808
 
            if __doc__ is None:
2809
 
                return []
2810
 
            return ['bzrlib.timestamp']
2811
 
        self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
2812
 
        expected_test_list = [
 
2116
        # This test is slow, so we do a single test with one test in each
 
2117
        # category
 
2118
        test_list = [
2813
2119
            # testmod_names
2814
2120
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
2815
2121
            ('bzrlib.tests.per_transport.TransportTests'
2816
 
             '.test_abspath(LocalTransport,LocalURLServer)'),
 
2122
             '.test_abspath(LocalURLServer)'),
2817
2123
            'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
 
2124
            # modules_to_doctest
 
2125
            'bzrlib.timestamp.format_highres_date',
2818
2126
            # plugins can't be tested that way since selftest may be run with
2819
2127
            # --no-plugins
2820
2128
            ]
2821
 
        if __doc__ is not None:
2822
 
            expected_test_list.extend([
2823
 
                # modules_to_doctest
2824
 
                'bzrlib.timestamp.format_highres_date',
2825
 
                ])
2826
 
        suite = tests.test_suite()
2827
 
        self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
2828
 
            set(calls))
2829
 
        self.assertSubset(expected_test_list, _test_ids(suite))
 
2129
        suite = tests.test_suite(test_list)
 
2130
        self.assertEquals(test_list, _test_ids(suite))
2830
2131
 
2831
2132
    def test_test_suite_list_and_start(self):
2832
 
        # We cannot test this at the same time as the main load, because we want
2833
 
        # to know that starting_with == None works. So a second load is
2834
 
        # incurred - note that the starting_with parameter causes a partial load
2835
 
        # rather than a full load so this test should be pretty quick.
2836
2133
        test_list = ['bzrlib.tests.test_selftest.TestTestSuite.test_test_suite']
2837
2134
        suite = tests.test_suite(test_list,
2838
2135
                                 ['bzrlib.tests.test_selftest.TestTestSuite'])
2883
2180
    def test_load_tests(self):
2884
2181
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
2885
2182
        loader = self._create_loader(test_list)
 
2183
 
2886
2184
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2887
2185
        self.assertEquals(test_list, _test_ids(suite))
2888
2186
 
2889
2187
    def test_exclude_tests(self):
2890
2188
        test_list = ['bogus']
2891
2189
        loader = self._create_loader(test_list)
 
2190
 
2892
2191
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2893
2192
        self.assertEquals([], _test_ids(suite))
2894
2193
 
2939
2238
        tpr.register('bar', 'bbb.aaa.rrr')
2940
2239
        tpr.register('bar', 'bBB.aAA.rRR')
2941
2240
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
2942
 
        self.assertThat(self.get_log(),
2943
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
 
2241
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
2242
                              r'.*bar.*bbb.aaa.rrr.*bBB.aAA.rRR')
2944
2243
 
2945
2244
    def test_get_unknown_prefix(self):
2946
2245
        tpr = self._get_registry()
2982
2281
                                                self.verbosity)
2983
2282
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2984
2283
        self.assertLength(1, calls)
 
2284
 
 
2285
    def test_done(self):
 
2286
        """run_suite should call result.done()"""
 
2287
        self.calls = 0
 
2288
        def one_more_call(): self.calls += 1
 
2289
        def test_function():
 
2290
            pass
 
2291
        test = unittest.FunctionTestCase(test_function)
 
2292
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
2293
            def done(self): one_more_call()
 
2294
        class MyRunner(tests.TextTestRunner):
 
2295
            def run(self, test):
 
2296
                return InstrumentedTestResult(self.stream, self.descriptions,
 
2297
                                              self.verbosity)
 
2298
        tests.run_suite(test, runner_class=MyRunner, stream=StringIO())
 
2299
        self.assertEquals(1, self.calls)