/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: John Arbash Meinel
  • Date: 2009-09-18 15:08:09 UTC
  • mfrom: (4704 +trunk)
  • mto: This revision was merged to the branch mainline in revision 4705.
  • Revision ID: john@arbash-meinel.com-20090918150809-t4kow9vdc8mosvkw
Bring in bzr.dev 4704 resolve NEWS

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 get_transport, pathfilter
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
        while isinstance(transport, pathfilter.PathFilteringTransport):
 
987
            # Unwrap pathfiltered transports
 
988
            transport = transport.server.backing_transport.clone(
 
989
                transport._filter('.'))
 
990
        url = transport.base
 
991
        # ReadonlySmartTCPServer_for_testing decorates the backing transport
 
992
        # urls it is given by prepending readonly+. This is appropriate as the
 
993
        # client shouldn't know that the server is readonly (or not readonly).
 
994
        # We could register all servers twice, with readonly+ prepending, but
 
995
        # that makes for a long list; this is about the same but easier to
 
996
        # read.
 
997
        if url.startswith('readonly+'):
 
998
            url = url[len('readonly+'):]
 
999
        self._preopen_isolate_url(url)
 
1000
 
 
1001
    def _preopen_isolate_url(self, url):
 
1002
        if not self._directory_isolation:
 
1003
            return
 
1004
        if self._directory_isolation == 'record':
 
1005
            self._bzr_selftest_roots.append(url)
 
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
 
 
1016
    def record_directory_isolation(self):
 
1017
        """Gather accessed directories to permit later access.
 
1018
        
 
1019
        This is used for tests that access the branch bzr is running from.
 
1020
        """
 
1021
        self._directory_isolation = "record"
 
1022
 
939
1023
    def start_server(self, transport_server, backing_server=None):
940
1024
        """Start transport_server for this test.
941
1025
 
947
1031
        else:
948
1032
            transport_server.setUp(backing_server)
949
1033
        self.addCleanup(transport_server.tearDown)
 
1034
        # Obtain a real transport because if the server supplies a password, it
 
1035
        # will be hidden from the base on the client side.
 
1036
        t = get_transport(transport_server.get_url())
 
1037
        # Some transport servers effectively chroot the backing transport;
 
1038
        # others like SFTPServer don't - users of the transport can walk up the
 
1039
        # transport to read the entire backing transport. This wouldn't matter
 
1040
        # except that the workdir tests are given - and that they expect the
 
1041
        # server's url to point at - is one directory under the safety net. So
 
1042
        # Branch operations into the transport will attempt to walk up one
 
1043
        # directory. Chrooting all servers would avoid this but also mean that
 
1044
        # we wouldn't be testing directly against non-root urls. Alternatively
 
1045
        # getting the test framework to start the server with a backing server
 
1046
        # at the actual safety net directory would work too, but this then
 
1047
        # means that the self.get_url/self.get_transport methods would need
 
1048
        # to transform all their results. On balance its cleaner to handle it
 
1049
        # here, and permit a higher url when we have one of these transports.
 
1050
        if t.base.endswith('/work/'):
 
1051
            # we have safety net/test root/work
 
1052
            t = t.clone('../..')
 
1053
        elif isinstance(transport_server, server.SmartTCPServer_for_testing):
 
1054
            # The smart server adds a path similar to work, which is traversed
 
1055
            # up from by the client. But the server is chrooted - the actual
 
1056
            # backing transport is not escaped from, and VFS requests to the
 
1057
            # root will error (because they try to escape the chroot).
 
1058
            t2 = t.clone('..')
 
1059
            while t2.base != t.base:
 
1060
                t = t2
 
1061
                t2 = t.clone('..')
 
1062
        self.permit_url(t.base)
 
1063
 
 
1064
    def _track_transports(self):
 
1065
        """Install checks for transport usage."""
 
1066
        # TestCase has no safe place it can write to.
 
1067
        self._bzr_selftest_roots = []
 
1068
        # Currently the easiest way to be sure that nothing is going on is to
 
1069
        # hook into bzr dir opening. This leaves a small window of error for
 
1070
        # transport tests, but they are well known, and we can improve on this
 
1071
        # step.
 
1072
        bzrdir.BzrDir.hooks.install_named_hook("pre_open",
 
1073
            self._preopen_isolate_transport, "Check bzr directories are safe.")
950
1074
 
951
1075
    def _ndiff_strings(self, a, b):
952
1076
        """Return ndiff between two strings containing lines.
1678
1802
        if retcode is not None:
1679
1803
            self.assertEquals(retcode, result,
1680
1804
                              message='Unexpected return code')
1681
 
        return out, err
 
1805
        return result, out, err
1682
1806
 
1683
1807
    def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
1684
1808
                working_dir=None, error_regexes=[], output_encoding=None):
1713
1837
        :keyword error_regexes: A list of expected error messages.  If
1714
1838
            specified they must be seen in the error output of the command.
1715
1839
        """
1716
 
        out, err = self._run_bzr_autosplit(
 
1840
        retcode, out, err = self._run_bzr_autosplit(
1717
1841
            args=args,
1718
1842
            retcode=retcode,
1719
1843
            encoding=encoding,
1870
1994
        """
1871
1995
        return Popen(*args, **kwargs)
1872
1996
 
 
1997
    def get_source_path(self):
 
1998
        """Return the path of the directory containing bzrlib."""
 
1999
        return os.path.dirname(os.path.dirname(bzrlib.__file__))
 
2000
 
1873
2001
    def get_bzr_path(self):
1874
2002
        """Return the path of the 'bzr' executable for this test suite."""
1875
 
        bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
 
2003
        bzr_path = self.get_source_path()+'/bzr'
1876
2004
        if not os.path.isfile(bzr_path):
1877
2005
            # We are probably installed. Assume sys.argv is the right file
1878
2006
            bzr_path = sys.argv[0]
2204
2332
        propagating. This method ensures than a test did not leaked.
2205
2333
        """
2206
2334
        root = TestCaseWithMemoryTransport.TEST_ROOT
 
2335
        self.permit_url(get_transport(root).base)
2207
2336
        wt = workingtree.WorkingTree.open(root)
2208
2337
        last_rev = wt.last_revision()
2209
2338
        if last_rev != 'null:':
2226
2355
            # specifically told when all tests are finished.  This will do.
2227
2356
            atexit.register(_rmtree_temp_dir, root)
2228
2357
 
 
2358
        self.permit_dir(TestCaseWithMemoryTransport.TEST_ROOT)
2229
2359
        self.addCleanup(self._check_safety_net)
2230
2360
 
2231
2361
    def makeAndChdirToTestDir(self):
2239
2369
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
2240
2370
        self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
2241
2371
        self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
 
2372
        self.permit_dir(self.test_dir)
2242
2373
 
2243
2374
    def make_branch(self, relpath, format=None):
2244
2375
        """Create a branch on the transport at relpath."""
2376
2507
            if os.path.exists(name):
2377
2508
                name = name_prefix + '_' + str(i)
2378
2509
            else:
2379
 
                os.mkdir(name)
 
2510
                # now create test and home directories within this dir
 
2511
                self.test_base_dir = name
 
2512
                self.addCleanup(self.deleteTestDir)
 
2513
                os.mkdir(self.test_base_dir)
2380
2514
                break
2381
 
        # now create test and home directories within this dir
2382
 
        self.test_base_dir = name
 
2515
        self.permit_dir(self.test_base_dir)
 
2516
        # 'sprouting' and 'init' of a branch both walk up the tree to find
 
2517
        # stacking policy to honour; create a bzr dir with an unshared
 
2518
        # repository (but not a branch - our code would be trying to escape
 
2519
        # then!) to stop them, and permit it to be read.
 
2520
        # control = bzrdir.BzrDir.create(self.test_base_dir)
 
2521
        # control.create_repository()
2383
2522
        self.test_home_dir = self.test_base_dir + '/home'
2384
2523
        os.mkdir(self.test_home_dir)
2385
2524
        self.test_dir = self.test_base_dir + '/work'
2391
2530
            f.write(self.id())
2392
2531
        finally:
2393
2532
            f.close()
2394
 
        self.addCleanup(self.deleteTestDir)
2395
2533
 
2396
2534
    def deleteTestDir(self):
2397
2535
        os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
3634
3772
        'bzrlib.tests.test_rio',
3635
3773
        'bzrlib.tests.test_rules',
3636
3774
        'bzrlib.tests.test_sampler',
 
3775
        'bzrlib.tests.test_script',
3637
3776
        'bzrlib.tests.test_selftest',
3638
3777
        'bzrlib.tests.test_serializer',
3639
3778
        'bzrlib.tests.test_setup',