Mercurial > treepkg > treepkg
view treepkg/subversion.py @ 546:149d18aca4f7
fix variable name
author | Bjoern Ricks <bricks@intevation.de> |
---|---|
date | Wed, 02 Feb 2011 10:00:48 +0000 |
parents | 8b49548aa8d4 |
children |
line wrap: on
line source
# Copyright (C) 2007, 2008, 2009 by Intevation GmbH # Authors: # Bernhard Herzog <bh@intevation.de> # # This program is free software under the GPL (>=v2) # Read the file COPYING coming with the software for details. """Collection of subversion utility code""" import os import shutil import re import StringIO from lxml import etree import run from cmdexpand import cmdexpand from util import extract_value_for_key class SubversionError(Exception): """Base class for subversion specific errors raised by TreePKG""" class SubversionUrlMismatchError(SubversionError): """The repository URL does not match the URL of a working copy""" def list_url(url): """Runs svn list with the given url and returns files listed as a list""" output = run.capture_output(cmdexpand("svn list $url", **locals())) return output.splitlines() def checkout(url, localdir, revision=None, recurse=True): """Runs svn to checkout the repository at url into the localdir""" args = [] if revision: args.extend(["--revision", revision]) if not recurse: args.append("-N") run.call(cmdexpand("svn checkout -q @args $url $localdir", **locals())) def update(localdir, revision=None, recurse=True): """Runs svn update on the localdir. The parameter revision, if given, is passed to svn as the value of the --revision option. """ args = [] if revision: args.extend(["--revision", revision]) if not recurse: args.append("-N") run.call(cmdexpand("svn update -q @args $localdir", **locals())) def export(src, dest, revision=None, recurse=True): """Runs svn export src dest""" args = [] if revision: args.extend(["--revision", revision]) if not recurse: args.append("-N") run.call(cmdexpand("svn export -q @args $src $dest", **locals())) def last_changed_revision(svn_working_copy): """return the last changed revision of an SVN working copy as an int""" # Make sure we run svn under the C locale to avoid localized # messages env = os.environ.copy() env["LANG"] = "C" output = run.capture_output(cmdexpand("svn info $svn_working_copy", **locals()), env=env) str_rev = extract_value_for_key(output.splitlines(), "Last Changed Rev:") if str_rev is None: raise SubversionError("Cannot determine last changed revision for %r" % svn_working_copy) return str_rev def svn_url(url_or_working_copy): """Returns the URL used for the working copy in svn_working_copy""" # Make sure we run svn under the C locale to avoid localized # messages env = os.environ.copy() env["LANG"] = "C" output = run.capture_output(cmdexpand("svn info $url_or_working_copy", **locals()), env=env) return extract_value_for_key(output.splitlines(), "URL:") def log_xml(url, base_revision): """Return the log in XML of the repository at url from base_revision to HEAD """ args = ["--revision", str(base_revision) + ":HEAD", "--verbose", "--xml"] return run.capture_output(cmdexpand("svn log @args $url", **locals())) def extract_tag_revisions(xml_log): """Extracts the revisions which changed an SVN tag since its creation This includes the revision which created the tag and all subsequent changes. The xml_log parameter should contain the xml-Version of the SVN log of the tag that includes at least the revision that created the tag and all the newer revisions. """ tree = etree.parse(StringIO.StringIO(xml_log)) tag_revisions = tree.xpath("logentry/@revision" "[.>=../../logentry/@revision" "[../paths/path[@copyfrom-path]]]") return tag_revisions class SvnRepository(object): """Describes a subversion repository""" def __init__(self, url, external_subdirs=(), subset=()): """Initialize the subversion repository description Parameters: url -- The url of the repository external_subdirs -- A list of subdirectories which are managed by svn externals definitions subset -- A sequence of (filename, recurse) pairs where filename is a filename (usually a directory name) relative to url and recurse should be a boolean indicating whether checkout filename with recursion. If recurse is False, svn checkout/export will be called with the -N option. The first item in subset should be for '.', which indicates the top-level directory under url. If a non-empty subset is given this will usually be ('.', False) so that the top-level directory is not checked out recursively. """ self.url = url self.external_subdirs = external_subdirs if not subset: # default subset is to checkout the top-level directory at # URL recursively. Alwas having a subset makes the code # simpler subset = [(".", True)] self.subset = subset def checkout(self, localdir, revision=None): """Checks out the repository into localdir. The revision parameter should be an int and indicates the revision to check out or it should be None to indicate that the newest version is to be checked out. """ base_url = self.url if not base_url.endswith("/"): base_url += "/" subdir, recurse = self.subset[0] checkout(base_url + subdir, os.path.join(localdir, subdir), revision=revision, recurse=recurse) for subdir, recurse in self.subset[1:]: update(os.path.join(localdir, subdir), revision=revision, recurse=recurse) if len(self.subset) > 1 and revision is None: # do an additional update on the whole working copy after # creating a subset checkout so that svn info will show # revision numbers that match the entire working copy # (externals are handled elsewhere). The repository might # have been changed between the initial checkout of the # top-level directory and the updates for the # subdirectories. update(localdir, revision=revision) def export(self, localdir, destdir): """Exports the working copy in localdir to destdir""" export(localdir, destdir) for subdir in self.external_subdirs: absdir = os.path.join(destdir, subdir) if not os.path.isdir(absdir): export(os.path.join(localdir, subdir), absdir) def export_tag(self, url, destdir, revision=None): """Exports the tag at url to destdir. Note: the implementation of this method would work for any URL but it really is intended to only be used for URLs to tags of the same code as represented by this object. """ base_url = url if not base_url.endswith("/"): base_url += "/" for subdir, recurse in self.subset: export(base_url + "/" + subdir, os.path.join(destdir, subdir), revision=revision, recurse=recurse) def last_changed_revision(self, localdir): """Returns the last changed revision of the working copy in localdir""" max_rev = max([int(last_changed_revision(os.path.join(localdir, d))) for d in [localdir] + list(self.external_subdirs)]) return str(max_rev) def check_working_copy(self, localdir): """Checks whether localdir contains a checkout of the repository. The check compares the expected URL with the one returned by svn info executed in localdir. Raises SubversionUrlMismatchError if the URLs do not match. """ localurl = svn_url(localdir) expected_url = svn_url(self.url) if localurl != expected_url: raise SubversionUrlMismatchError("Working copy in %r has URL %r," " expected %r" % (localdir, localurl, expected_url)) class SvnWorkingCopy(object): """Represents a checkout of a subversion repository""" def __init__(self, repository, localdir, logger=None): """ Initialize the working copy. Parameters: repository -- The SvnRepository instance describing the repository localdir -- The directory for the working copy logger -- logging object to use for some info/debug messages """ self.repository = repository self.localdir = localdir self.logger = logger def log_info(self, *args): if self.logger is not None: self.logger.info(*args) def update_or_checkout(self, revision=None): """Updates the working copy or creates by checking out the repository""" if os.path.exists(self.localdir): self.log_info("Updating the working copy in %r", self.localdir) self.repository.check_working_copy(self.localdir) update(self.localdir, revision=revision) else: self.log_info("The working copy in %r doesn't exist yet." " Checking out from %r", self.localdir, self.repository.url) self.repository.checkout(self.localdir, revision=revision) def export(self, destdir): """Exports the working copy to destdir""" self.repository.export(self.localdir, destdir) def export_tag(self, url, destdir, revision=None): """Exports the tag at url to destdir. The URL is expected to point to the same repository as the one used by the working copy and is intended to be used when exporting tagged versions of the code in the working copy. It's a method on the working copy so that the repository description including the subset settings are used. """ self.repository.export_tag(url, destdir, revision=revision) def last_changed_revision(self): """Returns the last changed rev of the working copy""" return self.repository.last_changed_revision(self.localdir) def get_revision(self): return self.last_changed_revision() def get_short_revision(self): # TODO: revision should be cached to avoid several calls to svn return self.get_revision() class ManualWorkingCopy(object): """A manually managed working copy""" def __init__(self, directory): self.directory = directory def update_or_checkout(self, revision=None, recurse=True): """This method does nothing""" pass def export(self, destdir): """Copies the entire working copy to destdir""" shutil.copytree(self.directory, destdir) def last_changed_revision(self): """Always returns 0""" return "0" class TagDetector(object): """Class to automatically find SVN tags and help package them The tags are found using three parameters: url -- The base url of the SVN tags directory to use pattern -- A regular expression matching the subdirectories to consider in the tag directory specified by the url subdir -- A subdirectory of the directory matched by pattern to export and use to determine revision number The subdir parameter is there to cope with the kdepim enterprise tags. The URL for a tag is of the form .../tags/kdepim/enterprise4.0.<date>.<rev> . Each such tag has subdirectories for kdepim, kdelibs, etc. The url and pattern are used to match the URL for the tag, and the subdir is used to select which part of the tag is meant. The subdir also determines which SVN directory's revision number is used. """ def __init__(self, url, pattern, subdir): self.url = url self.pattern = re.compile(pattern) self.subdir = subdir def list_tags(self): matches = [] if self.url: for tag in list_url(self.url): if self.pattern.match(tag.rstrip("/")): matches.append(tag) return sorted(matches) def newest_tag_revision(self): """Determines the newest tag revision and returns (tagurl, revno) If no tag can be found, the method returns the tuple (None, None). """ candidates = self.list_tags() urlrev = (None, None) if candidates: newest = candidates[-1] urlrev = self.determine_revision(self.url + "/" + newest, self.subdir) return urlrev def determine_revision(self, baseurl, subdir): urlrev = (None, None) revision_url = baseurl + "/" + subdir try: revision = last_changed_revision(revision_url) urlrev = (baseurl + "/" + subdir, revision) except SubversionError: pass return urlrev def log_xml(self, url): """Return the log in XML of the repository since the copy """ args = ["--stop-on-copy", "--verbose", "--xml"] return run.capture_output(cmdexpand("svn log @args $url", **locals())) def tag_pkg_parameters(self, tag_url): # FIXME: Don't hardcore svn tag path and regex match = re.search(r"/enterprise[^.]*\.[^.]*\." r"(?P<date>[0-9]{8})\.(?P<baserev>[0-9]+)/", tag_url) if match: date = match.group("date") # baserev is time since git migration baserev = match.group("baserev") xml_log = self.log_xml(tag_url) revisions = extract_tag_revisions(xml_log) tag_change_count = len(revisions) return (date, tag_change_count) else: raise RuntimeError("Cannot determine tag parameters from %s" % tag_url)