source: mergebot/trunk/mergebot/svn.py @ 47

#!/usr/bin/python
"""
Encapsulate logical Subversion operations so the various MergeBot actors can
operate at a higher level of abstraction.
"""

import os
import time
import re
import subprocess

def shell_quote(string):
    """Given a string, escape the characters interpretted by the shell."""
    for char in ["\\", "\"", "$"]:
        string = string.replace(char, "\\%s" % (char, ))
    return '"%s"' % (string, )


class SvnLib(object):
    """A library to provide a higher-level set of subversion operations."""
    def __init__(self):
        self.svn_version = self.get_svn_version()

    def get_svn_version(self):
        svn = subprocess.Popen(['svn', '--version', '--quiet'],
                                stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        version_string, _stderr = svn.communicate()
        return [int(x) for x in version_string.split('.')]
        
    def logcmd(self, cmd, logfile):
        """Log the cmd string, then execute it, appending its stdout and stderr
        to logfile."""
        open(logfile, "a").write("%s: %s\n" % (time.asctime(), cmd))
        return os.system("(%s) >>%s 2>&1" % (cmd, logfile))

    def get_rev_from_log(self, logentry):
        """Given a log entry split out of svn log, return its revision number"""
        return int(logentry.split()[0][1:])

    def does_url_exist_14(self, url):
        """Given a subversion url return true if it exists, false otherwise."""
        return not subprocess.call(['svn', 'log', '--limit=1',
                                    '--non-interactive', url],
                        stdout=open('/dev/null', 'w'),
                        stderr=open('/dev/null', 'w'))

    def does_url_exist_15(self, url):
        """Given a subversion url return true if it exists, false otherwise."""
        return not subprocess.call(['svn', 'ls', '--depth', 'empty',
                                    '--non-interactive', url],
                        stdout=open('/dev/null', 'w'),
                        stderr=open('/dev/null', 'w'))

    def does_url_exist(self, url):
        if self.svn_version[:2] == [1,4]:
            return self.does_url_exist_14(url)
        return self.does_url_exist_15(url)

    def get_branch_info(self, url, logfile):
        """Given a subversion url and a logfile, return (start_revision,
        end_revision) or None if it does not exist."""
        svncmd = os.popen("svn log --stop-on-copy --non-interactive %s 2>>%s" \
            % (url, logfile), "r")
        branchlog = svncmd.read()
        returnval = svncmd.close()
        if returnval:
            # This branch apparently doesn't exist
            return None
        logs = branchlog.split("-"*72 + "\n")
        # If there have been no commits on the branch since it was created,
        # there will only be one revision listed.... but the log will split
        # into 3 parts.
        endrev = self.get_rev_from_log(logs[1])
        startrev = self.get_rev_from_log(logs[-2])
        return (startrev, endrev)

    def create_branch(self, from_url, to_url, commit_message, logfile):
        """Create a branch copying from_url to to_url.  Commit as mergebot, and
        use the provided commit message."""
        svncmd = \
            "svn copy --username=mergebot --password=mergebot -m %s %s %s" \
            % (shell_quote(commit_message), from_url, to_url)
        return self.logcmd(svncmd, logfile)

    def delete_branch(self, url, commit_message, logfile):
        """This will generate a new revision.  Return the revision number, or
        -1 on failure.
        Assumes that the url exists.  You should call get_branch_info() to
        determine that first"""
        svncmd = "svn rm --no-auth-cache " \
            "--username=mergebot --password=mergebot " \
            "-m %s %s 2>>%s" % (shell_quote(commit_message), url, logfile)
        return self._svn_new_rev_command(svncmd)

    def checkout(self, from_url, workingdir, logfile):
        """Checkout from the given url into workingdir"""
        return os.system("svn checkout %s %s >>%s 2>&1" % (from_url, workingdir,
            logfile))

    def merge(self, from_url, workingdir, revision_range, logfile):
        """Returns a list (status, filename) tuples"""
        # There are a couple of different 'Skipped' messages.
        skipped_regex = re.compile("Skipped.* '(.*)'", re.M)
        start_rev, end_rev = revision_range
        pipe = os.popen("cd %s && svn merge --revision %s:%s %s . 2>>%s" % \
            (workingdir, start_rev, end_rev, from_url, logfile))
        output = pipe.readlines()
        # FIXME: check pipe.close for errors
        results = []
        for line in output:
            if line.startswith("Skipped"):
                # This kind of conflict requires special handling.
                filename = skipped_regex.findall(line)[0]
                status = "C"
            else:
                assert line[4] == ' ', "Unexpected output from svn merge " \
                    "operation; the 5th character should always be a space." \
                    "  Output was %r." % line
                filename = line[5:-1] # (strip trailing newline)
                status = line[:4].rstrip()
            results.append((status, filename))
        return results

    def conflicts_from_merge_results(self, results):
        """Given the output from merge, return a list of files that had
        conflicts."""
        conflicts = [filename for status, filename in results if 'C' in status]
        return conflicts

    def commit(self, workingdir, commit_message, logfile):
        """Returns newly committed revision number, or None if there was
        nothing to commit.  -1 on error."""
        svncmd = "cd %s && svn commit --no-auth-cache --username=mergebot " \
            "--password=mergebot -m %s 2>>%s" % (workingdir,
                shell_quote(commit_message), logfile)
        return self._svn_new_rev_command(svncmd)

    def _svn_new_rev_command(self, svncmd):
        """Given an svn command that results in a new revision, return the
        revision number, or -1 on error."""
        pipe = os.popen(svncmd)
        output = pipe.read()
        retval = pipe.close()
        if retval:
            new_revision = -1
        else:
            new_revisions = re.compile("Committed revision ([0-9]+)\\.",
                re.M).findall(output)
            if new_revisions:
                new_revision = new_revisions[0]
            else:
                new_revision = None
        return new_revision

# vim:foldcolumn=4 foldmethod=indent
# vim:tabstop=4 shiftwidth=4 softtabstop=4 expandtab