Mercurial > treepkg > treepkg
view treepkg/git.py @ 577:7a9841e4958f
don't forget to import shlex
author | Bjoern Ricks <bricks@intevation.de> |
---|---|
date | Sat, 03 Sep 2011 11:48:00 +0000 |
parents | b8acd77fca60 |
children | 97a5e09c84dc |
line wrap: on
line source
# Copyright (C) 2010 by Intevation GmbH # Authors: # Andre Heinecke <aheinecke@intevation.de> # # This program is free software under the GPL (>=v2) # Read the file COPYING coming with the software for details. """Collection of Git utility code""" import os import shutil import re import StringIO import run from run import capture_output from cmdexpand import cmdexpand from util import extract_value_for_key class GitError(Exception): """Base class for Git specific errors raised by TreePKG""" class GitRepository(object): """Describes a git repository""" def __init__(self, url, branch=None, logger=None): """Initialize the git repository description Parameters: url -- The url of the repository branch -- The name of the remote Branch to track defaults to master """ self.url = url self.logger = logger if not branch: # as default track master as local-master self.local_branch = "local-master" self.branch = "master" else: self.local_branch = branch self.branch = branch if ":" in self.branch: branches = self.branch.split(":") self.local_branch = branches[0] self.branch = branches[1] def checkout(self, localdir): """Clones the repository at url into the localdir""" run.call(cmdexpand("git clone -q $url $localdir", **locals())) if self.branch: self.checkout_branch(localdir) def checkout_branch(self, localdir): self.log_info("Switching to local branch '%s' for branch '%s'" %\ (self.local_branch, self.branch)) run.call(cmdexpand("git checkout -q --track -b $local $branch", branch=self.branch, local=self.local_branch), cwd=localdir) def export(self, localdir, destdir): """Exports the working copy in localdir to destdir""" dest = destdir + os.sep run.call(cmdexpand("git checkout-index -a -f --prefix=$dest", dest=dest), cwd=localdir) def copy(self, localdir, destdir): """Copies the working copy to destdir (including .git dir)""" shutils.copytree(localdir, destdir) def pull(self, localdir): self.log_info("Pulling the repo in '%s'" % localdir) run.call(cmdexpand("git pull -q"), cwd=localdir) def update(self, localdir): """Runs git pull on the localdir.""" self.pull(localdir) output = capture_output(cmdexpand("git branch"), cwd=localdir) branches = output.splitlines() cur_branch = None all_branches = [] for tbranch in branches: tbranch = tbranch.strip() if tbranch.startswith("*"): cur_branch = tbranch[2:] tbranch = cur_branch all_branches.append(tbranch) if not self.local_branch in all_branches: self.checkout_branch(localdir) self.log_info("Current branch is '%s'" % cur_branch) # TODO: check if self.local_branch is curbranch # doesn't hurt if a checkout is done on the current branch if self.branch: self.log_info("Switching to local branch '%s'" % self.local_branch) run.call(cmdexpand("git checkout -q $branch", branch=self.local_branch), cwd=localdir) def log_info(self, *args): if self.logger is not None: self.logger.info(*args) def last_changed_revision(self, localdir): """Returns the SHA1 sum of the latest commit in the working copy in localdir""" output = run.capture_output(cmdexpand("git rev-parse HEAD"), cwd=localdir) if output is None: raise GitError("Cannot determine last changed revision for %r" % git_working_copy) return output.strip() def check_working_copy(self, localdir): """FIXME STUB: Not implemented for git""" return None class GitWorkingCopy(object): """Represents a checkout of a git repository""" def __init__(self, repository, localdir, logger=None): """ Initialize the working copy. Parameters: repository -- The GitRepository 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=0): """Updates the working copy or creates by checking out the repository. Revision number included for compatibility """ gitdir = os.path.join(self.localdir, ".git") branch = self.repository.branch if os.path.exists(gitdir): self.log_info("Updating the working copy in %r for repo " \ "%s and branch %s", self.localdir, self.repository.url, branch) self.repository.update(self.localdir) else: # TODO: better check if localdir contains files if os.path.exists(self.localdir): raise GitError("Working copy dir %s already exists. " \ " files. Can't checkout from %s" % (self.localdir, self.repository.url)) self.log_info("The working copy in %r doesn't exist yet." " Checking out branch %s from %r", self.localdir, branch, self.repository.url) self.repository.checkout(self.localdir) def export(self, destdir): """Exports the working copy to destdir""" self.repository.export(self.localdir, destdir) def export_tag(self, url, destdir, revision=None): """Export tag to destir """ self.export(destdir) def last_changed_revision(self): """Returns the last changed rev of the working copy""" return self.get_revision() def list_tags(self, pattern): output = run.capture_output(cmdexpand("git tag -l $pattern", pattern=pattern), cwd=self.localdir) return output.splitlines() def get_revision(self, refname="HEAD"): """Return the SHA1 sum of the latest commit""" output = run.capture_output(cmdexpand("git rev-parse $refname", refname=refname), cwd=self.localdir) if output is None: raise GitError("Cannot determine revision for %r" % self.localdir) return output.strip() def get_short_revision(self, refname="HEAD"): """Return the short SHA1 sum of the latest commit""" revision = self.get_revision(refname) return revision[:7] class TagDetector: """Class to detect tags from a git repository The tags are found using the parameters: url -- The url of the git repository to use pattern -- A regular expression matching the tags """ def __init__(self, url, pattern, localdir): self.url = url self.pattern = pattern self.localdir = localdir self.repo = GitRepository(url) self.workingcopy = GitWorkingCopy(self.repo, localdir) def list_tags(self): self.repo.pull(self.localdir) tags = self.workingcopy.list_tags(self.pattern) return sorted(tags) def newest_tag_revision(self): candidates = self.list_tags() urlrev = (None, None) if candidates: newest = candidates[-1] try: rev = self.workingcopy.get_revision(newest) urlrev = (newest, rev) except GitError: pass return urlrev def tag_pkg_parameters(self, tag_name): # FIXME: Don't hardcore regex #match = re.search(r"enterprise[^.]*\.[^.]*\." match = re.search(r"enterprise[^.]*\.[^.]*" r"(?P<date>[0-9]{8})", tag_name) if match: date = match.group("date") return (date, 1) else: raise GitError("Cannot determine tag parameters from %s" % tag_name)