~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_selftest.py

  • Committer: John Arbash Meinel
  • Date: 2010-07-13 07:44:02 UTC
  • mto: This revision was merged to the branch mainline in revision 5342.
  • Revision ID: john@arbash-meinel.com-20100713074402-wp3oh7cyx76fkvmm
Bump trunk to 2.3-dev1

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