tests/__init__.py

   1 # Copyright (C) 2006-2007 Jelmer Vernooij <jelmer@samba.org>
   2
   3 # This program is free software; you can redistribute it and/or modify
   4 # it under the terms of the GNU General Public License as published by
   5 # the Free Software Foundation; either version 3 of the License, or
   6 # (at your option) any later version.
   7
   8 # This program is distributed in the hope that it will be useful,
   9 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 # GNU General Public License for more details.
  12
  13 # You should have received a copy of the GNU General Public License
  14 # along with this program; if not, write to the Free Software
  15 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
  16
  17 """Tests for the bzr-svn plugin."""
  18
  19 import os
  20 import sys
  21 import bzrlib
  22
  23 from cStringIO import StringIO
  24
  25 from bzrlib import osutils, urlutils
  26 from bzrlib.bzrdir import BzrDir
  27 from bzrlib.tests import TestCaseInTempDir, TestSkipped
  28 from bzrlib.trace import mutter
  29 from bzrlib.urlutils import local_path_to_url
  30 from bzrlib.workingtree import WorkingTree
  31
  32 import svn.core, svn.repos
  33 from bzrlib.plugins.svn.ra import RemoteAccess
  34
  35 class TestCaseWithSubversionRepository(TestCaseInTempDir):
  36     """A test case that provides the ability to build Subversion
  37     repositories."""
  38
  39     def setUp(self):
  40         super(TestCaseWithSubversionRepository, self).setUp()
  41         self.client_ctx = svn.client.create_context()
  42         self.client_ctx.log_msg_func2 = svn.client.svn_swig_py_get_commit_log_func
  43         self.client_ctx.log_msg_baton2 = self.log_message_func
  44
  45     def log_message_func(self, items, pool):
  46         return self.next_message
  47
  48     def make_repository(self, relpath, allow_revprop_changes=True):
  49         """Create a repository.
  50
  51         :return: Handle to the repository.
  52         """
  53         abspath = os.path.join(self.test_dir, relpath)
  54
  55         svn.repos.create(abspath, '', '', None, None)
  56
  57         if allow_revprop_changes:
  58             if sys.platform == 'win32':
  59                 revprop_hook = os.path.join(abspath, "hooks", "pre-revprop-change.bat")
  60                 open(revprop_hook, 'w').write("exit 0\n")
  61             else:
  62                 revprop_hook = os.path.join(abspath, "hooks", "pre-revprop-change")
  63                 open(revprop_hook, 'w').write("#!/bin/sh\n")
  64                 os.chmod(revprop_hook, os.stat(revprop_hook).st_mode | 0111)
  65
  66         return local_path_to_url(abspath)
  67
  68     def make_remote_bzrdir(self, relpath):
  69         """Create a repository."""
  70
  71         repos_url = self.make_repository(relpath)
  72
  73         return BzrDir.open("svn+%s" % repos_url)
  74
  75     def open_local_bzrdir(self, repos_url, relpath):
  76         """Open a local BzrDir."""
  77
  78         self.make_checkout(repos_url, relpath)
  79
  80         return BzrDir.open(relpath)
  81
  82     def make_local_bzrdir(self, repos_path, relpath):
  83         """Create a repository and checkout."""
  84
  85         repos_url = self.make_repository(repos_path)
  86
  87         return self.open_local_bzrdir(repos_url, relpath)
  88
  89     def make_checkout(self, repos_url, relpath):
  90         rev = svn.core.svn_opt_revision_t()
  91         rev.kind = svn.core.svn_opt_revision_head
  92
  93         svn.client.checkout2(repos_url, relpath,
  94                 rev, rev, True, False, self.client_ctx)
  95
  96     @staticmethod
  97     def create_checkout(branch, path, revision_id=None, lightweight=False):
  98         return branch.create_checkout(path, revision_id=revision_id,
  99                                           lightweight=lightweight)
 100
 101     @staticmethod
 102     def open_checkout(url):
 103         return WorkingTree.open(url)
 104
 105     @staticmethod
 106     def open_checkout_bzrdir(url):
 107         return BzrDir.open(url)
 108
 109     @staticmethod
 110     def create_branch_convenience(url):
 111         return BzrDir.create_branch_convenience(url)
 112
 113     def client_set_prop(self, path, name, value):
 114         if value is None:
 115             value = ""
 116         svn.client.propset2(name, value, path, False, True, self.client_ctx)
 117
 118     def client_get_prop(self, path, name, revnum=None, recursive=False):
 119         rev = svn.core.svn_opt_revision_t()
 120
 121         if revnum is None:
 122             rev.kind = svn.core.svn_opt_revision_working
 123         else:
 124             rev.kind = svn.core.svn_opt_revision_number
 125             rev.value.number = revnum
 126         ret = svn.client.propget2(name, path, rev, rev, recursive,
 127                                   self.client_ctx)
 128         if recursive:
 129             return ret
 130         else:
 131             return ret.values()[0]
 132
 133     def client_get_revprop(self, url, revnum, name):
 134         rev = svn.core.svn_opt_revision_t()
 135         rev.kind = svn.core.svn_opt_revision_number
 136         rev.value.number = revnum
 137         return svn.client.revprop_get(name, url, rev, self.client_ctx)[0]
 138
 139     def client_set_revprop(self, url, revnum, name, value):
 140         rev = svn.core.svn_opt_revision_t()
 141         rev.kind = svn.core.svn_opt_revision_number
 142         rev.value.number = revnum
 143         svn.client.revprop_set(name, value, url, rev, True, self.client_ctx)
 144
 145     def client_commit(self, dir, message=None, recursive=True):
 146         """Commit current changes in specified working copy.
 147
 148         :param relpath: List of paths to commit.
 149         """
 150         olddir = os.path.abspath('.')
 151         self.next_message = message
 152         os.chdir(dir)
 153         info = svn.client.commit2(["."], recursive, False, self.client_ctx)
 154         os.chdir(olddir)
 155         assert info is not None
 156         return (info.revision, info.date, info.author)
 157
 158     def client_add(self, relpath, recursive=True):
 159         """Add specified files to working copy.
 160
 161         :param relpath: Path to the files to add.
 162         """
 163         svn.client.add3(relpath, recursive, False, False, self.client_ctx)
 164
 165     def revnum_to_opt_rev(self, revnum):
 166         rev = svn.core.svn_opt_revision_t()
 167         if revnum is None:
 168             rev.kind = svn.core.svn_opt_revision_head
 169         else:
 170             assert isinstance(revnum, int)
 171             rev.kind = svn.core.svn_opt_revision_number
 172             rev.value.number = revnum
 173         return rev
 174
 175     def client_log(self, path, start_revnum=None, stop_revnum=None):
 176         assert isinstance(path, str)
 177         ret = {}
 178         def rcvr(orig_paths, rev, author, date, message, pool):
 179             ret[rev] = (orig_paths, author, date, message)
 180         svn.client.log([path], self.revnum_to_opt_rev(start_revnum),
 181                        self.revnum_to_opt_rev(stop_revnum),
 182                        True,
 183                        True,
 184                        rcvr,
 185                        self.client_ctx)
 186         return ret
 187
 188     def client_delete(self, relpath):
 189         """Remove specified files from working copy.
 190
 191         :param relpath: Path to the files to remove.
 192         """
 193         svn.client.delete2([relpath], True, self.client_ctx)
 194
 195     def client_copy(self, oldpath, newpath, revnum=None):
 196         """Copy file in working copy.
 197
 198         :param oldpath: Relative path to original file.
 199         :param newpath: Relative path to new file.
 200         """
 201         rev = svn.core.svn_opt_revision_t()
 202         if revnum is None:
 203             rev.kind = svn.core.svn_opt_revision_head
 204         else:
 205             rev.kind = svn.core.svn_opt_revision_number
 206             rev.value.number = revnum
 207         svn.client.copy2(oldpath, rev, newpath, self.client_ctx)
 208
 209     def client_update(self, path):
 210         rev = svn.core.svn_opt_revision_t()
 211         rev.kind = svn.core.svn_opt_revision_head
 212         svn.client.update(path, rev, True, self.client_ctx)
 213
 214     def build_tree(self, files):
 215         """Create a directory tree.
 216
 217         :param files: Dictionary with filenames as keys, contents as
 218             values. None as value indicates a directory.
 219         """
 220         for f in files:
 221             if files[f] is None:
 222                 try:
 223                     os.makedirs(f)
 224                 except OSError:
 225                     pass
 226             else:
 227                 try:
 228                     os.makedirs(os.path.dirname(f))
 229                 except OSError:
 230                     pass
 231                 open(f, 'w').write(files[f])
 232
 233     def make_client_and_bzrdir(self, repospath, clientpath):
 234         repos_url = self.make_client(repospath, clientpath)
 235
 236         return BzrDir.open("svn+%s" % repos_url)
 237
 238     def make_client(self, repospath, clientpath, allow_revprop_changes=True):
 239         """Create a repository and a checkout. Return the checkout.
 240
 241         :param relpath: Optional relpath to check out if not the full
 242             repository.
 243         :param clientpath: Path to checkout
 244         :return: Repository URL.
 245         """
 246         repos_url = self.make_repository(repospath,
 247             allow_revprop_changes=allow_revprop_changes)
 248         self.make_checkout(repos_url, clientpath)
 249         return repos_url
 250
 251     def dumpfile(self, repos):
 252         """Create a dumpfile for the specified repository.
 253
 254         :return: File name of the dumpfile.
 255         """
 256         raise NotImplementedError(self.dumpfile)
 257
 258     def open_fs(self, relpath):
 259         """Open a fs.
 260
 261         :return: FS.
 262         """
 263         repos = svn.repos.open(relpath)
 264
 265         return svn.repos.fs(repos)
 266
 267     def commit_editor(self, url, message="Test commit"):
 268         ra = RemoteAccess(url.encode('utf8'))
 269         class CommitEditor:
 270             def __init__(self, ra, editor, base_revnum, base_url):
 271                 self._used = False
 272                 self.ra = ra
 273                 self.base_revnum = base_revnum
 274                 self.editor = editor
 275                 self.data = {}
 276                 self.create = set()
 277                 self.props = {}
 278                 self.copyfrom = {}
 279                 self.base_url = base_url
 280
 281             def _parts(self, path):
 282                 return path.strip("/").split("/")
 283
 284             def add_dir(self, path, copyfrom_path=None, copyfrom_rev=-1):
 285                 self.create.add(path)
 286                 if copyfrom_path is not None:
 287                     if copyfrom_rev == -1:
 288                         copyfrom_rev = self.base_revnum
 289                     copyfrom_path = os.path.join(self.base_url, copyfrom_path)
 290                 self.copyfrom[path] = (copyfrom_path, copyfrom_rev)
 291                 self.open_dir(path)
 292
 293             def open_dir(self, path):
 294                 x = self.data
 295                 for p in self._parts(path):
 296                     if not p in x:
 297                         x[p] = {}
 298                     x = x[p]
 299                 return x
 300
 301             def add_file(self, path, contents=None):
 302                 self.create.add(path)
 303                 self.change_file(path, contents)
 304
 305             def change_file(self, path, contents=None):
 306                 parts = self._parts(path)
 307                 x = self.open_dir("/".join(parts[:-1]))
 308                 if contents is None:
 309                     contents = osutils.rand_chars(100)
 310                 x[parts[-1]] = contents
 311
 312             def delete(self, path):
 313                 parts = self._parts(path)
 314                 x = self.open_dir("/".join(parts[:-1]))
 315                 x[parts[-1]] = None
 316
 317             def change_dir_prop(self, path, propname, propval):
 318                 self.open_dir(path)
 319                 if not path in self.props:
 320                     self.props[path] = {}
 321                 self.props[path][propname] = propval
 322
 323             def change_file_prop(self, path, propname, propval):
 324                 parts = self._parts(path)
 325                 x = self.open_dir("/".join(parts[:-1]))
 326                 x[parts[-1]] = ()
 327                 if not path in self.props:
 328                     self.props[path] = {}
 329                 self.props[path][propname] = propval
 330
 331             def _process_dir(self, dir_baton, dir_dict, path):
 332                 for name, contents in dir_dict.items():
 333                     subpath = urlutils.join(path, name).strip("/")
 334                     if contents is None:
 335                         dir_baton.delete_entry(subpath, -1)
 336                     elif isinstance(contents, dict):
 337                         if subpath in self.create:
 338                             child_baton = dir_baton.add_directory(subpath, self.copyfrom[subpath][0], self.copyfrom[subpath][1])
 339                         else:
 340                             child_baton = dir_baton.open_directory(subpath, -1)
 341                         if subpath in self.props:
 342                             for k, v in self.props[subpath].items():
 343                                 child_baton.change_prop(k, v)
 344
 345                         self._process_dir(child_baton, dir_dict[name], subpath)
 346
 347                         child_baton.close()
 348                     else:
 349                         if subpath in self.create:
 350                             child_baton = dir_baton.add_file(subpath, None, -1)
 351                         else:
 352                             child_baton = dir_baton.open_file(subpath)
 353                         if isinstance(contents, str):
 354                             (txdelta, txbaton) = child_baton.apply_textdelta()
 355                             svn.delta.svn_txdelta_send_stream(StringIO(contents), txdelta, txbaton)
 356                         if subpath in self.props:
 357                             for k, v in self.props[subpath].items():
 358                                 child_baton.change_prop(k, v)
 359                         child_baton.close()
 360
 361             def done(self):
 362                 assert self._used == False
 363                 self._used = True
 364                 root_baton = self.editor.open_root(self.base_revnum)
 365                 self._process_dir(root_baton, self.data, "")
 366                 root_baton.close()
 367                 self.editor.close()
 368
 369                 my_revnum = ra.get_latest_revnum()
 370                 assert my_revnum > self.base_revnum
 371
 372                 return my_revnum
 373
 374         base_revnum = ra.get_latest_revnum()
 375         editor = ra.get_commit_editor({"svn:log": message})
 376         return CommitEditor(ra, editor, base_revnum, url)
 377
 378
 379 def test_suite():
 380     from unittest import TestSuite
 381
 382     from bzrlib.tests import TestUtil
 383
 384     loader = TestUtil.TestLoader()
 385
 386     suite = TestSuite()
 387
 388     testmod_names = [
 389             'test_branch',
 390             'test_branchprops',
 391             'test_changes',
 392             'test_checkout',
 393             'test_commit',
 394             'test_config',
 395             'test_convert',
 396             'test_errors',
 397             'test_fetch',
 398             'test_fileids',
 399             'test_logwalker',
 400             'test_mapping',
 401             'test_push',
 402             'test_radir',
 403             'test_repos',
 404             'test_revids',
 405             'test_revspec',
 406             'test_scheme',
 407             'test_svk',
 408             'test_transport',
 409             'test_tree',
 410             'test_upgrade',
 411             'test_workingtree',
 412             'test_blackbox']
 413     suite.addTest(loader.loadTestsFromModuleNames(["%s.%s" % (__name__, i) for i in testmod_names]))
 414
 415     return suite