/brz/remove-bazaar

To get this branch, use:
bzr branch http://gegoxaren.bato24.eu/bzr/brz/remove-bazaar

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Robert Collins
  • Date: 2009-09-17 11:54:41 UTC
  • mto: This revision was merged to the branch mainline in revision 4700.
  • Revision ID: robertc@robertcollins.net-20090917115441-2ug57z6eyrnb6zim
Add stronger test isolation by interception BzrDir.open and checking the thing being opened is known to the test suite.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/__init__.py
90
90
    deprecated_passed,
91
91
    )
92
92
import bzrlib.trace
93
 
from bzrlib.transport import get_transport
 
93
from bzrlib.transport import chroot, get_transport
94
94
import bzrlib.transport
95
95
from bzrlib.transport.local import LocalURLServer
96
96
from bzrlib.transport.memory import MemoryServer
819
819
        self._cleanups = []
820
820
        self._bzr_test_setUp_run = False
821
821
        self._bzr_test_tearDown_run = False
 
822
        self._directory_isolation = True
822
823
 
823
824
    def setUp(self):
824
825
        unittest.TestCase.setUp(self)
829
830
        self._benchcalls = []
830
831
        self._benchtime = None
831
832
        self._clear_hooks()
 
833
        self._track_transports()
832
834
        self._track_locks()
833
835
        self._clear_debug_flags()
834
836
        TestCase._active_threads = threading.activeCount()
875
877
        # this hook should always be installed
876
878
        request._install_hook()
877
879
 
 
880
    def disable_directory_isolation(self):
 
881
        """Turn off directory isolation checks."""
 
882
        self._directory_isolation = False
 
883
 
 
884
    def enable_directory_isolation(self):
 
885
        """Enable directory isolation checks."""
 
886
        self._directory_isolation = True
 
887
 
878
888
    def _silenceUI(self):
879
889
        """Turn off UI for duration of test"""
880
890
        # by default the UI is off; tests can turn it on if they want it.
935
945
    def _lock_broken(self, result):
936
946
        self._lock_actions.append(('broken', result))
937
947
 
 
948
    def permit_dir(self, name):
 
949
        """Permit a directory to be used by this test. See permit_url."""
 
950
        name_transport = get_transport(name)
 
951
        self.permit_url(name)
 
952
        self.permit_url(name_transport.base)
 
953
 
 
954
    def permit_url(self, url):
 
955
        """Declare that url is an ok url to use in this test.
 
956
        
 
957
        Do this for memory transports, temporary test directory etc.
 
958
        
 
959
        Do not do this for the current working directory, /tmp, or any other
 
960
        preexisting non isolated url.
 
961
        """
 
962
        if not url.endswith('/'):
 
963
            url += '/'
 
964
        self._bzr_selftest_roots.append(url)
 
965
 
 
966
    def permit_source_tree_branch_repo(self):
 
967
        """Permit the source tree bzr is running from to be opened.
 
968
 
 
969
        Some code such as bzrlib.version attempts to read from the bzr branch
 
970
        that bzr is executing from (if any). This method permits that directory
 
971
        to be used in the test suite.
 
972
        """
 
973
        path = self.get_source_path()
 
974
        self.disable_directory_isolation()
 
975
        try:
 
976
            try:
 
977
                tree = workingtree.WorkingTree.open(path)
 
978
            except errors.NoWorkingTree:
 
979
                return
 
980
        finally:
 
981
            self.enable_directory_isolation()
 
982
        self.permit_url(tree.bzrdir.root_transport.base)
 
983
        self.permit_url(tree.branch.bzrdir.root_transport.base)
 
984
        self.permit_url(tree.branch.repository.bzrdir.root_transport.base)
 
985
 
 
986
    def _preopen_isolate_transport(self, transport):
 
987
        """Check that all transport openings are done in the test work area."""
 
988
        if isinstance(transport, chroot.ChrootTransport):
 
989
            # Unwrap chrooted transports
 
990
            url = transport.server.backing_transport.clone(
 
991
                transport._safe_relpath('.')).base
 
992
        else:
 
993
            url = transport.base
 
994
        # ReadonlySmartTCPServer_for_testing decorates the backing transport
 
995
        # urls it is given by prepending readonly+. This is appropriate as the
 
996
        # client shouldn't know that the server is readonly (or not readonly).
 
997
        # We could register all servers twice, with readonly+ prepending, but
 
998
        # that makes for a long list; this is about the same but easier to
 
999
        # read.
 
1000
        if url.startswith('readonly+'):
 
1001
            url = url[len('readonly+'):]
 
1002
        self._preopen_isolate_url(url)
 
1003
 
 
1004
    def _preopen_isolate_url(self, url):
 
1005
        if not self._directory_isolation:
 
1006
            return
 
1007
        # This prevents all transports, including e.g. sftp ones backed on disk
 
1008
        # from working unless they are explicitly granted permission. We then
 
1009
        # depend on the code that sets up test transports to check that they are
 
1010
        # appropriately isolated and enable their use by calling
 
1011
        # self.permit_transport()
 
1012
        if not osutils.is_inside_any(self._bzr_selftest_roots, url):
 
1013
            raise errors.BzrError("Attempt to escape test isolation: %r %r"
 
1014
                % (url, self._bzr_selftest_roots))
 
1015
 
938
1016
    def start_server(self, transport_server, backing_server=None):
939
1017
        """Start transport_server for this test.
940
1018
 
946
1024
        else:
947
1025
            transport_server.setUp(backing_server)
948
1026
        self.addCleanup(transport_server.tearDown)
 
1027
        # Obtain a real transport because if the server supplies a password, it
 
1028
        # will be hidden from the base on the client side.
 
1029
        t = get_transport(transport_server.get_url())
 
1030
        # Some transport servers effectively chroot the backing transport;
 
1031
        # others like SFTPServer don't - users of the transport can walk up the
 
1032
        # transport to read the entire backing transport. This wouldn't matter
 
1033
        # except that the workdir tests are given - and that they expect the
 
1034
        # server's url to point at - is one directory under the safety net. So
 
1035
        # Branch operations into the transport will attempt to walk up one
 
1036
        # directory. Chrooting all servers would avoid this but also mean that
 
1037
        # we wouldn't be testing directly against non-root urls. Alternatively
 
1038
        # getting the test framework to start the server with a backing server
 
1039
        # at the actual safety net directory would work too, but this then
 
1040
        # means that the self.get_url/self.get_transport methods would need
 
1041
        # to transform all their results. On balance its cleaner to handle it
 
1042
        # here, and permit a higher url when we have one of these transports.
 
1043
        if t.base.endswith('/work/'):
 
1044
            # we have safety net/test root/work
 
1045
            t = t.clone('../..')
 
1046
        elif isinstance(transport_server, server.SmartTCPServer_for_testing):
 
1047
            # The smart server adds a path similar to work, which is traversed
 
1048
            # up from by the client. But the server is chrooted - the actual
 
1049
            # backing transport is not escaped from, and VFS requests to the
 
1050
            # root will error (because they try to escape the chroot).
 
1051
            t2 = t.clone('..')
 
1052
            while t2.base != t.base:
 
1053
                t = t2
 
1054
                t2 = t.clone('..')
 
1055
        self.permit_url(t.base)
 
1056
 
 
1057
    def _track_transports(self):
 
1058
        """Install checks for transport usage."""
 
1059
        # TestCase has no safe place it can write to.
 
1060
        self._bzr_selftest_roots = []
 
1061
        # Currently the easiest way to be sure that nothing is going on is to
 
1062
        # hook into bzr dir opening. This leaves a small window of error for
 
1063
        # transport tests, but they are well known, and we can improve on this
 
1064
        # step.
 
1065
        bzrdir.BzrDir.hooks.install_named_hook("pre_open",
 
1066
            self._preopen_isolate_transport, "Check bzr directories are safe.")
949
1067
 
950
1068
    def _ndiff_strings(self, a, b):
951
1069
        """Return ndiff between two strings containing lines.
1869
1987
        """
1870
1988
        return Popen(*args, **kwargs)
1871
1989
 
 
1990
    def get_source_path(self):
 
1991
        """Return the path of the directory containing bzrlib."""
 
1992
        return os.path.dirname(os.path.dirname(bzrlib.__file__))
 
1993
 
1872
1994
    def get_bzr_path(self):
1873
1995
        """Return the path of the 'bzr' executable for this test suite."""
1874
 
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
 
1996
        bzr_path = self.get_source_path()+'/bzr'
1875
1997
        if not os.path.isfile(bzr_path):
1876
1998
            # We are probably installed. Assume sys.argv is the right file
1877
1999
            bzr_path = sys.argv[0]
2203
2325
        propagating. This method ensures than a test did not leaked.
2204
2326
        """
2205
2327
        root = TestCaseWithMemoryTransport.TEST_ROOT
 
2328
        self.permit_url(get_transport(root).base)
2206
2329
        wt = workingtree.WorkingTree.open(root)
2207
2330
        last_rev = wt.last_revision()
2208
2331
        if last_rev != 'null:':
2225
2348
            # specifically told when all tests are finished.  This will do.
2226
2349
            atexit.register(_rmtree_temp_dir, root)
2227
2350
 
 
2351
        self.permit_dir(TestCaseWithMemoryTransport.TEST_ROOT)
2228
2352
        self.addCleanup(self._check_safety_net)
2229
2353
 
2230
2354
    def makeAndChdirToTestDir(self):
2238
2362
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2239
2363
        self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
2240
2364
        self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
 
2365
        self.permit_dir(self.test_dir)
2241
2366
 
2242
2367
    def make_branch(self, relpath, format=None):
2243
2368
        """Create a branch on the transport at relpath."""
2375
2500
            if os.path.exists(name):
2376
2501
                name = name_prefix + '_' + str(i)
2377
2502
            else:
2378
 
                os.mkdir(name)
 
2503
                # now create test and home directories within this dir
 
2504
                self.test_base_dir = name
 
2505
                self.addCleanup(self.deleteTestDir)
 
2506
                os.mkdir(self.test_base_dir)
2379
2507
                break
2380
 
        # now create test and home directories within this dir
2381
 
        self.test_base_dir = name
 
2508
        self.permit_dir(self.test_base_dir)
 
2509
        # 'sprouting' and 'init' of a branch both walk up the tree to find
 
2510
        # stacking policy to honour; create a bzr dir with an unshared
 
2511
        # repository (but not a branch - our code would be trying to escape
 
2512
        # then!) to stop them, and permit it to be read.
 
2513
        # control = bzrdir.BzrDir.create(self.test_base_dir)
 
2514
        # control.create_repository()
2382
2515
        self.test_home_dir = self.test_base_dir + '/home'
2383
2516
        os.mkdir(self.test_home_dir)
2384
2517
        self.test_dir = self.test_base_dir + '/work'
2390
2523
            f.write(self.id())
2391
2524
        finally:
2392
2525
            f.close()
2393
 
        self.addCleanup(self.deleteTestDir)
2394
2526
 
2395
2527
    def deleteTestDir(self):
2396
2528
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)