Mercurial > treepkg > treepkg
view treepkg/subversion.py @ 282:f58f9adb7dc3
Add functions to get SVN logs and extract tag revisions from it. Add
some tests for the log parsing.
author | Bernhard Herzog <bh@intevation.de> |
---|---|
date | Tue, 04 Aug 2009 10:09:12 +0000 |
parents | 4b700b39c32f |
children | 020421cd3ee2 |
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): pass 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 int(str_rev) 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=()): """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 """ self.url = url self.external_subdirs = external_subdirs def checkout(self, localdir, revision=None): """Checks out the repository into localdir. The revision parameter should be an and indicates the revision to check out. """ checkout(self.url, 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 last_changed_revision(self, localdir): """Returns the last changed revision of the working copy in localdir""" return max([last_changed_revision(os.path.join(localdir, d)) for d in [localdir] + list(self.external_subdirs)]) 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) 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 last_changed_revision(self): """Returns the last changed rev of the working copy""" return self.repository.last_changed_revision(self.localdir) 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. Normally, just appending the subdir to the tag URL would be enough for this, but the situation is more complex for kdebase_workspace and kdebase_runtime, whose code comes from different subdirectories of the kdebase-4.X-branch subdirectory (for enterprise4 tags). Here the revision number must be taken from kdebase-4.X-branch, but the URL to use when exporting the sources, must refer to e.g. kdebase-4.1-branch/kdebase_workspace. To achieve that, subdir may contain slashes to indicate subdirectories of subdirectories, but only the first part of subdir (up to the first slash) is used when determining the revision number. """ 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.split("/")[0] try: revision = last_changed_revision(revision_url) urlrev = (baseurl + "/" + subdir, revision) except SubversionError: pass return urlrev