← 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 04:02:41 UTC
  • mfrom: (5017.2.2 tariff)
  • Revision ID: pqm@pqm.ubuntu.com-20100211040241-w6n021dz0uus341n
(mbp) add import-tariff tests

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_selftest.py
1
 
# Copyright (C) 2005, 2006, 2007, 2008, 2009 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
from doctest import ELLIPSIS
20
21
import os
 
22
import signal
21
23
import sys
22
24
import time
23
25
import unittest
24
26
import warnings
25
27
 
 
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
 
26
36
import bzrlib
27
37
from bzrlib import (
28
38
    branchbuilder,
37
47
    repository,
38
48
    symbol_versioning,
39
49
    tests,
 
50
    transport,
40
51
    workingtree,
41
52
    )
42
53
from bzrlib.repofmt import (
 
54
    groupcompress_repo,
43
55
    pack_repo,
44
56
    weaverepo,
45
57
    )
49
61
    deprecated_method,
50
62
    )
51
63
from bzrlib.tests import (
 
64
    features,
 
65
    stub_sftp,
52
66
    test_lsprof,
53
67
    test_sftp_transport,
54
68
    TestUtil,
74
88
                          TestUtil._load_module_by_name,
75
89
                          'bzrlib.no-name-yet')
76
90
 
 
91
 
77
92
class MetaTestLog(tests.TestCase):
78
93
 
79
94
    def test_logging(self):
80
95
        """Test logs are captured when a test fails."""
81
96
        self.log('a test message')
82
 
        self._log_file.flush()
83
 
        self.assertContainsRe(self._get_log(keep_log_file=True),
84
 
                              'a test message\n')
 
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))
85
104
 
86
105
 
87
106
class TestUnicodeFilename(tests.TestCase):
124
143
        self.assertEqual(sample_permutation,
125
144
                         get_transport_test_permutations(MockModule()))
126
145
 
127
 
    def test_scenarios_invlude_all_modules(self):
 
146
    def test_scenarios_include_all_modules(self):
128
147
        # this checks that the scenario generator returns as many permutations
129
148
        # as there are in all the registered transport modules - we assume if
130
149
        # this matches its probably doing the right thing especially in
215
234
        from bzrlib.tests.per_repository import formats_to_scenarios
216
235
        formats = [("(c)", remote.RemoteRepositoryFormat()),
217
236
                   ("(d)", repository.format_registry.get(
218
 
                        'Bazaar pack repository format 1 (needs bzr 0.92)\n'))]
 
237
                    'Bazaar repository format 2a (needs bzr 1.16 or later)\n'))]
219
238
        no_vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
220
239
            None)
221
240
        vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
222
241
            vfs_transport_factory="vfs")
223
242
        # no_vfs generate scenarios without vfs_transport_factory
224
 
        self.assertEqual([
 
243
        expected = [
225
244
            ('RemoteRepositoryFormat(c)',
226
245
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
227
246
              'repository_format': remote.RemoteRepositoryFormat(),
228
247
              'transport_readonly_server': 'readonly',
229
248
              'transport_server': 'server'}),
230
 
            ('RepositoryFormatKnitPack1(d)',
 
249
            ('RepositoryFormat2a(d)',
231
250
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
232
 
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
 
251
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
233
252
              'transport_readonly_server': 'readonly',
234
 
              'transport_server': 'server'})],
235
 
            no_vfs_scenarios)
 
253
              'transport_server': 'server'})]
 
254
        self.assertEqual(expected, no_vfs_scenarios)
236
255
        self.assertEqual([
237
256
            ('RemoteRepositoryFormat(c)',
238
257
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
240
259
              'transport_readonly_server': 'readonly',
241
260
              'transport_server': 'server',
242
261
              'vfs_transport_factory': 'vfs'}),
243
 
            ('RepositoryFormatKnitPack1(d)',
 
262
            ('RepositoryFormat2a(d)',
244
263
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
245
 
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
 
264
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
246
265
              'transport_readonly_server': 'readonly',
247
266
              'transport_server': 'server',
248
267
              'vfs_transport_factory': 'vfs'})],
293
312
        from bzrlib.tests.per_interrepository import make_scenarios
294
313
        server1 = "a"
295
314
        server2 = "b"
296
 
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
 
315
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
297
316
        scenarios = make_scenarios(server1, server2, formats)
298
317
        self.assertEqual([
299
 
            ('str,str,str',
300
 
             {'interrepo_class': str,
301
 
              'repository_format': 'C1',
 
318
            ('C0,str,str',
 
319
             {'repository_format': 'C1',
302
320
              'repository_format_to': 'C2',
303
321
              'transport_readonly_server': 'b',
304
322
              'transport_server': 'a'}),
305
 
            ('int,str,str',
306
 
             {'interrepo_class': int,
307
 
              'repository_format': 'D1',
 
323
            ('D0,str,str',
 
324
             {'repository_format': 'D1',
308
325
              'repository_format_to': 'D2',
309
326
              'transport_readonly_server': 'b',
310
327
              'transport_server': 'a'})],
488
505
        self.assertEqualStat(real, fake)
489
506
 
490
507
    def test_assertEqualStat_notequal(self):
491
 
        self.build_tree(["foo", "bar"])
 
508
        self.build_tree(["foo", "longname"])
492
509
        self.assertRaises(AssertionError, self.assertEqualStat,
493
 
            os.lstat("foo"), os.lstat("bar"))
 
510
            os.lstat("foo"), os.lstat("longname"))
494
511
 
495
512
 
496
513
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
514
531
        cwd = osutils.getcwd()
515
532
        self.assertIsSameRealPath(self.test_dir, cwd)
516
533
 
 
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
 
517
542
    def test_make_branch_and_memory_tree(self):
518
543
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
519
544
 
575
600
                         self.get_transport().get_bytes(
576
601
                            'dir/.bzr/repository/format'))
577
602
 
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
 
 
591
603
    def test_dangling_locks_cause_failures(self):
592
604
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
593
605
            def test_function(self):
597
609
                l.attempt_lock()
598
610
        test = TestDanglingLock('test_function')
599
611
        result = test.run()
600
 
        self.assertEqual(1, len(result.errors))
 
612
        if self._lock_check_thorough:
 
613
            self.assertEqual(1, len(result.errors))
 
614
        else:
 
615
            # When _lock_check_thorough is disabled, then we don't trigger a
 
616
            # failure
 
617
            self.assertEqual(0, len(result.errors))
601
618
 
602
619
 
603
620
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
681
698
        self.assertEqual(url, t.clone('..').base)
682
699
 
683
700
 
684
 
class MockProgress(progress._BaseProgressBar):
685
 
    """Progress-bar standin that records calls.
686
 
 
687
 
    Useful for testing pb using code.
688
 
    """
689
 
 
690
 
    def __init__(self):
691
 
        progress._BaseProgressBar.__init__(self)
692
 
        self.calls = []
693
 
 
694
 
    def tick(self):
695
 
        self.calls.append(('tick',))
696
 
 
697
 
    def update(self, msg=None, current=None, total=None):
698
 
        self.calls.append(('update', msg, current, total))
699
 
 
700
 
    def clear(self):
701
 
        self.calls.append(('clear',))
702
 
 
703
 
    def note(self, msg, *args):
704
 
        self.calls.append(('note', msg, args))
 
701
class TestProfileResult(tests.TestCase):
 
702
 
 
703
    def test_profiles_tests(self):
 
704
        self.requireFeature(test_lsprof.LSProfFeature)
 
705
        terminal = testtools.tests.helpers.ExtendedTestResult()
 
706
        result = tests.ProfileResult(terminal)
 
707
        class Sample(tests.TestCase):
 
708
            def a(self):
 
709
                self.sample_function()
 
710
            def sample_function(self):
 
711
                pass
 
712
        test = Sample("a")
 
713
        test.run(result)
 
714
        case = terminal._events[0][1]
 
715
        self.assertLength(1, case._benchcalls)
 
716
        # We must be able to unpack it as the test reporting code wants
 
717
        (_, _, _), stats = case._benchcalls[0]
 
718
        self.assertTrue(callable(stats.pprint))
705
719
 
706
720
 
707
721
class TestTestResult(tests.TestCase):
711
725
                descriptions=0,
712
726
                verbosity=1,
713
727
                )
714
 
        test_case.run(result)
715
 
        timed_string = result._testTimeString(test_case)
 
728
        capture = testtools.tests.helpers.ExtendedTestResult()
 
729
        test_case.run(MultiTestResult(result, capture))
 
730
        run_case = capture._events[0][1]
 
731
        timed_string = result._testTimeString(run_case)
716
732
        self.assertContainsRe(timed_string, expected_re)
717
733
 
718
734
    def test_test_reporting(self):
736
752
        self.check_timing(ShortDelayTestCase('test_short_delay'),
737
753
                          r"^ +[0-9]+ms$")
738
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
 
739
760
    def test_assigned_benchmark_file_stores_date(self):
 
761
        self._patch_get_bzr_source_tree()
740
762
        output = StringIO()
741
763
        result = bzrlib.tests.TextTestResult(self._log_file,
742
764
                                        descriptions=0,
750
772
        self.assertContainsRe(output_string, "--date [0-9.]+")
751
773
 
752
774
    def test_benchhistory_records_test_times(self):
 
775
        self._patch_get_bzr_source_tree()
753
776
        result_stream = StringIO()
754
777
        result = bzrlib.tests.TextTestResult(
755
778
            self._log_file,
817
840
    def test_known_failure(self):
818
841
        """A KnownFailure being raised should trigger several result actions."""
819
842
        class InstrumentedTestResult(tests.ExtendedTestResult):
820
 
            def done(self): pass
 
843
            def stopTestRun(self): pass
821
844
            def startTests(self): pass
822
845
            def report_test_start(self, test): pass
823
 
            def report_known_failure(self, test, err):
824
 
                self._call = test, err
 
846
            def report_known_failure(self, test, err=None, details=None):
 
847
                self._call = test, 'known failure'
825
848
        result = InstrumentedTestResult(None, None, None, None)
826
 
        def test_function():
827
 
            raise tests.KnownFailure('failed!')
828
 
        test = unittest.FunctionTestCase(test_function)
 
849
        class Test(tests.TestCase):
 
850
            def test_function(self):
 
851
                raise tests.KnownFailure('failed!')
 
852
        test = Test("test_function")
829
853
        test.run(result)
830
854
        # it should invoke 'report_known_failure'.
831
855
        self.assertEqual(2, len(result._call))
832
 
        self.assertEqual(test, result._call[0])
833
 
        self.assertEqual(tests.KnownFailure, result._call[1][0])
834
 
        self.assertIsInstance(result._call[1][1], tests.KnownFailure)
 
856
        self.assertEqual(test.id(), result._call[0].id())
 
857
        self.assertEqual('known failure', result._call[1])
835
858
        # we dont introspec the traceback, if the rest is ok, it would be
836
859
        # exceptional for it not to be.
837
860
        # it should update the known_failure_count on the object.
862
885
        self.assertEqual(lines[1], '    foo')
863
886
        self.assertEqual(2, len(lines))
864
887
 
865
 
    def test_text_report_known_failure(self):
866
 
        # text test output formatting
867
 
        pb = MockProgress()
868
 
        result = bzrlib.tests.TextTestResult(
869
 
            StringIO(),
870
 
            descriptions=0,
871
 
            verbosity=1,
872
 
            pb=pb,
873
 
            )
874
 
        test = self.get_passing_test()
875
 
        # this seeds the state to handle reporting the test.
876
 
        result.startTest(test)
877
 
        # the err parameter has the shape:
878
 
        # (class, exception object, traceback)
879
 
        # KnownFailures dont get their tracebacks shown though, so we
880
 
        # can skip that.
881
 
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
882
 
        result.report_known_failure(test, err)
883
 
        self.assertEqual(
884
 
            [
885
 
            ('update', '[1 in 0s] passing_test', None, None),
886
 
            ('note', 'XFAIL: %s\n%s\n', ('passing_test', err[1]))
887
 
            ],
888
 
            pb.calls)
889
 
        # known_failures should be printed in the summary, so if we run a test
890
 
        # after there are some known failures, the update prefix should match
891
 
        # this.
892
 
        result.known_failure_count = 3
893
 
        test.run(result)
894
 
        self.assertEqual(
895
 
            [
896
 
            ('update', '[2 in 0s] passing_test', None, None),
897
 
            ],
898
 
            pb.calls[2:])
899
 
 
900
888
    def get_passing_test(self):
901
889
        """Return a test object that can't be run usefully."""
902
890
        def passing_test():
906
894
    def test_add_not_supported(self):
907
895
        """Test the behaviour of invoking addNotSupported."""
908
896
        class InstrumentedTestResult(tests.ExtendedTestResult):
909
 
            def done(self): pass
 
897
            def stopTestRun(self): pass
910
898
            def startTests(self): pass
911
899
            def report_test_start(self, test): pass
912
900
            def report_unsupported(self, test, feature):
944
932
        result.report_unsupported(test, feature)
945
933
        output = result_stream.getvalue()[prefix:]
946
934
        lines = output.splitlines()
947
 
        self.assertEqual(lines, ['NODEP        0ms',
948
 
                                 "    The feature 'Feature' is not available."])
949
 
 
950
 
    def test_text_report_unsupported(self):
951
 
        # text test output formatting
952
 
        pb = MockProgress()
953
 
        result = bzrlib.tests.TextTestResult(
954
 
            StringIO(),
955
 
            descriptions=0,
956
 
            verbosity=1,
957
 
            pb=pb,
958
 
            )
959
 
        test = self.get_passing_test()
960
 
        feature = tests.Feature()
961
 
        # this seeds the state to handle reporting the test.
962
 
        result.startTest(test)
963
 
        result.report_unsupported(test, feature)
964
 
        # no output on unsupported features
965
 
        self.assertEqual(
966
 
            [('update', '[1 in 0s] passing_test', None, None)
967
 
            ],
968
 
            pb.calls)
969
 
        # the number of missing features should be printed in the progress
970
 
        # summary, so check for that.
971
 
        result.unsupported = {'foo':0, 'bar':0}
972
 
        test.run(result)
973
 
        self.assertEqual(
974
 
            [
975
 
            ('update', '[2 in 0s, 2 missing] passing_test', None, None),
976
 
            ],
977
 
            pb.calls[1:])
 
935
        # We don't check for the final '0ms' since it may fail on slow hosts
 
936
        self.assertStartsWith(lines[0], 'NODEP')
 
937
        self.assertEqual(lines[1],
 
938
                         "    The feature 'Feature' is not available.")
978
939
 
979
940
    def test_unavailable_exception(self):
980
941
        """An UnavailableFeature being raised should invoke addNotSupported."""
981
942
        class InstrumentedTestResult(tests.ExtendedTestResult):
982
 
            def done(self): pass
 
943
            def stopTestRun(self): pass
983
944
            def startTests(self): pass
984
945
            def report_test_start(self, test): pass
985
946
            def addNotSupported(self, test, feature):
986
947
                self._call = test, feature
987
948
        result = InstrumentedTestResult(None, None, None, None)
988
949
        feature = tests.Feature()
989
 
        def test_function():
990
 
            raise tests.UnavailableFeature(feature)
991
 
        test = unittest.FunctionTestCase(test_function)
 
950
        class Test(tests.TestCase):
 
951
            def test_function(self):
 
952
                raise tests.UnavailableFeature(feature)
 
953
        test = Test("test_function")
992
954
        test.run(result)
993
955
        # it should invoke 'addNotSupported'.
994
956
        self.assertEqual(2, len(result._call))
995
 
        self.assertEqual(test, result._call[0])
 
957
        self.assertEqual(test.id(), result._call[0].id())
996
958
        self.assertEqual(feature, result._call[1])
997
959
        # and not count as an error
998
960
        self.assertEqual(0, result.error_count)
1011
973
                                             verbosity=1)
1012
974
        test = self.get_passing_test()
1013
975
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
1014
 
        result._addKnownFailure(test, err)
 
976
        result.addExpectedFailure(test, err)
1015
977
        self.assertFalse(result.wasStrictlySuccessful())
1016
978
        self.assertEqual(None, result._extractBenchmarkTime(test))
1017
979
 
1062
1024
        because of our use of global state.
1063
1025
        """
1064
1026
        old_root = tests.TestCaseInTempDir.TEST_ROOT
 
1027
        old_leak = tests.TestCase._first_thread_leaker_id
1065
1028
        try:
1066
1029
            tests.TestCaseInTempDir.TEST_ROOT = None
 
1030
            tests.TestCase._first_thread_leaker_id = None
1067
1031
            return testrunner.run(test)
1068
1032
        finally:
1069
1033
            tests.TestCaseInTempDir.TEST_ROOT = old_root
 
1034
            tests.TestCase._first_thread_leaker_id = old_leak
1070
1035
 
1071
1036
    def test_known_failure_failed_run(self):
1072
1037
        # run a test that generates a known failure which should be printed in
1073
1038
        # the final output when real failures occur.
1074
 
        def known_failure_test():
1075
 
            raise tests.KnownFailure('failed')
 
1039
        class Test(tests.TestCase):
 
1040
            def known_failure_test(self):
 
1041
                self.expectFailure('failed', self.assertTrue, False)
1076
1042
        test = unittest.TestSuite()
1077
 
        test.addTest(unittest.FunctionTestCase(known_failure_test))
 
1043
        test.addTest(Test("known_failure_test"))
1078
1044
        def failing_test():
1079
 
            raise AssertionError('foo')
 
1045
            self.fail('foo')
1080
1046
        test.addTest(unittest.FunctionTestCase(failing_test))
1081
1047
        stream = StringIO()
1082
1048
        runner = tests.TextTestRunner(stream=stream)
1083
1049
        result = self.run_test_runner(runner, test)
1084
1050
        lines = stream.getvalue().splitlines()
1085
 
        self.assertEqual([
1086
 
            '',
1087
 
            '======================================================================',
1088
 
            'FAIL: unittest.FunctionTestCase (failing_test)',
1089
 
            '----------------------------------------------------------------------',
1090
 
            'Traceback (most recent call last):',
1091
 
            '    raise AssertionError(\'foo\')',
1092
 
            'AssertionError: foo',
1093
 
            '',
1094
 
            '----------------------------------------------------------------------',
1095
 
            '',
1096
 
            'FAILED (failures=1, known_failure_count=1)'],
1097
 
            lines[3:8] + lines[9:13] + lines[14:])
 
1051
        self.assertContainsRe(stream.getvalue(),
 
1052
            '(?sm)^bzr selftest.*$'
 
1053
            '.*'
 
1054
            '^======================================================================\n'
 
1055
            '^FAIL: failing_test\n'
 
1056
            '^----------------------------------------------------------------------\n'
 
1057
            'Traceback \\(most recent call last\\):\n'
 
1058
            '  .*' # File .*, line .*, in failing_test' - but maybe not from .pyc
 
1059
            '    self.fail\\(\'foo\'\\)\n'
 
1060
            '.*'
 
1061
            '^----------------------------------------------------------------------\n'
 
1062
            '.*'
 
1063
            'FAILED \\(failures=1, known_failure_count=1\\)'
 
1064
            )
1098
1065
 
1099
1066
    def test_known_failure_ok_run(self):
1100
 
        # run a test that generates a known failure which should be printed in the final output.
1101
 
        def known_failure_test():
1102
 
            raise tests.KnownFailure('failed')
1103
 
        test = unittest.FunctionTestCase(known_failure_test)
 
1067
        # run a test that generates a known failure which should be printed in
 
1068
        # the final output.
 
1069
        class Test(tests.TestCase):
 
1070
            def known_failure_test(self):
 
1071
                self.expectFailure('failed', self.assertTrue, False)
 
1072
        test = Test("known_failure_test")
1104
1073
        stream = StringIO()
1105
1074
        runner = tests.TextTestRunner(stream=stream)
1106
1075
        result = self.run_test_runner(runner, test)
1111
1080
            '\n'
1112
1081
            'OK \\(known_failures=1\\)\n')
1113
1082
 
 
1083
    def test_result_decorator(self):
 
1084
        # decorate results
 
1085
        calls = []
 
1086
        class LoggingDecorator(tests.ForwardingResult):
 
1087
            def startTest(self, test):
 
1088
                tests.ForwardingResult.startTest(self, test)
 
1089
                calls.append('start')
 
1090
        test = unittest.FunctionTestCase(lambda:None)
 
1091
        stream = StringIO()
 
1092
        runner = tests.TextTestRunner(stream=stream,
 
1093
            result_decorators=[LoggingDecorator])
 
1094
        result = self.run_test_runner(runner, test)
 
1095
        self.assertLength(1, calls)
 
1096
 
1114
1097
    def test_skipped_test(self):
1115
1098
        # run a test that is skipped, and check the suite as a whole still
1116
1099
        # succeeds.
1169
1152
 
1170
1153
    def test_not_applicable(self):
1171
1154
        # run a test that is skipped because it's not applicable
1172
 
        def not_applicable_test():
1173
 
            raise tests.TestNotApplicable('this test never runs')
 
1155
        class Test(tests.TestCase):
 
1156
            def not_applicable_test(self):
 
1157
                raise tests.TestNotApplicable('this test never runs')
1174
1158
        out = StringIO()
1175
1159
        runner = tests.TextTestRunner(stream=out, verbosity=2)
1176
 
        test = unittest.FunctionTestCase(not_applicable_test)
 
1160
        test = Test("not_applicable_test")
1177
1161
        result = self.run_test_runner(runner, test)
1178
1162
        self._log_file.write(out.getvalue())
1179
1163
        self.assertTrue(result.wasSuccessful())
1183
1167
        self.assertContainsRe(out.getvalue(),
1184
1168
                r'(?m)^    this test never runs')
1185
1169
 
1186
 
    def test_not_applicable_demo(self):
1187
 
        # just so you can see it in the test output
1188
 
        raise tests.TestNotApplicable('this test is just a demonstation')
1189
 
 
1190
1170
    def test_unsupported_features_listed(self):
1191
1171
        """When unsupported features are encountered they are detailed."""
1192
1172
        class Feature1(tests.Feature):
1212
1192
            ],
1213
1193
            lines[-3:])
1214
1194
 
 
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
 
1215
1204
    def test_bench_history(self):
1216
 
        # tests that the running the benchmark produces a history file
1217
 
        # containing a timestamp and the revision id of the bzrlib source which
1218
 
        # was tested.
1219
 
        workingtree = _get_bzr_source_tree()
 
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()
1220
1209
        test = TestRunner('dummy_test')
1221
1210
        output = StringIO()
1222
1211
        runner = tests.TextTestRunner(stream=self._log_file,
1224
1213
        result = self.run_test_runner(runner, test)
1225
1214
        output_string = output.getvalue()
1226
1215
        self.assertContainsRe(output_string, "--date [0-9.]+")
1227
 
        if workingtree is not None:
1228
 
            revision_id = workingtree.get_parent_ids()[0]
1229
 
            self.assertEndsWith(output_string.rstrip(), revision_id)
1230
 
 
1231
 
    def assertLogDeleted(self, test):
1232
 
        log = test._get_log()
1233
 
        self.assertEqual("DELETED log file to reduce memory footprint", log)
1234
 
        self.assertEqual('', test._log_contents)
1235
 
        self.assertIs(None, test._log_file_name)
1236
 
 
1237
 
    def test_success_log_deleted(self):
1238
 
        """Successful tests have their log deleted"""
1239
 
 
1240
 
        class LogTester(tests.TestCase):
1241
 
 
1242
 
            def test_success(self):
1243
 
                self.log('this will be removed\n')
1244
 
 
1245
 
        sio = StringIO()
1246
 
        runner = tests.TextTestRunner(stream=sio)
1247
 
        test = LogTester('test_success')
1248
 
        result = self.run_test_runner(runner, test)
1249
 
 
1250
 
        self.assertLogDeleted(test)
1251
 
 
1252
 
    def test_skipped_log_deleted(self):
1253
 
        """Skipped tests have their log deleted"""
1254
 
 
1255
 
        class LogTester(tests.TestCase):
1256
 
 
1257
 
            def test_skipped(self):
1258
 
                self.log('this will be removed\n')
1259
 
                raise tests.TestSkipped()
1260
 
 
1261
 
        sio = StringIO()
1262
 
        runner = tests.TextTestRunner(stream=sio)
1263
 
        test = LogTester('test_skipped')
1264
 
        result = self.run_test_runner(runner, test)
1265
 
 
1266
 
        self.assertLogDeleted(test)
1267
 
 
1268
 
    def test_not_aplicable_log_deleted(self):
1269
 
        """Not applicable tests have their log deleted"""
1270
 
 
1271
 
        class LogTester(tests.TestCase):
1272
 
 
1273
 
            def test_not_applicable(self):
1274
 
                self.log('this will be removed\n')
1275
 
                raise tests.TestNotApplicable()
1276
 
 
1277
 
        sio = StringIO()
1278
 
        runner = tests.TextTestRunner(stream=sio)
1279
 
        test = LogTester('test_not_applicable')
1280
 
        result = self.run_test_runner(runner, test)
1281
 
 
1282
 
        self.assertLogDeleted(test)
1283
 
 
1284
 
    def test_known_failure_log_deleted(self):
1285
 
        """Know failure tests have their log deleted"""
1286
 
 
1287
 
        class LogTester(tests.TestCase):
1288
 
 
1289
 
            def test_known_failure(self):
1290
 
                self.log('this will be removed\n')
1291
 
                raise tests.KnownFailure()
1292
 
 
1293
 
        sio = StringIO()
1294
 
        runner = tests.TextTestRunner(stream=sio)
1295
 
        test = LogTester('test_known_failure')
1296
 
        result = self.run_test_runner(runner, test)
1297
 
 
1298
 
        self.assertLogDeleted(test)
1299
 
 
1300
 
    def test_fail_log_kept(self):
1301
 
        """Failed tests have their log kept"""
1302
 
 
1303
 
        class LogTester(tests.TestCase):
1304
 
 
1305
 
            def test_fail(self):
1306
 
                self.log('this will be kept\n')
1307
 
                self.fail('this test fails')
1308
 
 
1309
 
        sio = StringIO()
1310
 
        runner = tests.TextTestRunner(stream=sio)
1311
 
        test = LogTester('test_fail')
1312
 
        result = self.run_test_runner(runner, test)
1313
 
 
1314
 
        text = sio.getvalue()
1315
 
        self.assertContainsRe(text, 'this will be kept')
1316
 
        self.assertContainsRe(text, 'this test fails')
1317
 
 
1318
 
        log = test._get_log()
1319
 
        self.assertContainsRe(log, 'this will be kept')
1320
 
        self.assertEqual(log, test._log_contents)
1321
 
 
1322
 
    def test_error_log_kept(self):
1323
 
        """Tests with errors have their log kept"""
1324
 
 
1325
 
        class LogTester(tests.TestCase):
1326
 
 
1327
 
            def test_error(self):
1328
 
                self.log('this will be kept\n')
1329
 
                raise ValueError('random exception raised')
1330
 
 
1331
 
        sio = StringIO()
1332
 
        runner = tests.TextTestRunner(stream=sio)
1333
 
        test = LogTester('test_error')
1334
 
        result = self.run_test_runner(runner, test)
1335
 
 
1336
 
        text = sio.getvalue()
1337
 
        self.assertContainsRe(text, 'this will be kept')
1338
 
        self.assertContainsRe(text, 'random exception raised')
1339
 
 
1340
 
        log = test._get_log()
1341
 
        self.assertContainsRe(log, 'this will be kept')
1342
 
        self.assertEqual(log, test._log_contents)
 
1216
        self.assertLength(1, self._get_source_tree_calls)
 
1217
 
 
1218
    def test_startTestRun(self):
 
1219
        """run should call result.startTestRun()"""
 
1220
        calls = []
 
1221
        class LoggingDecorator(tests.ForwardingResult):
 
1222
            def startTestRun(self):
 
1223
                tests.ForwardingResult.startTestRun(self)
 
1224
                calls.append('startTestRun')
 
1225
        test = unittest.FunctionTestCase(lambda:None)
 
1226
        stream = StringIO()
 
1227
        runner = tests.TextTestRunner(stream=stream,
 
1228
            result_decorators=[LoggingDecorator])
 
1229
        result = self.run_test_runner(runner, test)
 
1230
        self.assertLength(1, calls)
 
1231
 
 
1232
    def test_stopTestRun(self):
 
1233
        """run should call result.stopTestRun()"""
 
1234
        calls = []
 
1235
        class LoggingDecorator(tests.ForwardingResult):
 
1236
            def stopTestRun(self):
 
1237
                tests.ForwardingResult.stopTestRun(self)
 
1238
                calls.append('stopTestRun')
 
1239
        test = unittest.FunctionTestCase(lambda:None)
 
1240
        stream = StringIO()
 
1241
        runner = tests.TextTestRunner(stream=stream,
 
1242
            result_decorators=[LoggingDecorator])
 
1243
        result = self.run_test_runner(runner, test)
 
1244
        self.assertLength(1, calls)
1343
1245
 
1344
1246
 
1345
1247
class SampleTestCase(tests.TestCase):
1350
1252
class _TestException(Exception):
1351
1253
    pass
1352
1254
 
 
1255
 
1353
1256
class TestTestCase(tests.TestCase):
1354
1257
    """Tests that test the core bzrlib TestCase."""
1355
1258
 
1404
1307
        # we could set something and run a test that will check
1405
1308
        # it gets santised, but this is probably sufficient for now:
1406
1309
        # if someone runs the test with -Dsomething it will error.
1407
 
        self.assertEqual(set(), bzrlib.debug.debug_flags)
 
1310
        flags = set()
 
1311
        if self._lock_check_thorough:
 
1312
            flags.add('strict_locks')
 
1313
        self.assertEqual(flags, bzrlib.debug.debug_flags)
1408
1314
 
1409
1315
    def change_selftest_debug_flags(self, new_flags):
1410
 
        orig_selftest_flags = tests.selftest_debug_flags
1411
 
        self.addCleanup(self._restore_selftest_debug_flags, orig_selftest_flags)
1412
 
        tests.selftest_debug_flags = set(new_flags)
1413
 
 
1414
 
    def _restore_selftest_debug_flags(self, flags):
1415
 
        tests.selftest_debug_flags = flags
 
1316
        self.overrideAttr(tests, 'selftest_debug_flags', set(new_flags))
1416
1317
 
1417
1318
    def test_allow_debug_flag(self):
1418
1319
        """The -Eallow_debug flag prevents bzrlib.debug.debug_flags from being
1425
1326
                self.flags = set(bzrlib.debug.debug_flags)
1426
1327
        test = TestThatRecordsFlags('test_foo')
1427
1328
        test.run(self.make_test_result())
1428
 
        self.assertEqual(set(['a-flag']), self.flags)
 
1329
        flags = set(['a-flag'])
 
1330
        if 'disable_lock_checks' not in tests.selftest_debug_flags:
 
1331
            flags.add('strict_locks')
 
1332
        self.assertEqual(flags, self.flags)
 
1333
 
 
1334
    def test_disable_lock_checks(self):
 
1335
        """The -Edisable_lock_checks flag disables thorough checks."""
 
1336
        class TestThatRecordsFlags(tests.TestCase):
 
1337
            def test_foo(nested_self):
 
1338
                self.flags = set(bzrlib.debug.debug_flags)
 
1339
                self.test_lock_check_thorough = nested_self._lock_check_thorough
 
1340
        self.change_selftest_debug_flags(set())
 
1341
        test = TestThatRecordsFlags('test_foo')
 
1342
        test.run(self.make_test_result())
 
1343
        # By default we do strict lock checking and thorough lock/unlock
 
1344
        # tracking.
 
1345
        self.assertTrue(self.test_lock_check_thorough)
 
1346
        self.assertEqual(set(['strict_locks']), self.flags)
 
1347
        # Now set the disable_lock_checks flag, and show that this changed.
 
1348
        self.change_selftest_debug_flags(set(['disable_lock_checks']))
 
1349
        test = TestThatRecordsFlags('test_foo')
 
1350
        test.run(self.make_test_result())
 
1351
        self.assertFalse(self.test_lock_check_thorough)
 
1352
        self.assertEqual(set(), self.flags)
 
1353
 
 
1354
    def test_this_fails_strict_lock_check(self):
 
1355
        class TestThatRecordsFlags(tests.TestCase):
 
1356
            def test_foo(nested_self):
 
1357
                self.flags1 = set(bzrlib.debug.debug_flags)
 
1358
                self.thisFailsStrictLockCheck()
 
1359
                self.flags2 = set(bzrlib.debug.debug_flags)
 
1360
        # Make sure lock checking is active
 
1361
        self.change_selftest_debug_flags(set())
 
1362
        test = TestThatRecordsFlags('test_foo')
 
1363
        test.run(self.make_test_result())
 
1364
        self.assertEqual(set(['strict_locks']), self.flags1)
 
1365
        self.assertEqual(set(), self.flags2)
1429
1366
 
1430
1367
    def test_debug_flags_restored(self):
1431
1368
        """The bzrlib debug flags should be restored to their original state
1442
1379
        self.assertEqual(set(['original-state']), bzrlib.debug.debug_flags)
1443
1380
 
1444
1381
    def make_test_result(self):
 
1382
        """Get a test result that writes to the test log file."""
1445
1383
        return tests.TextTestResult(self._log_file, descriptions=0, verbosity=1)
1446
1384
 
1447
1385
    def inner_test(self):
1455
1393
        result = self.make_test_result()
1456
1394
        self.inner_test.run(result)
1457
1395
        note("outer finish")
 
1396
        self.addCleanup(osutils.delete_any, self._log_file_name)
1458
1397
 
1459
1398
    def test_trace_nesting(self):
1460
1399
        # this tests that each test case nests its trace facility correctly.
1486
1425
        result = bzrlib.tests.VerboseTestResult(
1487
1426
            unittest._WritelnDecorator(output_stream),
1488
1427
            descriptions=0,
1489
 
            verbosity=2,
1490
 
            num_tests=sample_test.countTestCases())
 
1428
            verbosity=2)
1491
1429
        sample_test.run(result)
1492
1430
        self.assertContainsRe(
1493
1431
            output_stream.getvalue(),
1521
1459
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1522
1460
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1523
1461
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
 
1462
        del self._benchcalls[:]
1524
1463
 
1525
1464
    def test_knownFailure(self):
1526
1465
        """Self.knownFailure() should raise a KnownFailure exception."""
1527
1466
        self.assertRaises(tests.KnownFailure, self.knownFailure, "A Failure")
1528
1467
 
 
1468
    def test_open_bzrdir_safe_roots(self):
 
1469
        # even a memory transport should fail to open when its url isn't 
 
1470
        # permitted.
 
1471
        # Manually set one up (TestCase doesn't and shouldn't provide magic
 
1472
        # machinery)
 
1473
        transport_server = MemoryServer()
 
1474
        transport_server.start_server()
 
1475
        self.addCleanup(transport_server.stop_server)
 
1476
        t = transport.get_transport(transport_server.get_url())
 
1477
        bzrdir.BzrDir.create(t.base)
 
1478
        self.assertRaises(errors.BzrError,
 
1479
            bzrdir.BzrDir.open_from_transport, t)
 
1480
        # But if we declare this as safe, we can open the bzrdir.
 
1481
        self.permit_url(t.base)
 
1482
        self._bzr_selftest_roots.append(t.base)
 
1483
        bzrdir.BzrDir.open_from_transport(t)
 
1484
 
1529
1485
    def test_requireFeature_available(self):
1530
1486
        """self.requireFeature(available) is a no-op."""
1531
1487
        class Available(tests.Feature):
1575
1531
        """Test disabled tests behaviour with support aware results."""
1576
1532
        test = SampleTestCase('_test_pass')
1577
1533
        class DisabledFeature(object):
 
1534
            def __eq__(self, other):
 
1535
                return isinstance(other, DisabledFeature)
1578
1536
            def available(self):
1579
1537
                return False
1580
1538
        the_feature = DisabledFeature()
1591
1549
                self.calls.append(('addNotSupported', test, feature))
1592
1550
        result = InstrumentedTestResult()
1593
1551
        test.run(result)
 
1552
        case = result.calls[0][1]
1594
1553
        self.assertEqual([
1595
 
            ('startTest', test),
1596
 
            ('addNotSupported', test, the_feature),
1597
 
            ('stopTest', test),
 
1554
            ('startTest', case),
 
1555
            ('addNotSupported', case, the_feature),
 
1556
            ('stopTest', case),
1598
1557
            ],
1599
1558
            result.calls)
1600
1559
 
 
1560
    def test_start_server_registers_url(self):
 
1561
        transport_server = MemoryServer()
 
1562
        # A little strict, but unlikely to be changed soon.
 
1563
        self.assertEqual([], self._bzr_selftest_roots)
 
1564
        self.start_server(transport_server)
 
1565
        self.assertSubset([transport_server.get_url()],
 
1566
            self._bzr_selftest_roots)
 
1567
 
1601
1568
    def test_assert_list_raises_on_generator(self):
1602
1569
        def generator_which_will_raise():
1603
1570
            # This will not raise until after the first yield
1653
1620
        self.assertRaises(AssertionError,
1654
1621
            self.assertListRaises, _TestException, success_generator)
1655
1622
 
 
1623
    def test_overrideAttr_without_value(self):
 
1624
        self.test_attr = 'original' # Define a test attribute
 
1625
        obj = self # Make 'obj' visible to the embedded test
 
1626
        class Test(tests.TestCase):
 
1627
 
 
1628
            def setUp(self):
 
1629
                tests.TestCase.setUp(self)
 
1630
                self.orig = self.overrideAttr(obj, 'test_attr')
 
1631
 
 
1632
            def test_value(self):
 
1633
                self.assertEqual('original', self.orig)
 
1634
                self.assertEqual('original', obj.test_attr)
 
1635
                obj.test_attr = 'modified'
 
1636
                self.assertEqual('modified', obj.test_attr)
 
1637
 
 
1638
        test = Test('test_value')
 
1639
        test.run(unittest.TestResult())
 
1640
        self.assertEqual('original', obj.test_attr)
 
1641
 
 
1642
    def test_overrideAttr_with_value(self):
 
1643
        self.test_attr = 'original' # Define a test attribute
 
1644
        obj = self # Make 'obj' visible to the embedded test
 
1645
        class Test(tests.TestCase):
 
1646
 
 
1647
            def setUp(self):
 
1648
                tests.TestCase.setUp(self)
 
1649
                self.orig = self.overrideAttr(obj, 'test_attr', new='modified')
 
1650
 
 
1651
            def test_value(self):
 
1652
                self.assertEqual('original', self.orig)
 
1653
                self.assertEqual('modified', obj.test_attr)
 
1654
 
 
1655
        test = Test('test_value')
 
1656
        test.run(unittest.TestResult())
 
1657
        self.assertEqual('original', obj.test_attr)
 
1658
 
1656
1659
 
1657
1660
# NB: Don't delete this; it's not actually from 0.11!
1658
1661
@deprecated_function(deprecated_in((0, 11, 0)))
1701
1704
        self.assertEndsWith('foo', 'oo')
1702
1705
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1703
1706
 
 
1707
    def test_assertEqualDiff(self):
 
1708
        e = self.assertRaises(AssertionError,
 
1709
                              self.assertEqualDiff, '', '\n')
 
1710
        self.assertEquals(str(e),
 
1711
                          # Don't blink ! The '+' applies to the second string
 
1712
                          'first string is missing a final newline.\n+ \n')
 
1713
        e = self.assertRaises(AssertionError,
 
1714
                              self.assertEqualDiff, '\n', '')
 
1715
        self.assertEquals(str(e),
 
1716
                          # Don't blink ! The '-' applies to the second string
 
1717
                          'second string is missing a final newline.\n- \n')
 
1718
 
 
1719
 
 
1720
class TestDeprecations(tests.TestCase):
 
1721
 
1704
1722
    def test_applyDeprecated_not_deprecated(self):
1705
1723
        sample_object = ApplyDeprecatedHelper()
1706
1724
        # calling an undeprecated callable raises an assertion
1783
1801
        tree = self.make_branch_and_memory_tree('a')
1784
1802
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1785
1803
 
1786
 
 
1787
 
class TestSFTPMakeBranchAndTree(test_sftp_transport.TestCaseWithSFTPServer):
1788
 
 
1789
 
    def test_make_tree_for_sftp_branch(self):
1790
 
        """Transports backed by local directories create local trees."""
1791
 
 
 
1804
    def test_make_tree_for_local_vfs_backed_transport(self):
 
1805
        # make_branch_and_tree has to use local branch and repositories
 
1806
        # when the vfs transport and local disk are colocated, even if
 
1807
        # a different transport is in use for url generation.
 
1808
        from bzrlib.transport.fakevfat import FakeVFATServer
 
1809
        self.transport_server = FakeVFATServer
 
1810
        self.assertFalse(self.get_url('t1').startswith('file://'))
1792
1811
        tree = self.make_branch_and_tree('t1')
1793
1812
        base = tree.bzrdir.root_transport.base
1794
 
        self.failIf(base.startswith('sftp'),
1795
 
                'base %r is on sftp but should be local' % base)
 
1813
        self.assertStartsWith(base, 'file://')
1796
1814
        self.assertEquals(tree.bzrdir.root_transport,
1797
1815
                tree.branch.bzrdir.root_transport)
1798
1816
        self.assertEquals(tree.bzrdir.root_transport,
1799
1817
                tree.branch.repository.bzrdir.root_transport)
1800
1818
 
1801
1819
 
1802
 
class TestSelftest(tests.TestCase):
 
1820
class SelfTestHelper:
 
1821
 
 
1822
    def run_selftest(self, **kwargs):
 
1823
        """Run selftest returning its output."""
 
1824
        output = StringIO()
 
1825
        old_transport = bzrlib.tests.default_transport
 
1826
        old_root = tests.TestCaseWithMemoryTransport.TEST_ROOT
 
1827
        tests.TestCaseWithMemoryTransport.TEST_ROOT = None
 
1828
        try:
 
1829
            self.assertEqual(True, tests.selftest(stream=output, **kwargs))
 
1830
        finally:
 
1831
            bzrlib.tests.default_transport = old_transport
 
1832
            tests.TestCaseWithMemoryTransport.TEST_ROOT = old_root
 
1833
        output.seek(0)
 
1834
        return output
 
1835
 
 
1836
 
 
1837
class TestSelftest(tests.TestCase, SelfTestHelper):
1803
1838
    """Tests of bzrlib.tests.selftest."""
1804
1839
 
1805
1840
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
1813
1848
            test_suite_factory=factory)
1814
1849
        self.assertEqual([True], factory_called)
1815
1850
 
1816
 
 
1817
 
class TestKnownFailure(tests.TestCase):
1818
 
 
1819
 
    def test_known_failure(self):
1820
 
        """Check that KnownFailure is defined appropriately."""
1821
 
        # a KnownFailure is an assertion error for compatability with unaware
1822
 
        # runners.
1823
 
        self.assertIsInstance(tests.KnownFailure(""), AssertionError)
1824
 
 
1825
 
    def test_expect_failure(self):
1826
 
        try:
1827
 
            self.expectFailure("Doomed to failure", self.assertTrue, False)
1828
 
        except tests.KnownFailure, e:
1829
 
            self.assertEqual('Doomed to failure', e.args[0])
1830
 
        try:
1831
 
            self.expectFailure("Doomed to failure", self.assertTrue, True)
1832
 
        except AssertionError, e:
1833
 
            self.assertEqual('Unexpected success.  Should have failed:'
1834
 
                             ' Doomed to failure', e.args[0])
 
1851
    def factory(self):
 
1852
        """A test suite factory."""
 
1853
        class Test(tests.TestCase):
 
1854
            def a(self):
 
1855
                pass
 
1856
            def b(self):
 
1857
                pass
 
1858
            def c(self):
 
1859
                pass
 
1860
        return TestUtil.TestSuite([Test("a"), Test("b"), Test("c")])
 
1861
 
 
1862
    def test_list_only(self):
 
1863
        output = self.run_selftest(test_suite_factory=self.factory,
 
1864
            list_only=True)
 
1865
        self.assertEqual(3, len(output.readlines()))
 
1866
 
 
1867
    def test_list_only_filtered(self):
 
1868
        output = self.run_selftest(test_suite_factory=self.factory,
 
1869
            list_only=True, pattern="Test.b")
 
1870
        self.assertEndsWith(output.getvalue(), "Test.b\n")
 
1871
        self.assertLength(1, output.readlines())
 
1872
 
 
1873
    def test_list_only_excludes(self):
 
1874
        output = self.run_selftest(test_suite_factory=self.factory,
 
1875
            list_only=True, exclude_pattern="Test.b")
 
1876
        self.assertNotContainsRe("Test.b", output.getvalue())
 
1877
        self.assertLength(2, output.readlines())
 
1878
 
 
1879
    def test_lsprof_tests(self):
 
1880
        self.requireFeature(test_lsprof.LSProfFeature)
 
1881
        calls = []
 
1882
        class Test(object):
 
1883
            def __call__(test, result):
 
1884
                test.run(result)
 
1885
            def run(test, result):
 
1886
                self.assertIsInstance(result, tests.ForwardingResult)
 
1887
                calls.append("called")
 
1888
            def countTestCases(self):
 
1889
                return 1
 
1890
        self.run_selftest(test_suite_factory=Test, lsprof_tests=True)
 
1891
        self.assertLength(1, calls)
 
1892
 
 
1893
    def test_random(self):
 
1894
        # test randomising by listing a number of tests.
 
1895
        output_123 = self.run_selftest(test_suite_factory=self.factory,
 
1896
            list_only=True, random_seed="123")
 
1897
        output_234 = self.run_selftest(test_suite_factory=self.factory,
 
1898
            list_only=True, random_seed="234")
 
1899
        self.assertNotEqual(output_123, output_234)
 
1900
        # "Randominzing test order..\n\n
 
1901
        self.assertLength(5, output_123.readlines())
 
1902
        self.assertLength(5, output_234.readlines())
 
1903
 
 
1904
    def test_random_reuse_is_same_order(self):
 
1905
        # test randomising by listing a number of tests.
 
1906
        expected = self.run_selftest(test_suite_factory=self.factory,
 
1907
            list_only=True, random_seed="123")
 
1908
        repeated = self.run_selftest(test_suite_factory=self.factory,
 
1909
            list_only=True, random_seed="123")
 
1910
        self.assertEqual(expected.getvalue(), repeated.getvalue())
 
1911
 
 
1912
    def test_runner_class(self):
 
1913
        self.requireFeature(features.subunit)
 
1914
        from subunit import ProtocolTestCase
 
1915
        stream = self.run_selftest(runner_class=tests.SubUnitBzrRunner,
 
1916
            test_suite_factory=self.factory)
 
1917
        test = ProtocolTestCase(stream)
 
1918
        result = unittest.TestResult()
 
1919
        test.run(result)
 
1920
        self.assertEqual(3, result.testsRun)
 
1921
 
 
1922
    def test_starting_with_single_argument(self):
 
1923
        output = self.run_selftest(test_suite_factory=self.factory,
 
1924
            starting_with=['bzrlib.tests.test_selftest.Test.a'],
 
1925
            list_only=True)
 
1926
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n',
 
1927
            output.getvalue())
 
1928
 
 
1929
    def test_starting_with_multiple_argument(self):
 
1930
        output = self.run_selftest(test_suite_factory=self.factory,
 
1931
            starting_with=['bzrlib.tests.test_selftest.Test.a',
 
1932
                'bzrlib.tests.test_selftest.Test.b'],
 
1933
            list_only=True)
 
1934
        self.assertEqual('bzrlib.tests.test_selftest.Test.a\n'
 
1935
            'bzrlib.tests.test_selftest.Test.b\n',
 
1936
            output.getvalue())
 
1937
 
 
1938
    def check_transport_set(self, transport_server):
 
1939
        captured_transport = []
 
1940
        def seen_transport(a_transport):
 
1941
            captured_transport.append(a_transport)
 
1942
        class Capture(tests.TestCase):
 
1943
            def a(self):
 
1944
                seen_transport(bzrlib.tests.default_transport)
 
1945
        def factory():
 
1946
            return TestUtil.TestSuite([Capture("a")])
 
1947
        self.run_selftest(transport=transport_server, test_suite_factory=factory)
 
1948
        self.assertEqual(transport_server, captured_transport[0])
 
1949
 
 
1950
    def test_transport_sftp(self):
 
1951
        self.requireFeature(features.paramiko)
 
1952
        self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
 
1953
 
 
1954
    def test_transport_memory(self):
 
1955
        self.check_transport_set(bzrlib.transport.memory.MemoryServer)
 
1956
 
 
1957
 
 
1958
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
 
1959
    # Does IO: reads test.list
 
1960
 
 
1961
    def test_load_list(self):
 
1962
        # Provide a list with one test - this test.
 
1963
        test_id_line = '%s\n' % self.id()
 
1964
        self.build_tree_contents([('test.list', test_id_line)])
 
1965
        # And generate a list of the tests in  the suite.
 
1966
        stream = self.run_selftest(load_list='test.list', list_only=True)
 
1967
        self.assertEqual(test_id_line, stream.getvalue())
 
1968
 
 
1969
    def test_load_unknown(self):
 
1970
        # Provide a list with one test - this test.
 
1971
        # And generate a list of the tests in  the suite.
 
1972
        err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
 
1973
            load_list='missing file name', list_only=True)
 
1974
 
 
1975
 
 
1976
class TestRunBzr(tests.TestCase):
 
1977
 
 
1978
    out = ''
 
1979
    err = ''
 
1980
 
 
1981
    def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
 
1982
                         working_dir=None):
 
1983
        """Override _run_bzr_core to test how it is invoked by run_bzr.
 
1984
 
 
1985
        Attempts to run bzr from inside this class don't actually run it.
 
1986
 
 
1987
        We test how run_bzr actually invokes bzr in another location.  Here we
 
1988
        only need to test that it passes the right parameters to run_bzr.
 
1989
        """
 
1990
        self.argv = list(argv)
 
1991
        self.retcode = retcode
 
1992
        self.encoding = encoding
 
1993
        self.stdin = stdin
 
1994
        self.working_dir = working_dir
 
1995
        return self.retcode, self.out, self.err
 
1996
 
 
1997
    def test_run_bzr_error(self):
 
1998
        self.out = "It sure does!\n"
 
1999
        out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
 
2000
        self.assertEqual(['rocks'], self.argv)
 
2001
        self.assertEqual(34, self.retcode)
 
2002
        self.assertEqual('It sure does!\n', out)
 
2003
        self.assertEquals(out, self.out)
 
2004
        self.assertEqual('', err)
 
2005
        self.assertEquals(err, self.err)
 
2006
 
 
2007
    def test_run_bzr_error_regexes(self):
 
2008
        self.out = ''
 
2009
        self.err = "bzr: ERROR: foobarbaz is not versioned"
 
2010
        out, err = self.run_bzr_error(
 
2011
            ["bzr: ERROR: foobarbaz is not versioned"],
 
2012
            ['file-id', 'foobarbaz'])
 
2013
 
 
2014
    def test_encoding(self):
 
2015
        """Test that run_bzr passes encoding to _run_bzr_core"""
 
2016
        self.run_bzr('foo bar')
 
2017
        self.assertEqual(None, self.encoding)
 
2018
        self.assertEqual(['foo', 'bar'], self.argv)
 
2019
 
 
2020
        self.run_bzr('foo bar', encoding='baz')
 
2021
        self.assertEqual('baz', self.encoding)
 
2022
        self.assertEqual(['foo', 'bar'], self.argv)
 
2023
 
 
2024
    def test_retcode(self):
 
2025
        """Test that run_bzr passes retcode to _run_bzr_core"""
 
2026
        # Default is retcode == 0
 
2027
        self.run_bzr('foo bar')
 
2028
        self.assertEqual(0, self.retcode)
 
2029
        self.assertEqual(['foo', 'bar'], self.argv)
 
2030
 
 
2031
        self.run_bzr('foo bar', retcode=1)
 
2032
        self.assertEqual(1, self.retcode)
 
2033
        self.assertEqual(['foo', 'bar'], self.argv)
 
2034
 
 
2035
        self.run_bzr('foo bar', retcode=None)
 
2036
        self.assertEqual(None, self.retcode)
 
2037
        self.assertEqual(['foo', 'bar'], self.argv)
 
2038
 
 
2039
        self.run_bzr(['foo', 'bar'], retcode=3)
 
2040
        self.assertEqual(3, self.retcode)
 
2041
        self.assertEqual(['foo', 'bar'], self.argv)
 
2042
 
 
2043
    def test_stdin(self):
 
2044
        # test that the stdin keyword to run_bzr is passed through to
 
2045
        # _run_bzr_core as-is. We do this by overriding
 
2046
        # _run_bzr_core in this class, and then calling run_bzr,
 
2047
        # which is a convenience function for _run_bzr_core, so
 
2048
        # should invoke it.
 
2049
        self.run_bzr('foo bar', stdin='gam')
 
2050
        self.assertEqual('gam', self.stdin)
 
2051
        self.assertEqual(['foo', 'bar'], self.argv)
 
2052
 
 
2053
        self.run_bzr('foo bar', stdin='zippy')
 
2054
        self.assertEqual('zippy', self.stdin)
 
2055
        self.assertEqual(['foo', 'bar'], self.argv)
 
2056
 
 
2057
    def test_working_dir(self):
 
2058
        """Test that run_bzr passes working_dir to _run_bzr_core"""
 
2059
        self.run_bzr('foo bar')
 
2060
        self.assertEqual(None, self.working_dir)
 
2061
        self.assertEqual(['foo', 'bar'], self.argv)
 
2062
 
 
2063
        self.run_bzr('foo bar', working_dir='baz')
 
2064
        self.assertEqual('baz', self.working_dir)
 
2065
        self.assertEqual(['foo', 'bar'], self.argv)
 
2066
 
 
2067
    def test_reject_extra_keyword_arguments(self):
 
2068
        self.assertRaises(TypeError, self.run_bzr, "foo bar",
 
2069
                          error_regex=['error message'])
 
2070
 
 
2071
 
 
2072
class TestRunBzrCaptured(tests.TestCaseWithTransport):
 
2073
    # Does IO when testing the working_dir parameter.
 
2074
 
 
2075
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
 
2076
                         a_callable=None, *args, **kwargs):
 
2077
        self.stdin = stdin
 
2078
        self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
 
2079
        self.factory = bzrlib.ui.ui_factory
 
2080
        self.working_dir = osutils.getcwd()
 
2081
        stdout.write('foo\n')
 
2082
        stderr.write('bar\n')
 
2083
        return 0
 
2084
 
 
2085
    def test_stdin(self):
 
2086
        # test that the stdin keyword to _run_bzr_core is passed through to
 
2087
        # apply_redirected as a StringIO. We do this by overriding
 
2088
        # apply_redirected in this class, and then calling _run_bzr_core,
 
2089
        # which calls apply_redirected.
 
2090
        self.run_bzr(['foo', 'bar'], stdin='gam')
 
2091
        self.assertEqual('gam', self.stdin.read())
 
2092
        self.assertTrue(self.stdin is self.factory_stdin)
 
2093
        self.run_bzr(['foo', 'bar'], stdin='zippy')
 
2094
        self.assertEqual('zippy', self.stdin.read())
 
2095
        self.assertTrue(self.stdin is self.factory_stdin)
 
2096
 
 
2097
    def test_ui_factory(self):
 
2098
        # each invocation of self.run_bzr should get its
 
2099
        # own UI factory, which is an instance of TestUIFactory,
 
2100
        # with stdin, stdout and stderr attached to the stdin,
 
2101
        # stdout and stderr of the invoked run_bzr
 
2102
        current_factory = bzrlib.ui.ui_factory
 
2103
        self.run_bzr(['foo'])
 
2104
        self.failIf(current_factory is self.factory)
 
2105
        self.assertNotEqual(sys.stdout, self.factory.stdout)
 
2106
        self.assertNotEqual(sys.stderr, self.factory.stderr)
 
2107
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
 
2108
        self.assertEqual('bar\n', self.factory.stderr.getvalue())
 
2109
        self.assertIsInstance(self.factory, tests.TestUIFactory)
 
2110
 
 
2111
    def test_working_dir(self):
 
2112
        self.build_tree(['one/', 'two/'])
 
2113
        cwd = osutils.getcwd()
 
2114
 
 
2115
        # Default is to work in the current directory
 
2116
        self.run_bzr(['foo', 'bar'])
 
2117
        self.assertEqual(cwd, self.working_dir)
 
2118
 
 
2119
        self.run_bzr(['foo', 'bar'], working_dir=None)
 
2120
        self.assertEqual(cwd, self.working_dir)
 
2121
 
 
2122
        # The function should be run in the alternative directory
 
2123
        # but afterwards the current working dir shouldn't be changed
 
2124
        self.run_bzr(['foo', 'bar'], working_dir='one')
 
2125
        self.assertNotEqual(cwd, self.working_dir)
 
2126
        self.assertEndsWith(self.working_dir, 'one')
 
2127
        self.assertEqual(cwd, osutils.getcwd())
 
2128
 
 
2129
        self.run_bzr(['foo', 'bar'], working_dir='two')
 
2130
        self.assertNotEqual(cwd, self.working_dir)
 
2131
        self.assertEndsWith(self.working_dir, 'two')
 
2132
        self.assertEqual(cwd, osutils.getcwd())
 
2133
 
 
2134
 
 
2135
class StubProcess(object):
 
2136
    """A stub process for testing run_bzr_subprocess."""
 
2137
    
 
2138
    def __init__(self, out="", err="", retcode=0):
 
2139
        self.out = out
 
2140
        self.err = err
 
2141
        self.returncode = retcode
 
2142
 
 
2143
    def communicate(self):
 
2144
        return self.out, self.err
 
2145
 
 
2146
 
 
2147
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
 
2148
    """Base class for tests testing how we might run bzr."""
 
2149
 
 
2150
    def setUp(self):
 
2151
        tests.TestCaseWithTransport.setUp(self)
 
2152
        self.subprocess_calls = []
 
2153
 
 
2154
    def start_bzr_subprocess(self, process_args, env_changes=None,
 
2155
                             skip_if_plan_to_signal=False,
 
2156
                             working_dir=None,
 
2157
                             allow_plugins=False):
 
2158
        """capture what run_bzr_subprocess tries to do."""
 
2159
        self.subprocess_calls.append({'process_args':process_args,
 
2160
            'env_changes':env_changes,
 
2161
            'skip_if_plan_to_signal':skip_if_plan_to_signal,
 
2162
            'working_dir':working_dir, 'allow_plugins':allow_plugins})
 
2163
        return self.next_subprocess
 
2164
 
 
2165
 
 
2166
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
 
2167
 
 
2168
    def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
 
2169
        """Run run_bzr_subprocess with args and kwargs using a stubbed process.
 
2170
 
 
2171
        Inside TestRunBzrSubprocessCommands we use a stub start_bzr_subprocess
 
2172
        that will return static results. This assertion method populates those
 
2173
        results and also checks the arguments run_bzr_subprocess generates.
 
2174
        """
 
2175
        self.next_subprocess = process
 
2176
        try:
 
2177
            result = self.run_bzr_subprocess(*args, **kwargs)
 
2178
        except:
 
2179
            self.next_subprocess = None
 
2180
            for key, expected in expected_args.iteritems():
 
2181
                self.assertEqual(expected, self.subprocess_calls[-1][key])
 
2182
            raise
1835
2183
        else:
1836
 
            self.fail('Assertion not raised')
 
2184
            self.next_subprocess = None
 
2185
            for key, expected in expected_args.iteritems():
 
2186
                self.assertEqual(expected, self.subprocess_calls[-1][key])
 
2187
            return result
 
2188
 
 
2189
    def test_run_bzr_subprocess(self):
 
2190
        """The run_bzr_helper_external command behaves nicely."""
 
2191
        self.assertRunBzrSubprocess({'process_args':['--version']},
 
2192
            StubProcess(), '--version')
 
2193
        self.assertRunBzrSubprocess({'process_args':['--version']},
 
2194
            StubProcess(), ['--version'])
 
2195
        # retcode=None disables retcode checking
 
2196
        result = self.assertRunBzrSubprocess({},
 
2197
            StubProcess(retcode=3), '--version', retcode=None)
 
2198
        result = self.assertRunBzrSubprocess({},
 
2199
            StubProcess(out="is free software"), '--version')
 
2200
        self.assertContainsRe(result[0], 'is free software')
 
2201
        # Running a subcommand that is missing errors
 
2202
        self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
 
2203
            {'process_args':['--versionn']}, StubProcess(retcode=3),
 
2204
            '--versionn')
 
2205
        # Unless it is told to expect the error from the subprocess
 
2206
        result = self.assertRunBzrSubprocess({},
 
2207
            StubProcess(retcode=3), '--versionn', retcode=3)
 
2208
        # Or to ignore retcode checking
 
2209
        result = self.assertRunBzrSubprocess({},
 
2210
            StubProcess(err="unknown command", retcode=3), '--versionn',
 
2211
            retcode=None)
 
2212
        self.assertContainsRe(result[1], 'unknown command')
 
2213
 
 
2214
    def test_env_change_passes_through(self):
 
2215
        self.assertRunBzrSubprocess(
 
2216
            {'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
 
2217
            StubProcess(), '',
 
2218
            env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
 
2219
 
 
2220
    def test_no_working_dir_passed_as_None(self):
 
2221
        self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
 
2222
 
 
2223
    def test_no_working_dir_passed_through(self):
 
2224
        self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
 
2225
            working_dir='dir')
 
2226
 
 
2227
    def test_run_bzr_subprocess_no_plugins(self):
 
2228
        self.assertRunBzrSubprocess({'allow_plugins': False},
 
2229
            StubProcess(), '')
 
2230
 
 
2231
    def test_allow_plugins(self):
 
2232
        self.assertRunBzrSubprocess({'allow_plugins': True},
 
2233
            StubProcess(), '', allow_plugins=True)
 
2234
 
 
2235
 
 
2236
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
 
2237
 
 
2238
    def test_finish_bzr_subprocess_with_error(self):
 
2239
        """finish_bzr_subprocess allows specification of the desired exit code.
 
2240
        """
 
2241
        process = StubProcess(err="unknown command", retcode=3)
 
2242
        result = self.finish_bzr_subprocess(process, retcode=3)
 
2243
        self.assertEqual('', result[0])
 
2244
        self.assertContainsRe(result[1], 'unknown command')
 
2245
 
 
2246
    def test_finish_bzr_subprocess_ignoring_retcode(self):
 
2247
        """finish_bzr_subprocess allows the exit code to be ignored."""
 
2248
        process = StubProcess(err="unknown command", retcode=3)
 
2249
        result = self.finish_bzr_subprocess(process, retcode=None)
 
2250
        self.assertEqual('', result[0])
 
2251
        self.assertContainsRe(result[1], 'unknown command')
 
2252
 
 
2253
    def test_finish_subprocess_with_unexpected_retcode(self):
 
2254
        """finish_bzr_subprocess raises self.failureException if the retcode is
 
2255
        not the expected one.
 
2256
        """
 
2257
        process = StubProcess(err="unknown command", retcode=3)
 
2258
        self.assertRaises(self.failureException, self.finish_bzr_subprocess,
 
2259
                          process)
 
2260
 
 
2261
 
 
2262
class _DontSpawnProcess(Exception):
 
2263
    """A simple exception which just allows us to skip unnecessary steps"""
 
2264
 
 
2265
 
 
2266
class TestStartBzrSubProcess(tests.TestCase):
 
2267
 
 
2268
    def check_popen_state(self):
 
2269
        """Replace to make assertions when popen is called."""
 
2270
 
 
2271
    def _popen(self, *args, **kwargs):
 
2272
        """Record the command that is run, so that we can ensure it is correct"""
 
2273
        self.check_popen_state()
 
2274
        self._popen_args = args
 
2275
        self._popen_kwargs = kwargs
 
2276
        raise _DontSpawnProcess()
 
2277
 
 
2278
    def test_run_bzr_subprocess_no_plugins(self):
 
2279
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
 
2280
        command = self._popen_args[0]
 
2281
        self.assertEqual(sys.executable, command[0])
 
2282
        self.assertEqual(self.get_bzr_path(), command[1])
 
2283
        self.assertEqual(['--no-plugins'], command[2:])
 
2284
 
 
2285
    def test_allow_plugins(self):
 
2286
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2287
            allow_plugins=True)
 
2288
        command = self._popen_args[0]
 
2289
        self.assertEqual([], command[2:])
 
2290
 
 
2291
    def test_set_env(self):
 
2292
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2293
        # set in the child
 
2294
        def check_environment():
 
2295
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
 
2296
        self.check_popen_state = check_environment
 
2297
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2298
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
 
2299
        # not set in theparent
 
2300
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2301
 
 
2302
    def test_run_bzr_subprocess_env_del(self):
 
2303
        """run_bzr_subprocess can remove environment variables too."""
 
2304
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
 
2305
        def check_environment():
 
2306
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
 
2307
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
 
2308
        self.check_popen_state = check_environment
 
2309
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2310
            env_changes={'EXISTANT_ENV_VAR':None})
 
2311
        # Still set in parent
 
2312
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
 
2313
        del os.environ['EXISTANT_ENV_VAR']
 
2314
 
 
2315
    def test_env_del_missing(self):
 
2316
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
 
2317
        def check_environment():
 
2318
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
 
2319
        self.check_popen_state = check_environment
 
2320
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2321
            env_changes={'NON_EXISTANT_ENV_VAR':None})
 
2322
 
 
2323
    def test_working_dir(self):
 
2324
        """Test that we can specify the working dir for the child"""
 
2325
        orig_getcwd = osutils.getcwd
 
2326
        orig_chdir = os.chdir
 
2327
        chdirs = []
 
2328
        def chdir(path):
 
2329
            chdirs.append(path)
 
2330
        os.chdir = chdir
 
2331
        try:
 
2332
            def getcwd():
 
2333
                return 'current'
 
2334
            osutils.getcwd = getcwd
 
2335
            try:
 
2336
                self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
 
2337
                    working_dir='foo')
 
2338
            finally:
 
2339
                osutils.getcwd = orig_getcwd
 
2340
        finally:
 
2341
            os.chdir = orig_chdir
 
2342
        self.assertEqual(['foo', 'current'], chdirs)
 
2343
 
 
2344
 
 
2345
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
 
2346
    """Tests that really need to do things with an external bzr."""
 
2347
 
 
2348
    def test_start_and_stop_bzr_subprocess_send_signal(self):
 
2349
        """finish_bzr_subprocess raises self.failureException if the retcode is
 
2350
        not the expected one.
 
2351
        """
 
2352
        self.disable_missing_extensions_warning()
 
2353
        process = self.start_bzr_subprocess(['wait-until-signalled'],
 
2354
                                            skip_if_plan_to_signal=True)
 
2355
        self.assertEqual('running\n', process.stdout.readline())
 
2356
        result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
 
2357
                                            retcode=3)
 
2358
        self.assertEqual('', result[0])
 
2359
        self.assertEqual('bzr: interrupted\n', result[1])
1837
2360
 
1838
2361
 
1839
2362
class TestFeature(tests.TestCase):
1877
2400
        self.assertIs(feature, exception.args[0])
1878
2401
 
1879
2402
 
 
2403
simple_thunk_feature = tests._CompatabilityThunkFeature(
 
2404
    deprecated_in((2, 1, 0)),
 
2405
    'bzrlib.tests.test_selftest',
 
2406
    'simple_thunk_feature','UnicodeFilename',
 
2407
    replacement_module='bzrlib.tests'
 
2408
    )
 
2409
 
 
2410
class Test_CompatibilityFeature(tests.TestCase):
 
2411
 
 
2412
    def test_does_thunk(self):
 
2413
        res = self.callDeprecated(
 
2414
            ['bzrlib.tests.test_selftest.simple_thunk_feature was deprecated'
 
2415
             ' in version 2.1.0. Use bzrlib.tests.UnicodeFilename instead.'],
 
2416
            simple_thunk_feature.available)
 
2417
        self.assertEqual(tests.UnicodeFilename.available(), res)
 
2418
 
 
2419
 
 
2420
class TestModuleAvailableFeature(tests.TestCase):
 
2421
 
 
2422
    def test_available_module(self):
 
2423
        feature = tests.ModuleAvailableFeature('bzrlib.tests')
 
2424
        self.assertEqual('bzrlib.tests', feature.module_name)
 
2425
        self.assertEqual('bzrlib.tests', str(feature))
 
2426
        self.assertTrue(feature.available())
 
2427
        self.assertIs(tests, feature.module)
 
2428
 
 
2429
    def test_unavailable_module(self):
 
2430
        feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
 
2431
        self.assertEqual('bzrlib.no_such_module_exists', str(feature))
 
2432
        self.assertFalse(feature.available())
 
2433
        self.assertIs(None, feature.module)
 
2434
 
 
2435
 
1880
2436
class TestSelftestFiltering(tests.TestCase):
1881
2437
 
1882
2438
    def setUp(self):
1883
2439
        tests.TestCase.setUp(self)
1884
2440
        self.suite = TestUtil.TestSuite()
1885
2441
        self.loader = TestUtil.TestLoader()
1886
 
        self.suite.addTest(self.loader.loadTestsFromModuleNames([
1887
 
            'bzrlib.tests.test_selftest']))
 
2442
        self.suite.addTest(self.loader.loadTestsFromModule(
 
2443
            sys.modules['bzrlib.tests.test_selftest']))
1888
2444
        self.all_names = _test_ids(self.suite)
1889
2445
 
1890
2446
    def test_condition_id_re(self):
2061
2617
        # the test framework
2062
2618
        self.assertEquals('always fails', str(e))
2063
2619
        # check that there's no traceback in the test log
2064
 
        self.assertNotContainsRe(self._get_log(keep_log_file=True),
2065
 
            r'Traceback')
 
2620
        self.assertNotContainsRe(self.get_log(), r'Traceback')
2066
2621
 
2067
2622
    def test_run_bzr_user_error_caught(self):
2068
2623
        # Running bzr in blackbox mode, normal/expected/user errors should be
2069
2624
        # caught in the regular way and turned into an error message plus exit
2070
2625
        # code.
2071
 
        out, err = self.run_bzr(["log", "/nonexistantpath"], retcode=3)
 
2626
        transport_server = MemoryServer()
 
2627
        transport_server.start_server()
 
2628
        self.addCleanup(transport_server.stop_server)
 
2629
        url = transport_server.get_url()
 
2630
        self.permit_url(url)
 
2631
        out, err = self.run_bzr(["log", "%s/nonexistantpath" % url], retcode=3)
2072
2632
        self.assertEqual(out, '')
2073
2633
        self.assertContainsRe(err,
2074
2634
            'bzr: ERROR: Not a branch: ".*nonexistantpath/".\n')
2200
2760
 
2201
2761
class TestTestSuite(tests.TestCase):
2202
2762
 
 
2763
    def test__test_suite_testmod_names(self):
 
2764
        # Test that a plausible list of test module names are returned
 
2765
        # by _test_suite_testmod_names.
 
2766
        test_list = tests._test_suite_testmod_names()
 
2767
        self.assertSubset([
 
2768
            'bzrlib.tests.blackbox',
 
2769
            'bzrlib.tests.per_transport',
 
2770
            'bzrlib.tests.test_selftest',
 
2771
            ],
 
2772
            test_list)
 
2773
 
 
2774
    def test__test_suite_modules_to_doctest(self):
 
2775
        # Test that a plausible list of modules to doctest is returned
 
2776
        # by _test_suite_modules_to_doctest.
 
2777
        test_list = tests._test_suite_modules_to_doctest()
 
2778
        self.assertSubset([
 
2779
            'bzrlib.timestamp',
 
2780
            ],
 
2781
            test_list)
 
2782
 
2203
2783
    def test_test_suite(self):
2204
 
        # This test is slow, so we do a single test with one test in each
2205
 
        # category
2206
 
        test_list = [
 
2784
        # test_suite() loads the entire test suite to operate. To avoid this
 
2785
        # overhead, and yet still be confident that things are happening,
 
2786
        # we temporarily replace two functions used by test_suite with 
 
2787
        # test doubles that supply a few sample tests to load, and check they
 
2788
        # are loaded.
 
2789
        calls = []
 
2790
        def testmod_names():
 
2791
            calls.append("testmod_names")
 
2792
            return [
 
2793
                'bzrlib.tests.blackbox.test_branch',
 
2794
                'bzrlib.tests.per_transport',
 
2795
                'bzrlib.tests.test_selftest',
 
2796
                ]
 
2797
        self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
 
2798
        def doctests():
 
2799
            calls.append("modules_to_doctest")
 
2800
            return ['bzrlib.timestamp']
 
2801
        self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
 
2802
        expected_test_list = [
2207
2803
            # testmod_names
2208
2804
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
2209
2805
            ('bzrlib.tests.per_transport.TransportTests'
2210
 
             '.test_abspath(LocalURLServer)'),
 
2806
             '.test_abspath(LocalTransport,LocalURLServer)'),
2211
2807
            'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
2212
2808
            # modules_to_doctest
2213
2809
            'bzrlib.timestamp.format_highres_date',
2214
2810
            # plugins can't be tested that way since selftest may be run with
2215
2811
            # --no-plugins
2216
2812
            ]
2217
 
        suite = tests.test_suite(test_list)
2218
 
        self.assertEquals(test_list, _test_ids(suite))
 
2813
        suite = tests.test_suite()
 
2814
        self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
 
2815
            set(calls))
 
2816
        self.assertSubset(expected_test_list, _test_ids(suite))
2219
2817
 
2220
2818
    def test_test_suite_list_and_start(self):
 
2819
        # We cannot test this at the same time as the main load, because we want
 
2820
        # to know that starting_with == None works. So a second load is
 
2821
        # incurred - note that the starting_with parameter causes a partial load
 
2822
        # rather than a full load so this test should be pretty quick.
2221
2823
        test_list = ['bzrlib.tests.test_selftest.TestTestSuite.test_test_suite']
2222
2824
        suite = tests.test_suite(test_list,
2223
2825
                                 ['bzrlib.tests.test_selftest.TestTestSuite'])
2268
2870
    def test_load_tests(self):
2269
2871
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
2270
2872
        loader = self._create_loader(test_list)
2271
 
 
2272
2873
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2273
2874
        self.assertEquals(test_list, _test_ids(suite))
2274
2875
 
2275
2876
    def test_exclude_tests(self):
2276
2877
        test_list = ['bogus']
2277
2878
        loader = self._create_loader(test_list)
2278
 
 
2279
2879
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2280
2880
        self.assertEquals([], _test_ids(suite))
2281
2881
 
2326
2926
        tpr.register('bar', 'bbb.aaa.rrr')
2327
2927
        tpr.register('bar', 'bBB.aAA.rRR')
2328
2928
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
2329
 
        self.assertContainsRe(self._get_log(keep_log_file=True),
2330
 
                              r'.*bar.*bbb.aaa.rrr.*bBB.aAA.rRR')
 
2929
        self.assertThat(self.get_log(),
 
2930
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
2331
2931
 
2332
2932
    def test_get_unknown_prefix(self):
2333
2933
        tpr = self._get_registry()
2368
2968
                return tests.ExtendedTestResult(self.stream, self.descriptions,
2369
2969
                                                self.verbosity)
2370
2970
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2371
 
        self.assertEqual(calls, [suite])
2372
 
 
2373
 
    def test_done(self):
2374
 
        """run_suite should call result.done()"""
2375
 
        self.calls = 0
2376
 
        def one_more_call(): self.calls += 1
2377
 
        def test_function():
2378
 
            pass
2379
 
        test = unittest.FunctionTestCase(test_function)
2380
 
        class InstrumentedTestResult(tests.ExtendedTestResult):
2381
 
            def done(self): one_more_call()
2382
 
        class MyRunner(tests.TextTestRunner):
2383
 
            def run(self, test):
2384
 
                return InstrumentedTestResult(self.stream, self.descriptions,
2385
 
                                              self.verbosity)
2386
 
        tests.run_suite(test, runner_class=MyRunner, stream=StringIO())
2387
 
        self.assertEquals(1, self.calls)
 
2971
        self.assertLength(1, calls)