monotone.py

import tempfile
import os
from cStringIO import StringIO

from twisted.python import log
from twisted.application import service
from twisted.internet import defer, protocol, error, reactor
from twisted.internet.task import LoopingCall

from buildbot import util
from buildbot.interfaces import IChangeSource
from buildbot.changes.changes import Change

class _MTProtocol(protocol.ProcessProtocol):

    def __init__(self, deferred, cmdline):
        self.cmdline = cmdline
        self.deferred = deferred
        self.s = StringIO()

    def errReceived(self, text):
        log.msg("stderr: %s" % text)

    def outReceived(self, text):
        log.msg("stdout: %s" % text)
        self.s.write(text)

    def processEnded(self, reason):
        log.msg("Command %r exited with value %s" % (self.cmdline, reason))
        if isinstance(reason.value, error.ProcessDone):
            self.deferred.callback(self.s.getvalue())
        else:
            self.deferred.errback(reason)

class Monotone:
    """All methods of this class return a Deferred."""

    def __init__(self, bin, db):
        self.bin = bin
        self.db = db

    def _run_monotone(self, args):
        d = defer.Deferred()
        cmdline = (self.bin, "--db=" + self.db) + tuple(args)
        p = _MTProtocol(d, cmdline)
        log.msg("Running command: %r" % (cmdline,))
        log.msg("wd: %s" % os.getcwd())
        reactor.spawnProcess(p, self.bin, cmdline)
        return d

    def _process_revision_list(self, output):
        if output:
            return output.strip().split("\n")
        else:
            return []

    def get_interface_version(self):
        d = self._run_monotone(["automate", "interface_version"])
        d.addCallback(self._process_interface_version)
        return d

    def _process_interface_version(self, output):
        return tuple(map(int, output.strip().split(".")))

    def db_init(self):
        return self._run_monotone(["db", "init"])

    def db_migrate(self):
        return self._run_monotone(["db", "migrate"])

    def pull(self, server, pattern):
        return self._run_monotone(["pull", server, pattern])

    def get_revision(self, rid):
        return self._run_monotone(["cat", "revision", rid])

    def get_heads(self, branch, rcfile=""):
        cmd = ["automate", "heads", branch]
        if rcfile:
            cmd += ["--rcfile=" + rcfile]
        d = self._run_monotone(cmd)
        d.addCallback(self._process_revision_list)
        return d

    def erase_ancestors(self, revs):
        d = self._run_monotone(["automate", "erase_ancestors"] + revs)
        d.addCallback(self._process_revision_list)
        return d

    def ancestry_difference(self, new_rev, old_revs):
        d = self._run_monotone(["automate", "ancestry_difference", new_rev]
                               + old_revs)
        d.addCallback(self._process_revision_list)
        return d

    def descendents(self, rev):
        d = self._run_monotone(["automate", "descendents", rev])
        d.addCallback(self._process_revision_list)
        return d

    def log(self, rev, depth=None):
        if depth is not None:
            depth_arg = ["--last=%i" % (depth,)]
        else:
            depth_arg = []
        return self._run_monotone(["log", "-r", rev] + depth_arg)


class MonotoneSource(service.Service, util.ComparableMixin):
    """This source will poll a monotone server for changes and submit them to
    the change master.

    @param server_addr: monotone server specification (host:portno)

    @param branch: monotone branch to watch

    @param trusted_keys: list of keys whose code you trust

    @param db_path: path to monotone database to pull into

    @param pollinterval: interval in seconds between polls, defaults to 10 minutes
    @param monotone_exec: path to monotone executable, defaults to "monotone"
    """

    __implements__ = IChangeSource, service.Service.__implements__
    compare_attrs = ["server_addr", "trusted_keys", "db_path",
                     "pollinterval", "branch", "monotone_exec"]

    parent = None # filled in when we're added
    done_revisions = []
    last_revision = None
    loop = None
    d = None
    tmpfile = None
    monotone = None
    volatile = ["loop", "d", "tmpfile", "monotone"]

    def __init__(self, server_addr, branch, trusted_keys, db_path,
                 pollinterval=60 * 10, monotone_exec="monotone"):
        self.server_addr = server_addr
        self.branch = branch
        self.trusted_keys = trusted_keys
        self.db_path = db_path
        self.pollinterval = pollinterval
        self.monotone_exec = monotone_exec
        self.monotone = Monotone(self.monotone_exec, self.db_path)

    def startService(self):
        self.loop = LoopingCall(self.start_poll)
        self.loop.start(self.pollinterval)
        service.Service.startService(self)

    def stopService(self):
        self.loop.stop()
        return service.Service.stopService(self)

    def describe(self):
        return "monotone_source %s %s" % (self.server_addr,
                                          self.branch)

    def start_poll(self):
        if self.d is not None:
            log.msg("last poll still in progress, skipping next poll")
            return
        log.msg("starting poll")
        self.d = self._maybe_init_db()
        self.d.addCallback(self._do_netsync)
        self.d.addCallback(self._get_changes)
        self.d.addErrback(self._handle_error)

    def _handle_error(self, failure):
        log.err(failure)
        self.d = None

    def _maybe_init_db(self):
        if not os.path.exists(self.db_path):
            log.msg("init'ing db")
            return self.monotone.db_init()
        else:
            log.msg("db already exists, migrating")
            return self.monotone.db_migrate()

    def _do_netsync(self, output):
        return self.monotone.pull(self.server_addr, self.branch)

    def _get_changes(self, output):
        d = self._get_new_head()
        d.addCallback(self._process_new_head)
        return d

    def _get_new_head(self):
        # This function returns a deferred that resolves to a good pick of new
        # head (or None if there is no good new head.)

        # First need to get all new heads...
        rcfile = """function get_revision_cert_trust(signers, id, name, val)
                      local trusted_signers = { %s }
                      local ts_table = {}
                      for k, v in pairs(trusted_signers) do ts_table[v] = 1 end
                      for k, v in pairs(signers) do
                        if ts_table[v] then
                          return true
                        end
                      end
                      return false
                    end
        """
        trusted_list = ", ".join(['"' + key + '"' for key in self.trusted_keys])
        # mktemp is unsafe, but mkstemp is not 2.2 compatible.
        tmpfile_name = tempfile.mktemp()
        f = open(tmpfile_name, "w")
        f.write(rcfile % trusted_list)
        f.close()
        d = self.monotone.get_heads(self.branch, tmpfile_name)
        d.addCallback(self._find_new_head, tmpfile_name)
        return d

    def _find_new_head(self, new_heads, tmpfile_name):
        os.unlink(tmpfile_name)
        # Now get the old head's descendents...
        if self.last_revision is not None:
            d = self.monotone.descendents(self.last_revision)
        else:
            d = defer.succeed(new_heads)
        d.addCallback(self._pick_new_head, new_heads)
        return d

    def _pick_new_head(self, old_head_descendents, new_heads):
        for r in new_heads:
            if r in old_head_descendents:
                return r
        return None

    def _process_new_head(self, new_head):
        if new_head is None:
            log.msg("No new head")
            self.d = None
            return None
        # Okay, we have a new head; we need to get all the revisions since
        # then and create change objects for them.
        # Step 1: simplify set of processed revisions.
        d = self._simplify_revisions()
        # Step 2: get the list of new revisions
        d.addCallback(self._get_new_revisions, new_head)
        # Step 3: add a change for each
        d.addCallback(self._add_changes_for_revisions)
        # Step 4: all done
        d.addCallback(self._finish_changes, new_head)
        return d

    def _simplify_revisions(self):
        d = self.monotone.erase_ancestors(self.done_revisions)
        d.addCallback(self._reset_done_revisions)
        return d

    def _reset_done_revisions(self, new_done_revisions):
        self.done_revisions = new_done_revisions
        return None

    def _get_new_revisions(self, blah, new_head):
        if self.done_revisions:
            return self.monotone.ancestry_difference(new_head,
                                                     self.done_revisions)
        else:
            # Don't force feed the builder with every change since the
            # beginning of time when it's first started up.
            return defer.succeed([new_head])

    def _add_changes_for_revisions(self, revs):
        d = defer.succeed(None)
        for rid in revs:
            d.addCallback(self._add_change_for_revision, rid)
        return d

    def _add_change_for_revision(self, blah, rid):
        d = self.monotone.log(rid, 1)
        d.addCallback(self._add_change_from_log, rid)
        return d

    def _add_change_from_log(self, log, rid):
        d = self.monotone.get_revision(rid)
        d.addCallback(self._add_change_from_log_and_revision, log, rid)
        return d

    def _add_change_from_log_and_revision(self, revision, log, rid):
        # Stupid way to pull out everything inside quotes (which currently
        # uniquely identifies filenames inside a changeset).
        pieces = revision.split('"')
        files = []
        for i in range(len(pieces)):
            if (i % 2) == 1:
                files.append(pieces[i])
        # Also pull out author key and date
        author = "unknown author"
        pieces = log.split('\n')
        for p in pieces:
            if p.startswith("Author:"):
                author = p.split()[1]
        self.parent.addChange(Change(author, files, log, revision=rid))

    def _finish_changes(self, blah, new_head):
        self.done_revisions.append(new_head)
        self.last_revision = new_head
        self.d = None