branch.py

   1 # Copyright (C) 2005-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 """Handles branch-specific operations."""
  17
  18 from bzrlib.branch import Branch, BranchFormat, BranchCheckResult, PullResult
  19 from bzrlib.bzrdir import BzrDir
  20 from bzrlib.errors import NoSuchFile, DivergedBranches
  21 from bzrlib.inventory import (Inventory)
  22 from bzrlib.trace import mutter
  23 from bzrlib.workingtree import WorkingTree
  24
  25 import svn.client, svn.core
  26
  27 from commit import push_as_merged
  28 from repository import SvnRepository
  29 from transport import bzr_to_svn_url, svn_config
  30
  31
  32 class FakeControlFiles(object):
  33     """Dummy implementation of ControlFiles.
  34
  35     This is required as some code relies on controlfiles being
  36     available."""
  37     def get_utf8(self, name):
  38         raise NoSuchFile(name)
  39
  40     def get(self, name):
  41         raise NoSuchFile(name)
  42
  43     def break_lock(self):
  44         pass
  45
  46
  47 class SvnBranch(Branch):
  48     """Maps to a Branch in a Subversion repository """
  49     def __init__(self, base, repository, branch_path):
  50         """Instantiate a new SvnBranch.
  51
  52         :param repos: SvnRepository this branch is part of.
  53         :param branch_path: Relative path inside the repository this
  54             branch is located at.
  55         """
  56         super(SvnBranch, self).__init__()
  57         self.repository = repository
  58         assert isinstance(self.repository, SvnRepository)
  59         self.branch_path = branch_path
  60         self.control_files = FakeControlFiles()
  61         self.base = base.rstrip("/")
  62         self._format = SvnBranchFormat()
  63         self._revision_history = None
  64
  65     def check(self):
  66         """See Branch.Check.
  67
  68         Doesn't do anything for Subversion repositories at the moment (yet).
  69         """
  70         return BranchCheckResult(self)
  71
  72     def _create_heavyweight_checkout(self, to_location, revision_id=None):
  73         checkout_branch = BzrDir.create_branch_convenience(
  74             to_location, force_new_tree=False)
  75         checkout = checkout_branch.bzrdir
  76         checkout_branch.bind(self)
  77         # pull up to the specified revision_id to set the initial
  78         # branch tip correctly, and seed it with history.
  79         checkout_branch.pull(self, stop_revision=revision_id)
  80         return checkout.create_workingtree(revision_id)
  81
  82     """Look up the matching revision number on the mainline of the
  83     branch.
  84
  85     :param revid: Revision id to look up.
  86     :return: Revision number on the branch.
  87     :raises NoSuchRevision: If the revision id was not found.
  88     """
  89     def lookup_revision_id(self, revid):
  90         (bp, revnum) = self.repository.lookup_revision_id(revid)
  91         assert bp.strip("/") == self.branch_path.strip("/"), \
  92                 "Got %r, expected %r" % (bp, self.branch_path)
  93         return revnum
  94
  95     def _create_lightweight_checkout(self, to_location, revision_id=None):
  96         peg_rev = svn.core.svn_opt_revision_t()
  97         peg_rev.kind = svn.core.svn_opt_revision_head
  98
  99         rev = svn.core.svn_opt_revision_t()
 100         if revision_id is None:
 101             rev.kind = svn.core.svn_opt_revision_head
 102         else:
 103             revnum = self.lookup_revision_id(revision_id)
 104             rev.kind = svn.core.svn_opt_revision_number
 105             rev.value.number = revnum
 106             mutter('hist: %r' % self.revision_history())
 107
 108         client_ctx = svn.client.create_context()
 109         client_ctx.config = svn_config
 110         svn.client.checkout(bzr_to_svn_url(self.base), to_location, rev,
 111                             True, client_ctx)
 112
 113         return WorkingTree.open(to_location)
 114
 115     def create_checkout(self, to_location, revision_id=None, lightweight=False):
 116         if lightweight:
 117             return self._create_lightweight_checkout(to_location, revision_id)
 118         else:
 119             return self._create_heavyweight_checkout(to_location, revision_id)
 120
 121     def generate_revision_id(self, revnum):
 122         return self.repository.generate_revision_id(revnum, self.branch_path)
 123
 124     def _generate_revision_history(self, last_revnum):
 125         self._revision_history = []
 126         for (branch, rev) in self.repository.follow_branch(
 127                 self.branch_path, last_revnum):
 128             self._revision_history.append(
 129                     self.repository.generate_revision_id(rev, branch))
 130         self._revision_history.reverse()
 131
 132     def get_root_id(self):
 133         if self.last_revision() is None:
 134             inv = Inventory()
 135         else:
 136             inv = self.repository.get_inventory(self.last_revision())
 137         return inv.root.file_id
 138
 139     def _get_nick(self):
 140         bp = self.branch_path.strip("/")
 141         if self.branch_path == "":
 142             return None
 143         return bp
 144
 145     nick = property(_get_nick)
 146
 147     def set_revision_history(self, rev_history):
 148         raise NotImplementedError(self.set_revision_history)
 149
 150     def set_last_revision_info(self, revno, revid):
 151         pass
 152
 153     def set_push_location(self, location):
 154         raise NotImplementedError(self.set_push_location)
 155
 156     def get_push_location(self):
 157         # get_push_location not supported on Subversion
 158         return None
 159
 160     def revision_history(self):
 161         if self._revision_history is None:
 162             self._generate_revision_history(self.repository._latest_revnum)
 163         return self._revision_history
 164
 165     def last_revision(self):
 166         # Shortcut for finding the tip. This avoids expensive generation time
 167         # on large branches.
 168         if self._revision_history is None:
 169             for (branch, rev) in self.repository.follow_branch(
 170                 self.branch_path, self.repository._latest_revnum):
 171                 return self.repository.generate_revision_id(rev, branch)
 172             return None
 173
 174         ph = self.revision_history()
 175         if ph:
 176             return ph[-1]
 177         else:
 178             return None
 179
 180     def pull(self, source, overwrite=False, stop_revision=None):
 181         result = PullResult()
 182         result.source_branch = source
 183         result.master_branch = None
 184         result.target_branch = self
 185         source.lock_read()
 186         try:
 187             (result.old_revno, result.old_revid) = self.last_revision_info()
 188             try:
 189                 self.update_revisions(source, stop_revision)
 190             except DivergedBranches:
 191                 if overwrite:
 192                     raise NotImplementedError('overwrite not supported for '
 193                                               'Subversion branches')
 194                 raise
 195             (result.new_revno, result.new_revid) = self.last_revision_info()
 196             return result
 197         finally:
 198             source.unlock()
 199
 200     def generate_revision_history(self, revision_id, last_rev=None,
 201         other_branch=None):
 202         """Create a new revision history that will finish with revision_id.
 203
 204         :param revision_id: the new tip to use.
 205         :param last_rev: The previous last_revision. If not None, then this
 206             must be a ancestory of revision_id, or DivergedBranches is raised.
 207         :param other_branch: The other branch that DivergedBranches should
 208             raise with respect to.
 209         """
 210         # stop_revision must be a descendant of last_revision
 211         # make a new revision history from the graph
 212
 213     def update_revisions(self, other, stop_revision=None):
 214         if isinstance(other, SvnBranch):
 215             if (self.last_revision() == stop_revision or
 216                 self.last_revision() == other.last_revision()):
 217                 return
 218             # Import from another Subversion branch
 219             assert other.repository.uuid == self.repository.uuid, \
 220                     "can only import from elsewhere in the same repository."
 221
 222             # FIXME: Make sure branches haven't diverged
 223             # FIXME: svn.ra.del_dir(self.base_path)
 224             # FIXME: svn.ra.copy_dir(other.base_path, self.base_path)
 225             raise NotImplementedError(self.pull)
 226         else:
 227             for rev_id in self.missing_revisions(other, stop_revision):
 228                 mutter('pushing %r to Svn branch' % rev_id)
 229                 push_as_merged(self, other, rev_id)
 230
 231     # The remote server handles all this for us
 232     def lock_write(self):
 233         pass
 234
 235     def lock_read(self):
 236         pass
 237
 238     def unlock(self):
 239         pass
 240
 241     def get_parent(self):
 242         return self.base
 243
 244     def set_parent(self, url):
 245         pass # FIXME: Use svn.client.switch()
 246
 247     def append_revision(self, *revision_ids):
 248         #raise NotImplementedError(self.append_revision)
 249         pass
 250
 251     def get_physical_lock_status(self):
 252         return False
 253
 254     def sprout(self, to_bzrdir, revision_id=None):
 255         result = BranchFormat.get_default_format().initialize(to_bzrdir)
 256         self.copy_content_into(result, revision_id=revision_id)
 257         return result
 258
 259     def copy_content_into(self, destination, revision_id=None):
 260         new_history = self.revision_history()
 261         if revision_id is not None:
 262             try:
 263                 new_history = new_history[:new_history.index(revision_id) + 1]
 264             except ValueError:
 265                 rev = self.repository.get_revision(revision_id)
 266                 new_history = rev.get_history(self.repository)[1:]
 267         destination.set_revision_history(new_history)
 268         parent = self.get_parent()
 269         if parent:
 270             destination.set_parent(parent)
 271
 272     def __str__(self):
 273         return '%s(%r)' % (self.__class__.__name__, self.base)
 274
 275     __repr__ = __str__
 276
 277
 278
 279 class SvnBranchFormat(BranchFormat):
 280     """Branch format for Subversion Branches."""
 281     def __init__(self):
 282         BranchFormat.__init__(self)
 283
 284     def get_format_description(self):
 285         """See Branch.get_format_description."""
 286         return 'Subversion Smart Server'
 287
 288     def get_format_string(self):
 289         return 'Subversion Smart Server'
 290
 291     def initialize(self, to_bzrdir):
 292         raise NotImplementedError(self.initialize)
 293