~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: Andrew Bennetts
  • Date: 2009-07-20 04:26:55 UTC
  • mto: This revision was merged to the branch mainline in revision 4552.
  • Revision ID: andrew.bennetts@canonical.com-20090720042655-ggzsg8uurf8ljbfh
Add test for read_mergeable_from_transport raising NotABundle when TooManyRedirections happens.

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
 
    stub_sftp,
66
52
    test_lsprof,
67
 
    test_server,
68
53
    test_sftp_transport,
69
54
    TestUtil,
70
55
    )
71
56
from bzrlib.trace import note
72
 
from bzrlib.transport import memory
 
57
from bzrlib.transport.memory import MemoryServer, MemoryTransport
73
58
from bzrlib.version import _get_bzr_source_tree
74
59
 
75
60
 
89
74
                          TestUtil._load_module_by_name,
90
75
                          'bzrlib.no-name-yet')
91
76
 
92
 
 
93
77
class MetaTestLog(tests.TestCase):
94
78
 
95
79
    def test_logging(self):
96
80
        """Test logs are captured when a test fails."""
97
81
        self.log('a test message')
98
 
        details = self.getDetails()
99
 
        log = details['log']
100
 
        self.assertThat(log.content_type, Equals(ContentType(
101
 
            "text", "plain", {"charset": "utf8"})))
102
 
        self.assertThat(u"".join(log.iter_text()), Equals(self.get_log()))
103
 
        self.assertThat(self.get_log(),
104
 
            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')
105
85
 
106
86
 
107
87
class TestUnicodeFilename(tests.TestCase):
144
124
        self.assertEqual(sample_permutation,
145
125
                         get_transport_test_permutations(MockModule()))
146
126
 
147
 
    def test_scenarios_include_all_modules(self):
 
127
    def test_scenarios_invlude_all_modules(self):
148
128
        # this checks that the scenario generator returns as many permutations
149
129
        # as there are in all the registered transport modules - we assume if
150
130
        # this matches its probably doing the right thing especially in
235
215
        from bzrlib.tests.per_repository import formats_to_scenarios
236
216
        formats = [("(c)", remote.RemoteRepositoryFormat()),
237
217
                   ("(d)", repository.format_registry.get(
238
 
                    'Bazaar repository format 2a (needs bzr 1.16 or later)\n'))]
 
218
                        'Bazaar pack repository format 1 (needs bzr 0.92)\n'))]
239
219
        no_vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
240
220
            None)
241
221
        vfs_scenarios = formats_to_scenarios(formats, "server", "readonly",
242
222
            vfs_transport_factory="vfs")
243
223
        # no_vfs generate scenarios without vfs_transport_factory
244
 
        expected = [
 
224
        self.assertEqual([
245
225
            ('RemoteRepositoryFormat(c)',
246
226
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
247
227
              'repository_format': remote.RemoteRepositoryFormat(),
248
228
              'transport_readonly_server': 'readonly',
249
229
              'transport_server': 'server'}),
250
 
            ('RepositoryFormat2a(d)',
 
230
            ('RepositoryFormatKnitPack1(d)',
251
231
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
252
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
232
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
253
233
              'transport_readonly_server': 'readonly',
254
 
              'transport_server': 'server'})]
255
 
        self.assertEqual(expected, no_vfs_scenarios)
 
234
              'transport_server': 'server'})],
 
235
            no_vfs_scenarios)
256
236
        self.assertEqual([
257
237
            ('RemoteRepositoryFormat(c)',
258
238
             {'bzrdir_format': remote.RemoteBzrDirFormat(),
260
240
              'transport_readonly_server': 'readonly',
261
241
              'transport_server': 'server',
262
242
              'vfs_transport_factory': 'vfs'}),
263
 
            ('RepositoryFormat2a(d)',
 
243
            ('RepositoryFormatKnitPack1(d)',
264
244
             {'bzrdir_format': bzrdir.BzrDirMetaFormat1(),
265
 
              'repository_format': groupcompress_repo.RepositoryFormat2a(),
 
245
              'repository_format': pack_repo.RepositoryFormatKnitPack1(),
266
246
              'transport_readonly_server': 'readonly',
267
247
              'transport_server': 'server',
268
248
              'vfs_transport_factory': 'vfs'})],
313
293
        from bzrlib.tests.per_interrepository import make_scenarios
314
294
        server1 = "a"
315
295
        server2 = "b"
316
 
        formats = [("C0", "C1", "C2"), ("D0", "D1", "D2")]
 
296
        formats = [(str, "C1", "C2"), (int, "D1", "D2")]
317
297
        scenarios = make_scenarios(server1, server2, formats)
318
298
        self.assertEqual([
319
 
            ('C0,str,str',
320
 
             {'repository_format': 'C1',
 
299
            ('str,str,str',
 
300
             {'interrepo_class': str,
 
301
              'repository_format': 'C1',
321
302
              'repository_format_to': 'C2',
322
303
              'transport_readonly_server': 'b',
323
304
              'transport_server': 'a'}),
324
 
            ('D0,str,str',
325
 
             {'repository_format': 'D1',
 
305
            ('int,str,str',
 
306
             {'interrepo_class': int,
 
307
              'repository_format': 'D1',
326
308
              'repository_format_to': 'D2',
327
309
              'transport_readonly_server': 'b',
328
310
              'transport_server': 'a'})],
506
488
        self.assertEqualStat(real, fake)
507
489
 
508
490
    def test_assertEqualStat_notequal(self):
509
 
        self.build_tree(["foo", "longname"])
 
491
        self.build_tree(["foo", "bar"])
510
492
        self.assertRaises(AssertionError, self.assertEqualStat,
511
 
            os.lstat("foo"), os.lstat("longname"))
 
493
            os.lstat("foo"), os.lstat("bar"))
512
494
 
513
495
 
514
496
class TestTestCaseWithMemoryTransport(tests.TestCaseWithMemoryTransport):
532
514
        cwd = osutils.getcwd()
533
515
        self.assertIsSameRealPath(self.test_dir, cwd)
534
516
 
535
 
    def test_BZR_HOME_and_HOME_are_bytestrings(self):
536
 
        """The $BZR_HOME and $HOME environment variables should not be unicode.
537
 
 
538
 
        See https://bugs.launchpad.net/bzr/+bug/464174
539
 
        """
540
 
        self.assertIsInstance(os.environ['BZR_HOME'], str)
541
 
        self.assertIsInstance(os.environ['HOME'], str)
542
 
 
543
517
    def test_make_branch_and_memory_tree(self):
544
518
        """In TestCaseWithMemoryTransport we should not make the branch on disk.
545
519
 
601
575
                         self.get_transport().get_bytes(
602
576
                            'dir/.bzr/repository/format'))
603
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
 
604
591
    def test_dangling_locks_cause_failures(self):
605
592
        class TestDanglingLock(tests.TestCaseWithMemoryTransport):
606
593
            def test_function(self):
610
597
                l.attempt_lock()
611
598
        test = TestDanglingLock('test_function')
612
599
        result = test.run()
613
 
        if self._lock_check_thorough:
614
 
            self.assertEqual(1, len(result.errors))
615
 
        else:
616
 
            # When _lock_check_thorough is disabled, then we don't trigger a
617
 
            # failure
618
 
            self.assertEqual(0, len(result.errors))
 
600
        self.assertEqual(1, len(result.errors))
619
601
 
620
602
 
621
603
class TestTestCaseWithTransport(tests.TestCaseWithTransport):
623
605
 
624
606
    def test_get_readonly_url_none(self):
625
607
        from bzrlib.transport import get_transport
 
608
        from bzrlib.transport.memory import MemoryServer
626
609
        from bzrlib.transport.readonly import ReadonlyTransportDecorator
627
 
        self.vfs_transport_factory = memory.MemoryServer
 
610
        self.vfs_transport_factory = MemoryServer
628
611
        self.transport_readonly_server = None
629
612
        # calling get_readonly_transport() constructs a decorator on the url
630
613
        # for the server
639
622
    def test_get_readonly_url_http(self):
640
623
        from bzrlib.tests.http_server import HttpServer
641
624
        from bzrlib.transport import get_transport
 
625
        from bzrlib.transport.local import LocalURLServer
642
626
        from bzrlib.transport.http import HttpTransportBase
643
 
        self.transport_server = test_server.LocalURLServer
 
627
        self.transport_server = LocalURLServer
644
628
        self.transport_readonly_server = HttpServer
645
629
        # calling get_readonly_transport() gives us a HTTP server instance.
646
630
        url = self.get_readonly_url()
677
661
 
678
662
    def setUp(self):
679
663
        super(TestTestCaseTransports, self).setUp()
680
 
        self.vfs_transport_factory = memory.MemoryServer
 
664
        self.vfs_transport_factory = MemoryServer
681
665
 
682
666
    def test_make_bzrdir_preserves_transport(self):
683
667
        t = self.get_transport()
684
668
        result_bzrdir = self.make_bzrdir('subdir')
685
669
        self.assertIsInstance(result_bzrdir.transport,
686
 
                              memory.MemoryTransport)
 
670
                              MemoryTransport)
687
671
        # should not be on disk, should only be in memory
688
672
        self.failIfExists('subdir')
689
673
 
697
681
        self.assertEqual(url, t.clone('..').base)
698
682
 
699
683
 
700
 
class TestProfileResult(tests.TestCase):
701
 
 
702
 
    def test_profiles_tests(self):
703
 
        self.requireFeature(test_lsprof.LSProfFeature)
704
 
        terminal = testtools.tests.helpers.ExtendedTestResult()
705
 
        result = tests.ProfileResult(terminal)
706
 
        class Sample(tests.TestCase):
707
 
            def a(self):
708
 
                self.sample_function()
709
 
            def sample_function(self):
710
 
                pass
711
 
        test = Sample("a")
712
 
        test.run(result)
713
 
        case = terminal._events[0][1]
714
 
        self.assertLength(1, case._benchcalls)
715
 
        # We must be able to unpack it as the test reporting code wants
716
 
        (_, _, _), stats = case._benchcalls[0]
717
 
        self.assertTrue(callable(stats.pprint))
 
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))
718
705
 
719
706
 
720
707
class TestTestResult(tests.TestCase):
724
711
                descriptions=0,
725
712
                verbosity=1,
726
713
                )
727
 
        capture = testtools.tests.helpers.ExtendedTestResult()
728
 
        test_case.run(MultiTestResult(result, capture))
729
 
        run_case = capture._events[0][1]
730
 
        timed_string = result._testTimeString(run_case)
 
714
        test_case.run(result)
 
715
        timed_string = result._testTimeString(test_case)
731
716
        self.assertContainsRe(timed_string, expected_re)
732
717
 
733
718
    def test_test_reporting(self):
738
723
                self.time(time.sleep, 0.003)
739
724
        self.check_timing(ShortDelayTestCase('test_short_delay'),
740
725
                          r"^ +[0-9]+ms$")
741
 
        # if a benchmark time is given, we now show just that time followed by
742
 
        # a star
 
726
        # if a benchmark time is given, we want a x of y style result.
743
727
        self.check_timing(ShortDelayTestCase('test_short_benchmark'),
744
 
                          r"^ +[0-9]+ms\*$")
 
728
                          r"^ +[0-9]+ms/ +[0-9]+ms$")
745
729
 
746
730
    def test_unittest_reporting_unittest_class(self):
747
731
        # getting the time from a non-bzrlib test works ok
751
735
        self.check_timing(ShortDelayTestCase('test_short_delay'),
752
736
                          r"^ +[0-9]+ms$")
753
737
 
754
 
    def _patch_get_bzr_source_tree(self):
755
 
        # Reading from the actual source tree breaks isolation, but we don't
756
 
        # want to assume that thats *all* that would happen.
757
 
        self.overrideAttr(bzrlib.version, '_get_bzr_source_tree', lambda: None)
758
 
 
759
738
    def test_assigned_benchmark_file_stores_date(self):
760
 
        self._patch_get_bzr_source_tree()
761
739
        output = StringIO()
762
740
        result = bzrlib.tests.TextTestResult(self._log_file,
763
741
                                        descriptions=0,
771
749
        self.assertContainsRe(output_string, "--date [0-9.]+")
772
750
 
773
751
    def test_benchhistory_records_test_times(self):
774
 
        self._patch_get_bzr_source_tree()
775
752
        result_stream = StringIO()
776
753
        result = bzrlib.tests.TextTestResult(
777
754
            self._log_file,
839
816
    def test_known_failure(self):
840
817
        """A KnownFailure being raised should trigger several result actions."""
841
818
        class InstrumentedTestResult(tests.ExtendedTestResult):
842
 
            def stopTestRun(self): pass
 
819
            def done(self): pass
843
820
            def startTests(self): pass
844
821
            def report_test_start(self, test): pass
845
 
            def report_known_failure(self, test, err=None, details=None):
846
 
                self._call = test, 'known failure'
 
822
            def report_known_failure(self, test, err):
 
823
                self._call = test, err
847
824
        result = InstrumentedTestResult(None, None, None, None)
848
 
        class Test(tests.TestCase):
849
 
            def test_function(self):
850
 
                raise tests.KnownFailure('failed!')
851
 
        test = Test("test_function")
 
825
        def test_function():
 
826
            raise tests.KnownFailure('failed!')
 
827
        test = unittest.FunctionTestCase(test_function)
852
828
        test.run(result)
853
829
        # it should invoke 'report_known_failure'.
854
830
        self.assertEqual(2, len(result._call))
855
 
        self.assertEqual(test.id(), result._call[0].id())
856
 
        self.assertEqual('known failure', result._call[1])
 
831
        self.assertEqual(test, result._call[0])
 
832
        self.assertEqual(tests.KnownFailure, result._call[1][0])
 
833
        self.assertIsInstance(result._call[1][1], tests.KnownFailure)
857
834
        # we dont introspec the traceback, if the rest is ok, it would be
858
835
        # exceptional for it not to be.
859
836
        # it should update the known_failure_count on the object.
884
861
        self.assertEqual(lines[1], '    foo')
885
862
        self.assertEqual(2, len(lines))
886
863
 
 
864
    def test_text_report_known_failure(self):
 
865
        # text test output formatting
 
866
        pb = MockProgress()
 
867
        result = bzrlib.tests.TextTestResult(
 
868
            StringIO(),
 
869
            descriptions=0,
 
870
            verbosity=1,
 
871
            pb=pb,
 
872
            )
 
873
        test = self.get_passing_test()
 
874
        # this seeds the state to handle reporting the test.
 
875
        result.startTest(test)
 
876
        # the err parameter has the shape:
 
877
        # (class, exception object, traceback)
 
878
        # KnownFailures dont get their tracebacks shown though, so we
 
879
        # can skip that.
 
880
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
 
881
        result.report_known_failure(test, err)
 
882
        self.assertEqual(
 
883
            [
 
884
            ('update', '[1 in 0s] passing_test', None, None),
 
885
            ('note', 'XFAIL: %s\n%s\n', ('passing_test', err[1]))
 
886
            ],
 
887
            pb.calls)
 
888
        # known_failures should be printed in the summary, so if we run a test
 
889
        # after there are some known failures, the update prefix should match
 
890
        # this.
 
891
        result.known_failure_count = 3
 
892
        test.run(result)
 
893
        self.assertEqual(
 
894
            [
 
895
            ('update', '[2 in 0s] passing_test', None, None),
 
896
            ],
 
897
            pb.calls[2:])
 
898
 
887
899
    def get_passing_test(self):
888
900
        """Return a test object that can't be run usefully."""
889
901
        def passing_test():
893
905
    def test_add_not_supported(self):
894
906
        """Test the behaviour of invoking addNotSupported."""
895
907
        class InstrumentedTestResult(tests.ExtendedTestResult):
896
 
            def stopTestRun(self): pass
 
908
            def done(self): pass
897
909
            def startTests(self): pass
898
910
            def report_test_start(self, test): pass
899
911
            def report_unsupported(self, test, feature):
931
943
        result.report_unsupported(test, feature)
932
944
        output = result_stream.getvalue()[prefix:]
933
945
        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.")
 
946
        self.assertEqual(lines, ['NODEP                   0ms',
 
947
                                 "    The feature 'Feature' is not available."])
 
948
 
 
949
    def test_text_report_unsupported(self):
 
950
        # text test output formatting
 
951
        pb = MockProgress()
 
952
        result = bzrlib.tests.TextTestResult(
 
953
            StringIO(),
 
954
            descriptions=0,
 
955
            verbosity=1,
 
956
            pb=pb,
 
957
            )
 
958
        test = self.get_passing_test()
 
959
        feature = tests.Feature()
 
960
        # this seeds the state to handle reporting the test.
 
961
        result.startTest(test)
 
962
        result.report_unsupported(test, feature)
 
963
        # no output on unsupported features
 
964
        self.assertEqual(
 
965
            [('update', '[1 in 0s] passing_test', None, None)
 
966
            ],
 
967
            pb.calls)
 
968
        # the number of missing features should be printed in the progress
 
969
        # summary, so check for that.
 
970
        result.unsupported = {'foo':0, 'bar':0}
 
971
        test.run(result)
 
972
        self.assertEqual(
 
973
            [
 
974
            ('update', '[2 in 0s, 2 missing] passing_test', None, None),
 
975
            ],
 
976
            pb.calls[1:])
938
977
 
939
978
    def test_unavailable_exception(self):
940
979
        """An UnavailableFeature being raised should invoke addNotSupported."""
941
980
        class InstrumentedTestResult(tests.ExtendedTestResult):
942
 
            def stopTestRun(self): pass
 
981
            def done(self): pass
943
982
            def startTests(self): pass
944
983
            def report_test_start(self, test): pass
945
984
            def addNotSupported(self, test, feature):
946
985
                self._call = test, feature
947
986
        result = InstrumentedTestResult(None, None, None, None)
948
987
        feature = tests.Feature()
949
 
        class Test(tests.TestCase):
950
 
            def test_function(self):
951
 
                raise tests.UnavailableFeature(feature)
952
 
        test = Test("test_function")
 
988
        def test_function():
 
989
            raise tests.UnavailableFeature(feature)
 
990
        test = unittest.FunctionTestCase(test_function)
953
991
        test.run(result)
954
992
        # it should invoke 'addNotSupported'.
955
993
        self.assertEqual(2, len(result._call))
956
 
        self.assertEqual(test.id(), result._call[0].id())
 
994
        self.assertEqual(test, result._call[0])
957
995
        self.assertEqual(feature, result._call[1])
958
996
        # and not count as an error
959
997
        self.assertEqual(0, result.error_count)
972
1010
                                             verbosity=1)
973
1011
        test = self.get_passing_test()
974
1012
        err = (tests.KnownFailure, tests.KnownFailure('foo'), None)
975
 
        result.addExpectedFailure(test, err)
 
1013
        result._addKnownFailure(test, err)
976
1014
        self.assertFalse(result.wasStrictlySuccessful())
977
1015
        self.assertEqual(None, result._extractBenchmarkTime(test))
978
1016
 
1023
1061
        because of our use of global state.
1024
1062
        """
1025
1063
        old_root = tests.TestCaseInTempDir.TEST_ROOT
1026
 
        old_leak = tests.TestCase._first_thread_leaker_id
1027
1064
        try:
1028
1065
            tests.TestCaseInTempDir.TEST_ROOT = None
1029
 
            tests.TestCase._first_thread_leaker_id = None
1030
1066
            return testrunner.run(test)
1031
1067
        finally:
1032
1068
            tests.TestCaseInTempDir.TEST_ROOT = old_root
1033
 
            tests.TestCase._first_thread_leaker_id = old_leak
1034
1069
 
1035
1070
    def test_known_failure_failed_run(self):
1036
1071
        # run a test that generates a known failure which should be printed in
1037
1072
        # 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)
 
1073
        def known_failure_test():
 
1074
            raise tests.KnownFailure('failed')
1041
1075
        test = unittest.TestSuite()
1042
 
        test.addTest(Test("known_failure_test"))
 
1076
        test.addTest(unittest.FunctionTestCase(known_failure_test))
1043
1077
        def failing_test():
1044
 
            self.fail('foo')
 
1078
            raise AssertionError('foo')
1045
1079
        test.addTest(unittest.FunctionTestCase(failing_test))
1046
1080
        stream = StringIO()
1047
1081
        runner = tests.TextTestRunner(stream=stream)
1048
1082
        result = self.run_test_runner(runner, test)
1049
1083
        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
 
            )
 
1084
        self.assertEqual([
 
1085
            '',
 
1086
            '======================================================================',
 
1087
            'FAIL: unittest.FunctionTestCase (failing_test)',
 
1088
            '----------------------------------------------------------------------',
 
1089
            'Traceback (most recent call last):',
 
1090
            '    raise AssertionError(\'foo\')',
 
1091
            'AssertionError: foo',
 
1092
            '',
 
1093
            '----------------------------------------------------------------------',
 
1094
            '',
 
1095
            'FAILED (failures=1, known_failure_count=1)'],
 
1096
            lines[3:8] + lines[9:13] + lines[14:])
1064
1097
 
1065
1098
    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")
 
1099
        # run a test that generates a known failure which should be printed in the final output.
 
1100
        def known_failure_test():
 
1101
            raise tests.KnownFailure('failed')
 
1102
        test = unittest.FunctionTestCase(known_failure_test)
1072
1103
        stream = StringIO()
1073
1104
        runner = tests.TextTestRunner(stream=stream)
1074
1105
        result = self.run_test_runner(runner, test)
1079
1110
            '\n'
1080
1111
            'OK \\(known_failures=1\\)\n')
1081
1112
 
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
1113
    def test_skipped_test(self):
1097
1114
        # run a test that is skipped, and check the suite as a whole still
1098
1115
        # succeeds.
1151
1168
 
1152
1169
    def test_not_applicable(self):
1153
1170
        # 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')
 
1171
        def not_applicable_test():
 
1172
            raise tests.TestNotApplicable('this test never runs')
1157
1173
        out = StringIO()
1158
1174
        runner = tests.TextTestRunner(stream=out, verbosity=2)
1159
 
        test = Test("not_applicable_test")
 
1175
        test = unittest.FunctionTestCase(not_applicable_test)
1160
1176
        result = self.run_test_runner(runner, test)
1161
1177
        self._log_file.write(out.getvalue())
1162
1178
        self.assertTrue(result.wasSuccessful())
1166
1182
        self.assertContainsRe(out.getvalue(),
1167
1183
                r'(?m)^    this test never runs')
1168
1184
 
 
1185
    def test_not_applicable_demo(self):
 
1186
        # just so you can see it in the test output
 
1187
        raise tests.TestNotApplicable('this test is just a demonstation')
 
1188
 
1169
1189
    def test_unsupported_features_listed(self):
1170
1190
        """When unsupported features are encountered they are detailed."""
1171
1191
        class Feature1(tests.Feature):
1191
1211
            ],
1192
1212
            lines[-3:])
1193
1213
 
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
1214
    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()
 
1215
        # tests that the running the benchmark produces a history file
 
1216
        # containing a timestamp and the revision id of the bzrlib source which
 
1217
        # was tested.
 
1218
        workingtree = _get_bzr_source_tree()
1208
1219
        test = TestRunner('dummy_test')
1209
1220
        output = StringIO()
1210
1221
        runner = tests.TextTestRunner(stream=self._log_file,
1212
1223
        result = self.run_test_runner(runner, test)
1213
1224
        output_string = output.getvalue()
1214
1225
        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)
 
1226
        if workingtree is not None:
 
1227
            revision_id = workingtree.get_parent_ids()[0]
 
1228
            self.assertEndsWith(output_string.rstrip(), revision_id)
 
1229
 
 
1230
    def assertLogDeleted(self, test):
 
1231
        log = test._get_log()
 
1232
        self.assertEqual("DELETED log file to reduce memory footprint", log)
 
1233
        self.assertEqual('', test._log_contents)
 
1234
        self.assertIs(None, test._log_file_name)
 
1235
 
 
1236
    def test_success_log_deleted(self):
 
1237
        """Successful tests have their log deleted"""
 
1238
 
 
1239
        class LogTester(tests.TestCase):
 
1240
 
 
1241
            def test_success(self):
 
1242
                self.log('this will be removed\n')
 
1243
 
 
1244
        sio = StringIO()
 
1245
        runner = tests.TextTestRunner(stream=sio)
 
1246
        test = LogTester('test_success')
 
1247
        result = self.run_test_runner(runner, test)
 
1248
 
 
1249
        self.assertLogDeleted(test)
 
1250
 
 
1251
    def test_skipped_log_deleted(self):
 
1252
        """Skipped tests have their log deleted"""
 
1253
 
 
1254
        class LogTester(tests.TestCase):
 
1255
 
 
1256
            def test_skipped(self):
 
1257
                self.log('this will be removed\n')
 
1258
                raise tests.TestSkipped()
 
1259
 
 
1260
        sio = StringIO()
 
1261
        runner = tests.TextTestRunner(stream=sio)
 
1262
        test = LogTester('test_skipped')
 
1263
        result = self.run_test_runner(runner, test)
 
1264
 
 
1265
        self.assertLogDeleted(test)
 
1266
 
 
1267
    def test_not_aplicable_log_deleted(self):
 
1268
        """Not applicable tests have their log deleted"""
 
1269
 
 
1270
        class LogTester(tests.TestCase):
 
1271
 
 
1272
            def test_not_applicable(self):
 
1273
                self.log('this will be removed\n')
 
1274
                raise tests.TestNotApplicable()
 
1275
 
 
1276
        sio = StringIO()
 
1277
        runner = tests.TextTestRunner(stream=sio)
 
1278
        test = LogTester('test_not_applicable')
 
1279
        result = self.run_test_runner(runner, test)
 
1280
 
 
1281
        self.assertLogDeleted(test)
 
1282
 
 
1283
    def test_known_failure_log_deleted(self):
 
1284
        """Know failure tests have their log deleted"""
 
1285
 
 
1286
        class LogTester(tests.TestCase):
 
1287
 
 
1288
            def test_known_failure(self):
 
1289
                self.log('this will be removed\n')
 
1290
                raise tests.KnownFailure()
 
1291
 
 
1292
        sio = StringIO()
 
1293
        runner = tests.TextTestRunner(stream=sio)
 
1294
        test = LogTester('test_known_failure')
 
1295
        result = self.run_test_runner(runner, test)
 
1296
 
 
1297
        self.assertLogDeleted(test)
 
1298
 
 
1299
    def test_fail_log_kept(self):
 
1300
        """Failed tests have their log kept"""
 
1301
 
 
1302
        class LogTester(tests.TestCase):
 
1303
 
 
1304
            def test_fail(self):
 
1305
                self.log('this will be kept\n')
 
1306
                self.fail('this test fails')
 
1307
 
 
1308
        sio = StringIO()
 
1309
        runner = tests.TextTestRunner(stream=sio)
 
1310
        test = LogTester('test_fail')
 
1311
        result = self.run_test_runner(runner, test)
 
1312
 
 
1313
        text = sio.getvalue()
 
1314
        self.assertContainsRe(text, 'this will be kept')
 
1315
        self.assertContainsRe(text, 'this test fails')
 
1316
 
 
1317
        log = test._get_log()
 
1318
        self.assertContainsRe(log, 'this will be kept')
 
1319
        self.assertEqual(log, test._log_contents)
 
1320
 
 
1321
    def test_error_log_kept(self):
 
1322
        """Tests with errors have their log kept"""
 
1323
 
 
1324
        class LogTester(tests.TestCase):
 
1325
 
 
1326
            def test_error(self):
 
1327
                self.log('this will be kept\n')
 
1328
                raise ValueError('random exception raised')
 
1329
 
 
1330
        sio = StringIO()
 
1331
        runner = tests.TextTestRunner(stream=sio)
 
1332
        test = LogTester('test_error')
 
1333
        result = self.run_test_runner(runner, test)
 
1334
 
 
1335
        text = sio.getvalue()
 
1336
        self.assertContainsRe(text, 'this will be kept')
 
1337
        self.assertContainsRe(text, 'random exception raised')
 
1338
 
 
1339
        log = test._get_log()
 
1340
        self.assertContainsRe(log, 'this will be kept')
 
1341
        self.assertEqual(log, test._log_contents)
1244
1342
 
1245
1343
 
1246
1344
class SampleTestCase(tests.TestCase):
1251
1349
class _TestException(Exception):
1252
1350
    pass
1253
1351
 
1254
 
 
1255
1352
class TestTestCase(tests.TestCase):
1256
1353
    """Tests that test the core bzrlib TestCase."""
1257
1354
 
1306
1403
        # we could set something and run a test that will check
1307
1404
        # it gets santised, but this is probably sufficient for now:
1308
1405
        # 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)
 
1406
        self.assertEqual(set(), bzrlib.debug.debug_flags)
1313
1407
 
1314
1408
    def change_selftest_debug_flags(self, new_flags):
1315
 
        self.overrideAttr(tests, 'selftest_debug_flags', set(new_flags))
 
1409
        orig_selftest_flags = tests.selftest_debug_flags
 
1410
        self.addCleanup(self._restore_selftest_debug_flags, orig_selftest_flags)
 
1411
        tests.selftest_debug_flags = set(new_flags)
 
1412
 
 
1413
    def _restore_selftest_debug_flags(self, flags):
 
1414
        tests.selftest_debug_flags = flags
1316
1415
 
1317
1416
    def test_allow_debug_flag(self):
1318
1417
        """The -Eallow_debug flag prevents bzrlib.debug.debug_flags from being
1325
1424
                self.flags = set(bzrlib.debug.debug_flags)
1326
1425
        test = TestThatRecordsFlags('test_foo')
1327
1426
        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)
 
1427
        self.assertEqual(set(['a-flag']), self.flags)
1365
1428
 
1366
1429
    def test_debug_flags_restored(self):
1367
1430
        """The bzrlib debug flags should be restored to their original state
1378
1441
        self.assertEqual(set(['original-state']), bzrlib.debug.debug_flags)
1379
1442
 
1380
1443
    def make_test_result(self):
1381
 
        """Get a test result that writes to the test log file."""
1382
1444
        return tests.TextTestResult(self._log_file, descriptions=0, verbosity=1)
1383
1445
 
1384
1446
    def inner_test(self):
1392
1454
        result = self.make_test_result()
1393
1455
        self.inner_test.run(result)
1394
1456
        note("outer finish")
1395
 
        self.addCleanup(osutils.delete_any, self._log_file_name)
1396
1457
 
1397
1458
    def test_trace_nesting(self):
1398
1459
        # this tests that each test case nests its trace facility correctly.
1424
1485
        result = bzrlib.tests.VerboseTestResult(
1425
1486
            unittest._WritelnDecorator(output_stream),
1426
1487
            descriptions=0,
1427
 
            verbosity=2)
 
1488
            verbosity=2,
 
1489
            num_tests=sample_test.countTestCases())
1428
1490
        sample_test.run(result)
1429
1491
        self.assertContainsRe(
1430
1492
            output_stream.getvalue(),
1431
 
            r"\d+ms\*\n$")
 
1493
            r"\d+ms/ +\d+ms\n$")
1432
1494
 
1433
1495
    def test_hooks_sanitised(self):
1434
1496
        """The bzrlib hooks should be sanitised by setUp."""
1458
1520
        self.assertEqual((time.sleep, (0.003,), {}), self._benchcalls[1][0])
1459
1521
        self.assertIsInstance(self._benchcalls[0][1], bzrlib.lsprof.Stats)
1460
1522
        self.assertIsInstance(self._benchcalls[1][1], bzrlib.lsprof.Stats)
1461
 
        del self._benchcalls[:]
1462
1523
 
1463
1524
    def test_knownFailure(self):
1464
1525
        """Self.knownFailure() should raise a KnownFailure exception."""
1465
1526
        self.assertRaises(tests.KnownFailure, self.knownFailure, "A Failure")
1466
1527
 
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
1528
    def test_requireFeature_available(self):
1485
1529
        """self.requireFeature(available) is a no-op."""
1486
1530
        class Available(tests.Feature):
1530
1574
        """Test disabled tests behaviour with support aware results."""
1531
1575
        test = SampleTestCase('_test_pass')
1532
1576
        class DisabledFeature(object):
1533
 
            def __eq__(self, other):
1534
 
                return isinstance(other, DisabledFeature)
1535
1577
            def available(self):
1536
1578
                return False
1537
1579
        the_feature = DisabledFeature()
1548
1590
                self.calls.append(('addNotSupported', test, feature))
1549
1591
        result = InstrumentedTestResult()
1550
1592
        test.run(result)
1551
 
        case = result.calls[0][1]
1552
1593
        self.assertEqual([
1553
 
            ('startTest', case),
1554
 
            ('addNotSupported', case, the_feature),
1555
 
            ('stopTest', case),
 
1594
            ('startTest', test),
 
1595
            ('addNotSupported', test, the_feature),
 
1596
            ('stopTest', test),
1556
1597
            ],
1557
1598
            result.calls)
1558
1599
 
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
1600
    def test_assert_list_raises_on_generator(self):
1568
1601
        def generator_which_will_raise():
1569
1602
            # This will not raise until after the first yield
1619
1652
        self.assertRaises(AssertionError,
1620
1653
            self.assertListRaises, _TestException, success_generator)
1621
1654
 
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
1655
 
1659
1656
# NB: Don't delete this; it's not actually from 0.11!
1660
1657
@deprecated_function(deprecated_in((0, 11, 0)))
1689
1686
    def test_assert_isinstance(self):
1690
1687
        self.assertIsInstance(2, int)
1691
1688
        self.assertIsInstance(u'', basestring)
1692
 
        e = self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1693
 
        self.assertEquals(str(e),
1694
 
            "None is an instance of <type 'NoneType'> rather than <type 'int'>")
 
1689
        self.assertRaises(AssertionError, self.assertIsInstance, None, int)
1695
1690
        self.assertRaises(AssertionError, self.assertIsInstance, 23.3, int)
1696
 
        e = self.assertRaises(AssertionError,
1697
 
            self.assertIsInstance, None, int, "it's just not")
1698
 
        self.assertEquals(str(e),
1699
 
            "None is an instance of <type 'NoneType'> rather than <type 'int'>"
1700
 
            ": it's just not")
1701
1691
 
1702
1692
    def test_assertEndsWith(self):
1703
1693
        self.assertEndsWith('foo', 'oo')
1704
1694
        self.assertRaises(AssertionError, self.assertEndsWith, 'o', 'oo')
1705
1695
 
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
1696
    def test_applyDeprecated_not_deprecated(self):
1722
1697
        sample_object = ApplyDeprecatedHelper()
1723
1698
        # calling an undeprecated callable raises an assertion
1800
1775
        tree = self.make_branch_and_memory_tree('a')
1801
1776
        self.assertIsInstance(tree, bzrlib.memorytree.MemoryTree)
1802
1777
 
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://'))
 
1778
 
 
1779
class TestSFTPMakeBranchAndTree(test_sftp_transport.TestCaseWithSFTPServer):
 
1780
 
 
1781
    def test_make_tree_for_sftp_branch(self):
 
1782
        """Transports backed by local directories create local trees."""
 
1783
 
1809
1784
        tree = self.make_branch_and_tree('t1')
1810
1785
        base = tree.bzrdir.root_transport.base
1811
 
        self.assertStartsWith(base, 'file://')
 
1786
        self.failIf(base.startswith('sftp'),
 
1787
                'base %r is on sftp but should be local' % base)
1812
1788
        self.assertEquals(tree.bzrdir.root_transport,
1813
1789
                tree.branch.bzrdir.root_transport)
1814
1790
        self.assertEquals(tree.bzrdir.root_transport,
1815
1791
                tree.branch.repository.bzrdir.root_transport)
1816
1792
 
1817
1793
 
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):
 
1794
class TestSelftest(tests.TestCase):
1836
1795
    """Tests of bzrlib.tests.selftest."""
1837
1796
 
1838
1797
    def test_selftest_benchmark_parameter_invokes_test_suite__benchmark__(self):
1846
1805
            test_suite_factory=factory)
1847
1806
        self.assertEqual([True], factory_called)
1848
1807
 
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
 
        self.check_transport_set(stub_sftp.SFTPAbsoluteServer)
1951
 
 
1952
 
    def test_transport_memory(self):
1953
 
        self.check_transport_set(memory.MemoryServer)
1954
 
 
1955
 
 
1956
 
class TestSelftestWithIdList(tests.TestCaseInTempDir, SelfTestHelper):
1957
 
    # Does IO: reads test.list
1958
 
 
1959
 
    def test_load_list(self):
1960
 
        # Provide a list with one test - this test.
1961
 
        test_id_line = '%s\n' % self.id()
1962
 
        self.build_tree_contents([('test.list', test_id_line)])
1963
 
        # And generate a list of the tests in  the suite.
1964
 
        stream = self.run_selftest(load_list='test.list', list_only=True)
1965
 
        self.assertEqual(test_id_line, stream.getvalue())
1966
 
 
1967
 
    def test_load_unknown(self):
1968
 
        # Provide a list with one test - this test.
1969
 
        # And generate a list of the tests in  the suite.
1970
 
        err = self.assertRaises(errors.NoSuchFile, self.run_selftest,
1971
 
            load_list='missing file name', list_only=True)
1972
 
 
1973
 
 
1974
 
class TestRunBzr(tests.TestCase):
1975
 
 
1976
 
    out = ''
1977
 
    err = ''
1978
 
 
1979
 
    def _run_bzr_core(self, argv, retcode=0, encoding=None, stdin=None,
1980
 
                         working_dir=None):
1981
 
        """Override _run_bzr_core to test how it is invoked by run_bzr.
1982
 
 
1983
 
        Attempts to run bzr from inside this class don't actually run it.
1984
 
 
1985
 
        We test how run_bzr actually invokes bzr in another location.  Here we
1986
 
        only need to test that it passes the right parameters to run_bzr.
1987
 
        """
1988
 
        self.argv = list(argv)
1989
 
        self.retcode = retcode
1990
 
        self.encoding = encoding
1991
 
        self.stdin = stdin
1992
 
        self.working_dir = working_dir
1993
 
        return self.retcode, self.out, self.err
1994
 
 
1995
 
    def test_run_bzr_error(self):
1996
 
        self.out = "It sure does!\n"
1997
 
        out, err = self.run_bzr_error(['^$'], ['rocks'], retcode=34)
1998
 
        self.assertEqual(['rocks'], self.argv)
1999
 
        self.assertEqual(34, self.retcode)
2000
 
        self.assertEqual('It sure does!\n', out)
2001
 
        self.assertEquals(out, self.out)
2002
 
        self.assertEqual('', err)
2003
 
        self.assertEquals(err, self.err)
2004
 
 
2005
 
    def test_run_bzr_error_regexes(self):
2006
 
        self.out = ''
2007
 
        self.err = "bzr: ERROR: foobarbaz is not versioned"
2008
 
        out, err = self.run_bzr_error(
2009
 
            ["bzr: ERROR: foobarbaz is not versioned"],
2010
 
            ['file-id', 'foobarbaz'])
2011
 
 
2012
 
    def test_encoding(self):
2013
 
        """Test that run_bzr passes encoding to _run_bzr_core"""
2014
 
        self.run_bzr('foo bar')
2015
 
        self.assertEqual(None, self.encoding)
2016
 
        self.assertEqual(['foo', 'bar'], self.argv)
2017
 
 
2018
 
        self.run_bzr('foo bar', encoding='baz')
2019
 
        self.assertEqual('baz', self.encoding)
2020
 
        self.assertEqual(['foo', 'bar'], self.argv)
2021
 
 
2022
 
    def test_retcode(self):
2023
 
        """Test that run_bzr passes retcode to _run_bzr_core"""
2024
 
        # Default is retcode == 0
2025
 
        self.run_bzr('foo bar')
2026
 
        self.assertEqual(0, self.retcode)
2027
 
        self.assertEqual(['foo', 'bar'], self.argv)
2028
 
 
2029
 
        self.run_bzr('foo bar', retcode=1)
2030
 
        self.assertEqual(1, self.retcode)
2031
 
        self.assertEqual(['foo', 'bar'], self.argv)
2032
 
 
2033
 
        self.run_bzr('foo bar', retcode=None)
2034
 
        self.assertEqual(None, self.retcode)
2035
 
        self.assertEqual(['foo', 'bar'], self.argv)
2036
 
 
2037
 
        self.run_bzr(['foo', 'bar'], retcode=3)
2038
 
        self.assertEqual(3, self.retcode)
2039
 
        self.assertEqual(['foo', 'bar'], self.argv)
2040
 
 
2041
 
    def test_stdin(self):
2042
 
        # test that the stdin keyword to run_bzr is passed through to
2043
 
        # _run_bzr_core as-is. We do this by overriding
2044
 
        # _run_bzr_core in this class, and then calling run_bzr,
2045
 
        # which is a convenience function for _run_bzr_core, so
2046
 
        # should invoke it.
2047
 
        self.run_bzr('foo bar', stdin='gam')
2048
 
        self.assertEqual('gam', self.stdin)
2049
 
        self.assertEqual(['foo', 'bar'], self.argv)
2050
 
 
2051
 
        self.run_bzr('foo bar', stdin='zippy')
2052
 
        self.assertEqual('zippy', self.stdin)
2053
 
        self.assertEqual(['foo', 'bar'], self.argv)
2054
 
 
2055
 
    def test_working_dir(self):
2056
 
        """Test that run_bzr passes working_dir to _run_bzr_core"""
2057
 
        self.run_bzr('foo bar')
2058
 
        self.assertEqual(None, self.working_dir)
2059
 
        self.assertEqual(['foo', 'bar'], self.argv)
2060
 
 
2061
 
        self.run_bzr('foo bar', working_dir='baz')
2062
 
        self.assertEqual('baz', self.working_dir)
2063
 
        self.assertEqual(['foo', 'bar'], self.argv)
2064
 
 
2065
 
    def test_reject_extra_keyword_arguments(self):
2066
 
        self.assertRaises(TypeError, self.run_bzr, "foo bar",
2067
 
                          error_regex=['error message'])
2068
 
 
2069
 
 
2070
 
class TestRunBzrCaptured(tests.TestCaseWithTransport):
2071
 
    # Does IO when testing the working_dir parameter.
2072
 
 
2073
 
    def apply_redirected(self, stdin=None, stdout=None, stderr=None,
2074
 
                         a_callable=None, *args, **kwargs):
2075
 
        self.stdin = stdin
2076
 
        self.factory_stdin = getattr(bzrlib.ui.ui_factory, "stdin", None)
2077
 
        self.factory = bzrlib.ui.ui_factory
2078
 
        self.working_dir = osutils.getcwd()
2079
 
        stdout.write('foo\n')
2080
 
        stderr.write('bar\n')
2081
 
        return 0
2082
 
 
2083
 
    def test_stdin(self):
2084
 
        # test that the stdin keyword to _run_bzr_core is passed through to
2085
 
        # apply_redirected as a StringIO. We do this by overriding
2086
 
        # apply_redirected in this class, and then calling _run_bzr_core,
2087
 
        # which calls apply_redirected.
2088
 
        self.run_bzr(['foo', 'bar'], stdin='gam')
2089
 
        self.assertEqual('gam', self.stdin.read())
2090
 
        self.assertTrue(self.stdin is self.factory_stdin)
2091
 
        self.run_bzr(['foo', 'bar'], stdin='zippy')
2092
 
        self.assertEqual('zippy', self.stdin.read())
2093
 
        self.assertTrue(self.stdin is self.factory_stdin)
2094
 
 
2095
 
    def test_ui_factory(self):
2096
 
        # each invocation of self.run_bzr should get its
2097
 
        # own UI factory, which is an instance of TestUIFactory,
2098
 
        # with stdin, stdout and stderr attached to the stdin,
2099
 
        # stdout and stderr of the invoked run_bzr
2100
 
        current_factory = bzrlib.ui.ui_factory
2101
 
        self.run_bzr(['foo'])
2102
 
        self.failIf(current_factory is self.factory)
2103
 
        self.assertNotEqual(sys.stdout, self.factory.stdout)
2104
 
        self.assertNotEqual(sys.stderr, self.factory.stderr)
2105
 
        self.assertEqual('foo\n', self.factory.stdout.getvalue())
2106
 
        self.assertEqual('bar\n', self.factory.stderr.getvalue())
2107
 
        self.assertIsInstance(self.factory, tests.TestUIFactory)
2108
 
 
2109
 
    def test_working_dir(self):
2110
 
        self.build_tree(['one/', 'two/'])
2111
 
        cwd = osutils.getcwd()
2112
 
 
2113
 
        # Default is to work in the current directory
2114
 
        self.run_bzr(['foo', 'bar'])
2115
 
        self.assertEqual(cwd, self.working_dir)
2116
 
 
2117
 
        self.run_bzr(['foo', 'bar'], working_dir=None)
2118
 
        self.assertEqual(cwd, self.working_dir)
2119
 
 
2120
 
        # The function should be run in the alternative directory
2121
 
        # but afterwards the current working dir shouldn't be changed
2122
 
        self.run_bzr(['foo', 'bar'], working_dir='one')
2123
 
        self.assertNotEqual(cwd, self.working_dir)
2124
 
        self.assertEndsWith(self.working_dir, 'one')
2125
 
        self.assertEqual(cwd, osutils.getcwd())
2126
 
 
2127
 
        self.run_bzr(['foo', 'bar'], working_dir='two')
2128
 
        self.assertNotEqual(cwd, self.working_dir)
2129
 
        self.assertEndsWith(self.working_dir, 'two')
2130
 
        self.assertEqual(cwd, osutils.getcwd())
2131
 
 
2132
 
 
2133
 
class StubProcess(object):
2134
 
    """A stub process for testing run_bzr_subprocess."""
2135
 
    
2136
 
    def __init__(self, out="", err="", retcode=0):
2137
 
        self.out = out
2138
 
        self.err = err
2139
 
        self.returncode = retcode
2140
 
 
2141
 
    def communicate(self):
2142
 
        return self.out, self.err
2143
 
 
2144
 
 
2145
 
class TestWithFakedStartBzrSubprocess(tests.TestCaseWithTransport):
2146
 
    """Base class for tests testing how we might run bzr."""
2147
 
 
2148
 
    def setUp(self):
2149
 
        tests.TestCaseWithTransport.setUp(self)
2150
 
        self.subprocess_calls = []
2151
 
 
2152
 
    def start_bzr_subprocess(self, process_args, env_changes=None,
2153
 
                             skip_if_plan_to_signal=False,
2154
 
                             working_dir=None,
2155
 
                             allow_plugins=False):
2156
 
        """capture what run_bzr_subprocess tries to do."""
2157
 
        self.subprocess_calls.append({'process_args':process_args,
2158
 
            'env_changes':env_changes,
2159
 
            'skip_if_plan_to_signal':skip_if_plan_to_signal,
2160
 
            'working_dir':working_dir, 'allow_plugins':allow_plugins})
2161
 
        return self.next_subprocess
2162
 
 
2163
 
 
2164
 
class TestRunBzrSubprocess(TestWithFakedStartBzrSubprocess):
2165
 
 
2166
 
    def assertRunBzrSubprocess(self, expected_args, process, *args, **kwargs):
2167
 
        """Run run_bzr_subprocess with args and kwargs using a stubbed process.
2168
 
 
2169
 
        Inside TestRunBzrSubprocessCommands we use a stub start_bzr_subprocess
2170
 
        that will return static results. This assertion method populates those
2171
 
        results and also checks the arguments run_bzr_subprocess generates.
2172
 
        """
2173
 
        self.next_subprocess = process
2174
 
        try:
2175
 
            result = self.run_bzr_subprocess(*args, **kwargs)
2176
 
        except:
2177
 
            self.next_subprocess = None
2178
 
            for key, expected in expected_args.iteritems():
2179
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2180
 
            raise
 
1808
 
 
1809
class TestKnownFailure(tests.TestCase):
 
1810
 
 
1811
    def test_known_failure(self):
 
1812
        """Check that KnownFailure is defined appropriately."""
 
1813
        # a KnownFailure is an assertion error for compatability with unaware
 
1814
        # runners.
 
1815
        self.assertIsInstance(tests.KnownFailure(""), AssertionError)
 
1816
 
 
1817
    def test_expect_failure(self):
 
1818
        try:
 
1819
            self.expectFailure("Doomed to failure", self.assertTrue, False)
 
1820
        except tests.KnownFailure, e:
 
1821
            self.assertEqual('Doomed to failure', e.args[0])
 
1822
        try:
 
1823
            self.expectFailure("Doomed to failure", self.assertTrue, True)
 
1824
        except AssertionError, e:
 
1825
            self.assertEqual('Unexpected success.  Should have failed:'
 
1826
                             ' Doomed to failure', e.args[0])
2181
1827
        else:
2182
 
            self.next_subprocess = None
2183
 
            for key, expected in expected_args.iteritems():
2184
 
                self.assertEqual(expected, self.subprocess_calls[-1][key])
2185
 
            return result
2186
 
 
2187
 
    def test_run_bzr_subprocess(self):
2188
 
        """The run_bzr_helper_external command behaves nicely."""
2189
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2190
 
            StubProcess(), '--version')
2191
 
        self.assertRunBzrSubprocess({'process_args':['--version']},
2192
 
            StubProcess(), ['--version'])
2193
 
        # retcode=None disables retcode checking
2194
 
        result = self.assertRunBzrSubprocess({},
2195
 
            StubProcess(retcode=3), '--version', retcode=None)
2196
 
        result = self.assertRunBzrSubprocess({},
2197
 
            StubProcess(out="is free software"), '--version')
2198
 
        self.assertContainsRe(result[0], 'is free software')
2199
 
        # Running a subcommand that is missing errors
2200
 
        self.assertRaises(AssertionError, self.assertRunBzrSubprocess,
2201
 
            {'process_args':['--versionn']}, StubProcess(retcode=3),
2202
 
            '--versionn')
2203
 
        # Unless it is told to expect the error from the subprocess
2204
 
        result = self.assertRunBzrSubprocess({},
2205
 
            StubProcess(retcode=3), '--versionn', retcode=3)
2206
 
        # Or to ignore retcode checking
2207
 
        result = self.assertRunBzrSubprocess({},
2208
 
            StubProcess(err="unknown command", retcode=3), '--versionn',
2209
 
            retcode=None)
2210
 
        self.assertContainsRe(result[1], 'unknown command')
2211
 
 
2212
 
    def test_env_change_passes_through(self):
2213
 
        self.assertRunBzrSubprocess(
2214
 
            {'env_changes':{'new':'value', 'changed':'newvalue', 'deleted':None}},
2215
 
            StubProcess(), '',
2216
 
            env_changes={'new':'value', 'changed':'newvalue', 'deleted':None})
2217
 
 
2218
 
    def test_no_working_dir_passed_as_None(self):
2219
 
        self.assertRunBzrSubprocess({'working_dir': None}, StubProcess(), '')
2220
 
 
2221
 
    def test_no_working_dir_passed_through(self):
2222
 
        self.assertRunBzrSubprocess({'working_dir': 'dir'}, StubProcess(), '',
2223
 
            working_dir='dir')
2224
 
 
2225
 
    def test_run_bzr_subprocess_no_plugins(self):
2226
 
        self.assertRunBzrSubprocess({'allow_plugins': False},
2227
 
            StubProcess(), '')
2228
 
 
2229
 
    def test_allow_plugins(self):
2230
 
        self.assertRunBzrSubprocess({'allow_plugins': True},
2231
 
            StubProcess(), '', allow_plugins=True)
2232
 
 
2233
 
 
2234
 
class TestFinishBzrSubprocess(TestWithFakedStartBzrSubprocess):
2235
 
 
2236
 
    def test_finish_bzr_subprocess_with_error(self):
2237
 
        """finish_bzr_subprocess allows specification of the desired exit code.
2238
 
        """
2239
 
        process = StubProcess(err="unknown command", retcode=3)
2240
 
        result = self.finish_bzr_subprocess(process, retcode=3)
2241
 
        self.assertEqual('', result[0])
2242
 
        self.assertContainsRe(result[1], 'unknown command')
2243
 
 
2244
 
    def test_finish_bzr_subprocess_ignoring_retcode(self):
2245
 
        """finish_bzr_subprocess allows the exit code to be ignored."""
2246
 
        process = StubProcess(err="unknown command", retcode=3)
2247
 
        result = self.finish_bzr_subprocess(process, retcode=None)
2248
 
        self.assertEqual('', result[0])
2249
 
        self.assertContainsRe(result[1], 'unknown command')
2250
 
 
2251
 
    def test_finish_subprocess_with_unexpected_retcode(self):
2252
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2253
 
        not the expected one.
2254
 
        """
2255
 
        process = StubProcess(err="unknown command", retcode=3)
2256
 
        self.assertRaises(self.failureException, self.finish_bzr_subprocess,
2257
 
                          process)
2258
 
 
2259
 
 
2260
 
class _DontSpawnProcess(Exception):
2261
 
    """A simple exception which just allows us to skip unnecessary steps"""
2262
 
 
2263
 
 
2264
 
class TestStartBzrSubProcess(tests.TestCase):
2265
 
 
2266
 
    def check_popen_state(self):
2267
 
        """Replace to make assertions when popen is called."""
2268
 
 
2269
 
    def _popen(self, *args, **kwargs):
2270
 
        """Record the command that is run, so that we can ensure it is correct"""
2271
 
        self.check_popen_state()
2272
 
        self._popen_args = args
2273
 
        self._popen_kwargs = kwargs
2274
 
        raise _DontSpawnProcess()
2275
 
 
2276
 
    def test_run_bzr_subprocess_no_plugins(self):
2277
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [])
2278
 
        command = self._popen_args[0]
2279
 
        self.assertEqual(sys.executable, command[0])
2280
 
        self.assertEqual(self.get_bzr_path(), command[1])
2281
 
        self.assertEqual(['--no-plugins'], command[2:])
2282
 
 
2283
 
    def test_allow_plugins(self):
2284
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2285
 
            allow_plugins=True)
2286
 
        command = self._popen_args[0]
2287
 
        self.assertEqual([], command[2:])
2288
 
 
2289
 
    def test_set_env(self):
2290
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2291
 
        # set in the child
2292
 
        def check_environment():
2293
 
            self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2294
 
        self.check_popen_state = check_environment
2295
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2296
 
            env_changes={'EXISTANT_ENV_VAR':'set variable'})
2297
 
        # not set in theparent
2298
 
        self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2299
 
 
2300
 
    def test_run_bzr_subprocess_env_del(self):
2301
 
        """run_bzr_subprocess can remove environment variables too."""
2302
 
        self.failIf('EXISTANT_ENV_VAR' in os.environ)
2303
 
        def check_environment():
2304
 
            self.assertFalse('EXISTANT_ENV_VAR' in os.environ)
2305
 
        os.environ['EXISTANT_ENV_VAR'] = 'set variable'
2306
 
        self.check_popen_state = check_environment
2307
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2308
 
            env_changes={'EXISTANT_ENV_VAR':None})
2309
 
        # Still set in parent
2310
 
        self.assertEqual('set variable', os.environ['EXISTANT_ENV_VAR'])
2311
 
        del os.environ['EXISTANT_ENV_VAR']
2312
 
 
2313
 
    def test_env_del_missing(self):
2314
 
        self.failIf('NON_EXISTANT_ENV_VAR' in os.environ)
2315
 
        def check_environment():
2316
 
            self.assertFalse('NON_EXISTANT_ENV_VAR' in os.environ)
2317
 
        self.check_popen_state = check_environment
2318
 
        self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2319
 
            env_changes={'NON_EXISTANT_ENV_VAR':None})
2320
 
 
2321
 
    def test_working_dir(self):
2322
 
        """Test that we can specify the working dir for the child"""
2323
 
        orig_getcwd = osutils.getcwd
2324
 
        orig_chdir = os.chdir
2325
 
        chdirs = []
2326
 
        def chdir(path):
2327
 
            chdirs.append(path)
2328
 
        os.chdir = chdir
2329
 
        try:
2330
 
            def getcwd():
2331
 
                return 'current'
2332
 
            osutils.getcwd = getcwd
2333
 
            try:
2334
 
                self.assertRaises(_DontSpawnProcess, self.start_bzr_subprocess, [],
2335
 
                    working_dir='foo')
2336
 
            finally:
2337
 
                osutils.getcwd = orig_getcwd
2338
 
        finally:
2339
 
            os.chdir = orig_chdir
2340
 
        self.assertEqual(['foo', 'current'], chdirs)
2341
 
 
2342
 
 
2343
 
class TestActuallyStartBzrSubprocess(tests.TestCaseWithTransport):
2344
 
    """Tests that really need to do things with an external bzr."""
2345
 
 
2346
 
    def test_start_and_stop_bzr_subprocess_send_signal(self):
2347
 
        """finish_bzr_subprocess raises self.failureException if the retcode is
2348
 
        not the expected one.
2349
 
        """
2350
 
        self.disable_missing_extensions_warning()
2351
 
        process = self.start_bzr_subprocess(['wait-until-signalled'],
2352
 
                                            skip_if_plan_to_signal=True)
2353
 
        self.assertEqual('running\n', process.stdout.readline())
2354
 
        result = self.finish_bzr_subprocess(process, send_signal=signal.SIGINT,
2355
 
                                            retcode=3)
2356
 
        self.assertEqual('', result[0])
2357
 
        self.assertEqual('bzr: interrupted\n', result[1])
 
1828
            self.fail('Assertion not raised')
2358
1829
 
2359
1830
 
2360
1831
class TestFeature(tests.TestCase):
2398
1869
        self.assertIs(feature, exception.args[0])
2399
1870
 
2400
1871
 
2401
 
simple_thunk_feature = tests._CompatabilityThunkFeature(
2402
 
    deprecated_in((2, 1, 0)),
2403
 
    'bzrlib.tests.test_selftest',
2404
 
    'simple_thunk_feature','UnicodeFilename',
2405
 
    replacement_module='bzrlib.tests'
2406
 
    )
2407
 
 
2408
 
class Test_CompatibilityFeature(tests.TestCase):
2409
 
 
2410
 
    def test_does_thunk(self):
2411
 
        res = self.callDeprecated(
2412
 
            ['bzrlib.tests.test_selftest.simple_thunk_feature was deprecated'
2413
 
             ' in version 2.1.0. Use bzrlib.tests.UnicodeFilename instead.'],
2414
 
            simple_thunk_feature.available)
2415
 
        self.assertEqual(tests.UnicodeFilename.available(), res)
2416
 
 
2417
 
 
2418
 
class TestModuleAvailableFeature(tests.TestCase):
2419
 
 
2420
 
    def test_available_module(self):
2421
 
        feature = tests.ModuleAvailableFeature('bzrlib.tests')
2422
 
        self.assertEqual('bzrlib.tests', feature.module_name)
2423
 
        self.assertEqual('bzrlib.tests', str(feature))
2424
 
        self.assertTrue(feature.available())
2425
 
        self.assertIs(tests, feature.module)
2426
 
 
2427
 
    def test_unavailable_module(self):
2428
 
        feature = tests.ModuleAvailableFeature('bzrlib.no_such_module_exists')
2429
 
        self.assertEqual('bzrlib.no_such_module_exists', str(feature))
2430
 
        self.assertFalse(feature.available())
2431
 
        self.assertIs(None, feature.module)
2432
 
 
2433
 
 
2434
1872
class TestSelftestFiltering(tests.TestCase):
2435
1873
 
2436
1874
    def setUp(self):
2437
1875
        tests.TestCase.setUp(self)
2438
1876
        self.suite = TestUtil.TestSuite()
2439
1877
        self.loader = TestUtil.TestLoader()
2440
 
        self.suite.addTest(self.loader.loadTestsFromModule(
2441
 
            sys.modules['bzrlib.tests.test_selftest']))
 
1878
        self.suite.addTest(self.loader.loadTestsFromModuleNames([
 
1879
            'bzrlib.tests.test_selftest']))
2442
1880
        self.all_names = _test_ids(self.suite)
2443
1881
 
2444
1882
    def test_condition_id_re(self):
2615
2053
        # the test framework
2616
2054
        self.assertEquals('always fails', str(e))
2617
2055
        # check that there's no traceback in the test log
2618
 
        self.assertNotContainsRe(self.get_log(), r'Traceback')
 
2056
        self.assertNotContainsRe(self._get_log(keep_log_file=True),
 
2057
            r'Traceback')
2619
2058
 
2620
2059
    def test_run_bzr_user_error_caught(self):
2621
2060
        # Running bzr in blackbox mode, normal/expected/user errors should be
2622
2061
        # caught in the regular way and turned into an error message plus exit
2623
2062
        # code.
2624
 
        transport_server = memory.MemoryServer()
2625
 
        transport_server.start_server()
2626
 
        self.addCleanup(transport_server.stop_server)
2627
 
        url = transport_server.get_url()
2628
 
        self.permit_url(url)
2629
 
        out, err = self.run_bzr(["log", "%s/nonexistantpath" % url], retcode=3)
 
2063
        out, err = self.run_bzr(["log", "/nonexistantpath"], retcode=3)
2630
2064
        self.assertEqual(out, '')
2631
2065
        self.assertContainsRe(err,
2632
2066
            'bzr: ERROR: Not a branch: ".*nonexistantpath/".\n')
2758
2192
 
2759
2193
class TestTestSuite(tests.TestCase):
2760
2194
 
2761
 
    def test__test_suite_testmod_names(self):
2762
 
        # Test that a plausible list of test module names are returned
2763
 
        # by _test_suite_testmod_names.
2764
 
        test_list = tests._test_suite_testmod_names()
2765
 
        self.assertSubset([
2766
 
            'bzrlib.tests.blackbox',
2767
 
            'bzrlib.tests.per_transport',
2768
 
            'bzrlib.tests.test_selftest',
2769
 
            ],
2770
 
            test_list)
2771
 
 
2772
 
    def test__test_suite_modules_to_doctest(self):
2773
 
        # Test that a plausible list of modules to doctest is returned
2774
 
        # by _test_suite_modules_to_doctest.
2775
 
        test_list = tests._test_suite_modules_to_doctest()
2776
 
        self.assertSubset([
2777
 
            'bzrlib.timestamp',
2778
 
            ],
2779
 
            test_list)
2780
 
 
2781
2195
    def test_test_suite(self):
2782
 
        # test_suite() loads the entire test suite to operate. To avoid this
2783
 
        # overhead, and yet still be confident that things are happening,
2784
 
        # we temporarily replace two functions used by test_suite with 
2785
 
        # test doubles that supply a few sample tests to load, and check they
2786
 
        # are loaded.
2787
 
        calls = []
2788
 
        def testmod_names():
2789
 
            calls.append("testmod_names")
2790
 
            return [
2791
 
                'bzrlib.tests.blackbox.test_branch',
2792
 
                'bzrlib.tests.per_transport',
2793
 
                'bzrlib.tests.test_selftest',
2794
 
                ]
2795
 
        self.overrideAttr(tests, '_test_suite_testmod_names', testmod_names)
2796
 
        def doctests():
2797
 
            calls.append("modules_to_doctest")
2798
 
            return ['bzrlib.timestamp']
2799
 
        self.overrideAttr(tests, '_test_suite_modules_to_doctest', doctests)
2800
 
        expected_test_list = [
 
2196
        # This test is slow, so we do a single test with one test in each
 
2197
        # category
 
2198
        test_list = [
2801
2199
            # testmod_names
2802
2200
            'bzrlib.tests.blackbox.test_branch.TestBranch.test_branch',
2803
2201
            ('bzrlib.tests.per_transport.TransportTests'
2804
 
             '.test_abspath(LocalTransport,LocalURLServer)'),
 
2202
             '.test_abspath(LocalURLServer)'),
2805
2203
            'bzrlib.tests.test_selftest.TestTestSuite.test_test_suite',
2806
2204
            # modules_to_doctest
2807
2205
            'bzrlib.timestamp.format_highres_date',
2808
2206
            # plugins can't be tested that way since selftest may be run with
2809
2207
            # --no-plugins
2810
2208
            ]
2811
 
        suite = tests.test_suite()
2812
 
        self.assertEqual(set(["testmod_names", "modules_to_doctest"]),
2813
 
            set(calls))
2814
 
        self.assertSubset(expected_test_list, _test_ids(suite))
 
2209
        suite = tests.test_suite(test_list)
 
2210
        self.assertEquals(test_list, _test_ids(suite))
2815
2211
 
2816
2212
    def test_test_suite_list_and_start(self):
2817
 
        # We cannot test this at the same time as the main load, because we want
2818
 
        # to know that starting_with == None works. So a second load is
2819
 
        # incurred - note that the starting_with parameter causes a partial load
2820
 
        # rather than a full load so this test should be pretty quick.
2821
2213
        test_list = ['bzrlib.tests.test_selftest.TestTestSuite.test_test_suite']
2822
2214
        suite = tests.test_suite(test_list,
2823
2215
                                 ['bzrlib.tests.test_selftest.TestTestSuite'])
2868
2260
    def test_load_tests(self):
2869
2261
        test_list = ['bzrlib.tests.test_sampler.DemoTest.test_nothing']
2870
2262
        loader = self._create_loader(test_list)
 
2263
 
2871
2264
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2872
2265
        self.assertEquals(test_list, _test_ids(suite))
2873
2266
 
2874
2267
    def test_exclude_tests(self):
2875
2268
        test_list = ['bogus']
2876
2269
        loader = self._create_loader(test_list)
 
2270
 
2877
2271
        suite = loader.loadTestsFromModuleName('bzrlib.tests.test_sampler')
2878
2272
        self.assertEquals([], _test_ids(suite))
2879
2273
 
2924
2318
        tpr.register('bar', 'bbb.aaa.rrr')
2925
2319
        tpr.register('bar', 'bBB.aAA.rRR')
2926
2320
        self.assertEquals('bbb.aaa.rrr', tpr.get('bar'))
2927
 
        self.assertThat(self.get_log(),
2928
 
            DocTestMatches("...bar...bbb.aaa.rrr...BB.aAA.rRR", ELLIPSIS))
 
2321
        self.assertContainsRe(self._get_log(keep_log_file=True),
 
2322
                              r'.*bar.*bbb.aaa.rrr.*bBB.aAA.rRR')
2929
2323
 
2930
2324
    def test_get_unknown_prefix(self):
2931
2325
        tpr = self._get_registry()
2966
2360
                return tests.ExtendedTestResult(self.stream, self.descriptions,
2967
2361
                                                self.verbosity)
2968
2362
        tests.run_suite(suite, runner_class=MyRunner, stream=StringIO())
2969
 
        self.assertLength(1, calls)
 
2363
        self.assertEqual(calls, [suite])
 
2364
 
 
2365
    def test_done(self):
 
2366
        """run_suite should call result.done()"""
 
2367
        self.calls = 0
 
2368
        def one_more_call(): self.calls += 1
 
2369
        def test_function():
 
2370
            pass
 
2371
        test = unittest.FunctionTestCase(test_function)
 
2372
        class InstrumentedTestResult(tests.ExtendedTestResult):
 
2373
            def done(self): one_more_call()
 
2374
        class MyRunner(tests.TextTestRunner):
 
2375
            def run(self, test):
 
2376
                return InstrumentedTestResult(self.stream, self.descriptions,
 
2377
                                              self.verbosity)
 
2378
        tests.run_suite(test, runner_class=MyRunner, stream=StringIO())
 
2379
        self.assertEquals(1, self.calls)