remote.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 2 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 """Subversion BzrDir formats."""
  17
  18 from bzrlib import urlutils
  19 from bzrlib.bzrdir import BzrDirFormat, BzrDir, format_registry
  20 from bzrlib.errors import (NotBranchError, NotLocalUrl, NoRepositoryPresent,
  21                            NoWorkingTree, AlreadyBranchError)
  22 from bzrlib.transport.local import LocalTransport
  23
  24 from svn.core import SubversionException
  25 import svn.core, svn.repos
  26
  27 from errors import NoSvnRepositoryPresent
  28 from format import get_rich_root_format, SvnRemoteFormat
  29 from repository import SvnRepository
  30 from transport import SvnRaTransport, bzr_to_svn_url, get_svn_ra_transport
  31
  32
  33 class SvnRemoteAccess(BzrDir):
  34     """BzrDir implementation for Subversion connections.
  35
  36     This is used for all non-checkout connections
  37     to Subversion repositories.
  38     """
  39     def __init__(self, _transport, _format=None):
  40         """See BzrDir.__init__()."""
  41         _transport = get_svn_ra_transport(_transport)
  42         if _format is None:
  43             _format = SvnRemoteFormat()
  44         self._format = _format
  45         self.transport = None
  46         self.root_transport = _transport
  47
  48         svn_url = bzr_to_svn_url(self.root_transport.base)
  49         self.svn_root_url = _transport.get_repos_root()
  50
  51         assert svn_url.startswith(self.svn_root_url)
  52         self.branch_path = svn_url[len(self.svn_root_url):]
  53
  54     def clone(self, url, revision_id=None, force_new_repo=False):
  55         """See BzrDir.clone().
  56
  57         Not supported on Subversion connections.
  58         """
  59         raise NotImplementedError(SvnRemoteAccess.clone)
  60
  61     def sprout(self, url, revision_id=None, force_new_repo=False,
  62             recurse='down', possible_transports=None):
  63         """See BzrDir.sprout()."""
  64         # FIXME: Use possible_transports
  65         # FIXME: Use recurse
  66         format = get_rich_root_format()
  67         result = format.initialize(url)
  68         repo = self.find_repository()
  69         if force_new_repo:
  70             result_repo = repo.clone(result, revision_id)
  71         else:
  72             try:
  73                 result_repo = result.find_repository()
  74                 result_repo.fetch(repo, revision_id=revision_id)
  75             except NoRepositoryPresent:
  76                 result_repo = repo.clone(result, revision_id)
  77         branch = self.open_branch()
  78         result_branch = branch.sprout(result, revision_id)
  79         if result_branch.repository.make_working_trees():
  80             result.create_workingtree()
  81         return result
  82
  83     def open_repository(self, _unsupported=False):
  84         """Open the repository associated with this BzrDir.
  85
  86         :return: instance of SvnRepository.
  87         """
  88         if self.branch_path == "":
  89             return SvnRepository(self, self.root_transport)
  90         raise NoSvnRepositoryPresent(self.root_transport.base)
  91
  92     def find_repository(self):
  93         """Open the repository associated with this BzrDir.
  94
  95         :return: instance of SvnRepository.
  96         """
  97         transport = self.root_transport
  98         if self.svn_root_url != transport.base:
  99             transport = transport.clone_root()
 100         return SvnRepository(self, transport, self.branch_path)
 101
 102     def open_workingtree(self, _unsupported=False,
 103             recommend_upgrade=True):
 104         """See BzrDir.open_workingtree().
 105
 106         Will always raise NotLocalUrl as this
 107         BzrDir can not be associated with working trees.
 108         """
 109         # Working trees never exist on remote Subversion repositories
 110         raise NoWorkingTree(self.root_transport.base)
 111
 112     def create_workingtree(self, revision_id=None):
 113         """See BzrDir.create_workingtree().
 114
 115         Will always raise NotLocalUrl as this
 116         BzrDir can not be associated with working trees.
 117         """
 118         raise NotLocalUrl(self.root_transport.base)
 119
 120     def needs_format_conversion(self, format=None):
 121         """See BzrDir.needs_format_conversion()."""
 122         # if the format is not the same as the system default,
 123         # an upgrade is needed.
 124         if format is None:
 125             format = BzrDirFormat.get_default_format()
 126         return not isinstance(self._format, format.__class__)
 127
 128     def import_branch(self, source, stop_revision=None):
 129         """Create a new branch in this repository, possibly
 130         with the specified history, optionally importing revisions.
 131
 132         :param source: Source branch
 133         :param stop_revision: Tip of new branch
 134         :return: Branch object
 135         """
 136         from commit import push_new
 137         if stop_revision is None:
 138             stop_revision = source.last_revision()
 139         target_branch_path = self.branch_path.strip("/")
 140         repos = self.find_repository()
 141         full_branch_url = urlutils.join(repos.transport.base,
 142                                         target_branch_path)
 143         if repos.transport.check_path(target_branch_path,
 144             repos.transport.get_latest_revnum()) != svn.core.svn_node_none:
 145             raise AlreadyBranchError(full_branch_url)
 146         push_new(repos, target_branch_path, source, stop_revision)
 147         branch = self.open_branch()
 148         branch.pull(source, stop_revision=stop_revision)
 149         return branch
 150
 151     def create_branch(self):
 152         """See BzrDir.create_branch()."""
 153         from branch import SvnBranch
 154         repos = self.find_repository()
 155
 156         if self.branch_path != "":
 157             # TODO: Set NULL_REVISION in SVN_PROP_BZR_BRANCHING_SCHEME
 158             repos.transport.mkdir(self.branch_path.strip("/"))
 159         elif repos.transport.get_latest_revnum() > 0:
 160             # Bail out if there are already revisions in this repository
 161             raise AlreadyBranchError(self.root_transport.base)
 162         branch = SvnBranch(self.root_transport.base, repos, self.branch_path)
 163         branch.bzrdir = self
 164         return branch
 165
 166     def open_branch(self, unsupported=True):
 167         """See BzrDir.open_branch()."""
 168         from branch import SvnBranch
 169         repos = self.find_repository()
 170         branch = SvnBranch(self.root_transport.base, repos, self.branch_path)
 171         branch.bzrdir = self
 172         return branch
 173
 174     def create_repository(self, shared=False, format=None):
 175         """See BzrDir.create_repository."""
 176         return self.open_repository()
 177
 178