svntest/sandbox.py

   1 #
   2 # sandbox.py :  tools for manipulating a test's working area ("a sandbox")
   3 #
   4 # ====================================================================
   5 #    Licensed to the Apache Software Foundation (ASF) under one
   6 #    or more contributor license agreements.  See the NOTICE file
   7 #    distributed with this work for additional information
   8 #    regarding copyright ownership.  The ASF licenses this file
   9 #    to you under the Apache License, Version 2.0 (the
  10 #    "License"); you may not use this file except in compliance
  11 #    with the License.  You may obtain a copy of the License at
  12 #
  13 #      http://www.apache.org/licenses/LICENSE-2.0
  14 #
  15 #    Unless required by applicable law or agreed to in writing,
  16 #    software distributed under the License is distributed on an
  17 #    "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  18 #    KIND, either express or implied.  See the License for the
  19 #    specific language governing permissions and limitations
  20 #    under the License.
  21 # ====================================================================
  22 #
  23
  24 import os
  25 import shutil
  26 import copy
  27
  28 import svntest
  29
  30
  31 class Sandbox:
  32   """Manages a sandbox (one or more repository/working copy pairs) for
  33   a test to operate within."""
  34
  35   dependents = None
  36
  37   def __init__(self, module, idx):
  38     self._set_name("%s-%d" % (module, idx))
  39     # This flag is set to True by build() and returned by is_built()
  40     self._is_built = False
  41
  42   def _set_name(self, name, read_only=False):
  43     """A convenience method for renaming a sandbox, useful when
  44     working with multiple repositories in the same unit test."""
  45     if not name is None:
  46       self.name = name
  47     self.read_only = read_only
  48     self.wc_dir = os.path.join(svntest.main.general_wc_dir, self.name)
  49     if not read_only:
  50       self.repo_dir = os.path.join(svntest.main.general_repo_dir, self.name)
  51       self.repo_url = (svntest.main.options.test_area_url + '/'
  52                        + svntest.main.pathname2url(self.repo_dir))
  53     else:
  54       self.repo_dir = svntest.main.pristine_dir
  55       self.repo_url = svntest.main.pristine_url
  56
  57     ### TODO: Move this into to the build() method
  58     # For dav tests we need a single authz file which must be present,
  59     # so we recreate it each time a sandbox is created with some default
  60     # contents.
  61     if self.repo_url.startswith("http"):
  62       # this dir doesn't exist out of the box, so we may have to make it
  63       if not os.path.exists(svntest.main.work_dir):
  64         os.makedirs(svntest.main.work_dir)
  65       self.authz_file = os.path.join(svntest.main.work_dir, "authz")
  66       open(self.authz_file, 'w').write("[/]\n* = rw\n")
  67
  68     # For svnserve tests we have a per-repository authz file, and it
  69     # doesn't need to be there in order for things to work, so we don't
  70     # have any default contents.
  71     elif self.repo_url.startswith("svn"):
  72       self.authz_file = os.path.join(self.repo_dir, "conf", "authz")
  73
  74     self.test_paths = [self.wc_dir, self.repo_dir]
  75
  76   def clone_dependent(self, copy_wc=False):
  77     """A convenience method for creating a near-duplicate of this
  78     sandbox, useful when working with multiple repositories in the
  79     same unit test.  If COPY_WC is true, make an exact copy of this
  80     sandbox's working copy at the new sandbox's working copy
  81     directory.  Any necessary cleanup operations are triggered by
  82     cleanup of the original sandbox."""
  83
  84     if not self.dependents:
  85       self.dependents = []
  86     clone = copy.deepcopy(self)
  87     self.dependents.append(clone)
  88     clone._set_name("%s-%d" % (self.name, len(self.dependents)))
  89     if copy_wc:
  90       self.add_test_path(clone.wc_dir)
  91       shutil.copytree(self.wc_dir, clone.wc_dir, symlinks=True)
  92     return clone
  93
  94   def build(self, name=None, create_wc=True, read_only=False):
  95     """Make a 'Greek Tree' repo (or refer to the central one if READ_ONLY),
  96        and check out a WC from it (unless CREATE_WC is false). Change the
  97        sandbox's name to NAME. See actions.make_repo_and_wc() for details."""
  98     self._set_name(name, read_only)
  99     if svntest.actions.make_repo_and_wc(self, create_wc, read_only):
 100       raise svntest.Failure("Could not build repository and sandbox '%s'"
 101                             % self.name)
 102     else:
 103       self._is_built = True
 104
 105   def add_test_path(self, path, remove=True):
 106     self.test_paths.append(path)
 107     if remove:
 108       svntest.main.safe_rmtree(path)
 109
 110   def add_repo_path(self, suffix, remove=True):
 111     """Generate a path, under the general repositories directory, with
 112        a name that ends in SUFFIX, e.g. suffix="2" -> ".../basic_tests.2".
 113        If REMOVE is true, remove anything currently on disk at that path.
 114        Remember that path so that the automatic clean-up mechanism can
 115        delete it at the end of the test. Generate a repository URL to
 116        refer to a repository at that path. Do not create a repository.
 117        Return (REPOS-PATH, REPOS-URL)."""
 118     path = (os.path.join(svntest.main.general_repo_dir, self.name)
 119             + '.' + suffix)
 120     url = svntest.main.options.test_area_url + \
 121                                         '/' + svntest.main.pathname2url(path)
 122     self.add_test_path(path, remove)
 123     return path, url
 124
 125   def add_wc_path(self, suffix, remove=True):
 126     """Generate a path, under the general working copies directory, with
 127        a name that ends in SUFFIX, e.g. suffix="2" -> ".../basic_tests.2".
 128        If REMOVE is true, remove anything currently on disk at that path.
 129        Remember that path so that the automatic clean-up mechanism can
 130        delete it at the end of the test. Do not create a working copy.
 131        Return the generated WC-PATH."""
 132     path = self.wc_dir + '.' + suffix
 133     self.add_test_path(path, remove)
 134     return path
 135
 136   def cleanup_test_paths(self):
 137     "Clean up detritus from this sandbox, and any dependents."
 138     if self.dependents:
 139       # Recursively cleanup any dependent sandboxes.
 140       for sbox in self.dependents:
 141         sbox.cleanup_test_paths()
 142     # cleanup all test specific working copies and repositories
 143     for path in self.test_paths:
 144       if not path is svntest.main.pristine_dir:
 145         _cleanup_test_path(path)
 146
 147   def is_built(self):
 148     "Returns True when build() has been called on this instance."
 149     return self._is_built
 150
 151   def ospath(self, relpath, wc_dir=None):
 152     if wc_dir is None:
 153       wc_dir = self.wc_dir
 154     return os.path.join(wc_dir, svntest.wc.to_ospath(relpath))
 155
 156   def simple_commit(self, target=None):
 157     assert not self.read_only
 158     if target is None:
 159       target = self.wc_dir
 160     svntest.main.run_svn(False, 'commit',
 161                          '-m', svntest.main.make_log_msg(),
 162                          target)
 163
 164   def simple_rm(self, *targets):
 165     assert len(targets) > 0
 166     if len(targets) == 1 and is_url(targets[0]):
 167       assert not self.read_only
 168       targets = ('-m', svntests.main.make_log_msg(), targets[0])
 169     svntest.main.run_svn(False, 'rm', *targets)
 170
 171   def simple_mkdir(self, *targets):
 172     assert len(targets) > 0
 173     if len(targets) == 1 and is_url(targets[0]):
 174       assert not self.read_only
 175       targets = ('-m', svntests.main.make_log_msg(), targets[0])
 176     svntest.main.run_svn(False, 'mkdir', *targets)
 177
 178   def simple_add(self, *targets):
 179     assert len(targets) > 0
 180     svntest.main.run_svn(False, 'add', *targets)
 181
 182   def simple_revert(self, *targets):
 183     assert len(targets) > 0
 184     svntest.main.run_svn(False, 'revert', *targets)
 185
 186   def simple_propset(self, name, value, *targets):
 187     assert len(targets) > 0
 188     svntest.main.run_svn(False, 'propset', name, value, *targets)
 189
 190   def simple_propdel(self, name, *targets):
 191     assert len(targets) > 0
 192     svntest.main.run_svn(False, 'propdel', name, *targets)
 193
 194
 195 def is_url(target):
 196   return (target.startswith('^/')
 197           or target.startswith('file://')
 198           or target.startswith('http://')
 199           or target.startswith('https://')
 200           or target.startswith('svn://')
 201           or target.startswith('svn+ssh://'))
 202
 203
 204 _deferred_test_paths = []
 205
 206 def cleanup_deferred_test_paths():
 207   global _deferred_test_paths
 208   test_paths = _deferred_test_paths
 209   _deferred_test_paths = []
 210   for path in test_paths:
 211     _cleanup_test_path(path, True)
 212
 213
 214 def _cleanup_test_path(path, retrying=False):
 215   if svntest.main.options.verbose:
 216     if retrying:
 217       print("CLEANUP: RETRY: %s" % path)
 218     else:
 219       print("CLEANUP: %s" % path)
 220   try:
 221     svntest.main.safe_rmtree(path)
 222   except:
 223     if svntest.main.verbose_mode:
 224       print("WARNING: cleanup failed, will try again later")
 225     _deferred_test_paths.append(path)