source: mergebot/trunk/mergebot/SvnOps.py @ 16

#!/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

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, )

def logcmd(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(logentry):
    """Given a log entry split out of svn log, return its revision number"""
    return int(logentry.split()[0][1:])

def get_branch_info(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 = get_rev_from_log(logs[1])
    startrev = get_rev_from_log(logs[-2])
    return (startrev, endrev)

def create_branch(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 logcmd(svncmd, logfile)

def delete_branch(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 _svn_new_rev_command(svncmd)

def checkout(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(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(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(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 _svn_new_rev_command(svncmd)

def _svn_new_rev_command(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