~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-09-01 08:02:42 UTC
  • mfrom: (5390.3.3 faster-revert-593560)
  • Revision ID: pqm@pqm.ubuntu.com-20100901080242-esg62ody4frwmy66
(spiv) Avoid repeatedly calling self.target.all_file_ids() in
 InterTree.iter_changes. (Andrew Bennetts)

Show diffs side-by-side

added added

removed removed

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