← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-09-18 02:51:19 UTC
  • mfrom: (4691.2.5 integration)
  • Revision ID: pqm@pqm.ubuntu.com-20090918025119-eqkrmtj09t2hqs2t
(robertc) Add test suite enforcement of BzrDir objects outside the
        test area. (Robert Collins)

Show diffs side-by-side

added added

removed removed

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