applyotron

#! /usr/bin/env python3

# Copyright (C) 2015 Ross Burton <ross.burton@intel.com>
# MIT licensed

import email, subprocess, os, sys

class PatchSource:
      def enumerate(self):
            """
            Return a summary of the patches to be applied as a list of strings.
            """
            pass

      def apply(self):
            """
            Apply the patches, raising an exception if there is an error.
            """
            pass

class ClipboardPatch(PatchSource):
      def __init__(self):
            import tempfile

            # Save the clipboard contents for passing to git-am in apply()
            # TODO: respect platform, eg pbcopy for osx
            self.mail = subprocess.check_output(["xsel", "-b"])

            # Check the clipboard contains something that looks like a patch.
            # As the clipboard contains an email it could be encoded, so use
            # git-mailinfo to extract the patch itself first.  It's a shame
            # git-am doesn't have a --dry-run option.
            with tempfile.NamedTemporaryFile(prefix="applyotron") as patchfile:
                  process = subprocess.Popen(["git", "mailinfo", "/dev/null", patchfile.name],
                                             stdin=subprocess.PIPE, stdout=subprocess.PIPE)
                  process.communicate(self.mail)
                  retcode = process.wait()
                  if retcode:
                        raise subprocess.CalledProcessError(retcode, "git-mailinfo")

                  process = subprocess.check_call(["git", "apply", "--3way", "--check", patchfile.name])

      def enumerate(self):
            return (email.message_from_bytes(self.mail)["Subject"],)

      def apply(self):
            gitam = subprocess.Popen(["git", "am", "--3way", "--signoff", "--whitespace=nowarn"],
                                  stdin=subprocess.PIPE)
            gitam.communicate(self.mail)
            retcode = gitam.poll()
            if retcode:
                  raise subprocess.CalledProcessError(retcode, "git")


class FilePatch(PatchSource):
      def __init__(self, files):
            self.files = files

      def enumerate(self):
            return (email.message_from_file(open(f))["Subject"] for f in self.files)

      def apply(self):
            subprocess.check_call(["git", "am", "--3way", "--signoff", "--whitespace=nowarn"] + self.files)


class RevisionsPatch(PatchSource):
      def __init__(self, revisions):
            self.revs = revisions

      def enumerate(self):
            output = subprocess.check_output(["git", "show", "--no-patch", "--format=format:%s"] + self.revs)
            return output.decode("utf-8").split("\n")

      def apply(self):
            subprocess.check_call(["git", "cherry-pick", "--signoff"] + self.revs)


class GMailPatch(PatchSource):
      def __init__(self):
            import collections

            server = self.connect()

            # TODO Configure this in some way.
            # l:^ss_co means "has:orange-guillemet"
            messages = server.gmail_search("in:inbox l:^ss_co")
            # TODO abort if no messages found

            # List of (msgid, mail body) pairs
            self.patches = collections.OrderedDict()
            response = server.fetch(messages, ['ENVELOPE RFC822'])
            for msgid, data in sorted(response.items(), key=lambda d: d[1][b'ENVELOPE'].date):
                  self.patches[msgid] = data[b'RFC822']

      def connect(self):
            import configparser, imapclient

            cp = configparser.ConfigParser()
            cp.read(os.path.expanduser("~/.config/handlepatches.conf"))

            server = imapclient.IMAPClient("imap.gmail.com", ssl=True, use_uid=True)
            server.login(cp.get("Config", "IMAPUSer"), cp.get("Config", "IMAPPassword"))
            server.select_folder("INBOX")
            return server

      def enumerate(self):
            return (email.message_from_bytes(mail)["Subject"] for mail in self.patches.values())

      def apply(self):
            # Apply the patches
            for mail in self.patches.values():
                  # TODO: verify that the subject contains PATCH?
                  gitam = subprocess.Popen(['git', 'am', '--3way', '--signoff', '--whitespace=nowarn'],
                                           stdin=subprocess.PIPE)
                  gitam.communicate(mail)
                  retcode = gitam.wait()
                  if retcode:
                        raise subprocess.CalledProcessError(retcode, 'git')

            # If they all applied, clear the flags
            print("Updating mail...")
            import imapclient
            server = self.connect()
            msgids = list(self.patches.keys())
            server.add_flags(msgids, imapclient.SEEN)
            server.remove_flags(msgids, imapclient.FLAGGED)
            server.remove_gmail_labels(msgids, ("Patches/Queued",))
            server.delete_messages(msgids)
            server.close_folder()
            server.logout()

def quiet_call(cmd):
    """
    Calls a command, reading both stdout and stderr.  If success no output is
    visible, if error then an exception is thrown with the full output.
    """
    subprocess.check_output(cmd, stderr=subprocess.STDOUT)

import shutil
if shutil.which("wipe-sysroot"):
    def wipe_sysroot():
        quiet_call(["wipe-sysroot"])
else:
    def wipe_sysroot():
        pass

def build(targets):
    subprocess.check_call(["bitbake"] + targets)

def buildhistory():
    print("Comparing buildhistory...")
    subprocess.check_call(["buildhistory-diff"])

def checkout(branch):
    quiet_call(["git", "checkout", branch])

def apply_on_branch(patch):
    """
    Create a new branch with the patches on and return the name of this branch,
    whilst remaining on HEAD.
    """
    import random
    # Not pretty but mostly works
    branchname = "applyotron-%04d" % random.randint(0,9999)

    # TODO: have a finally here to abort the apply and delete working branch
    head = subprocess.check_output(["git", "symbolic-ref", "--short", "HEAD"]).decode().strip()
    quiet_call(["git", "checkout", "-b", branchname])
    patch.apply()
    checkout(head)
    return head, branchname

def merge(head, wip):
    checkout(head)
    quiet_call(["git", "merge", wip])
    quiet_call(["git", "branch", "--delete", wip])

def main():
      import argparse

      args = argparse.ArgumentParser(description="Applyotron")
      args.add_argument("-l", "--list", action="store_true", help="List patches found, but don't build or apply")
      group = args.add_mutually_exclusive_group(required=True)
      group.add_argument("-c", "--clipboard", action="store_true", help="Take patch from clipboard")
      group.add_argument("-f", "--file", action="append", help="Take patch from file")
      group.add_argument("-r", "--revisions", action="append", help="Cherry-pick revision range")
      group.add_argument("-g", "--gmail", action="store_true", help="Take patches from GMail")
      args.add_argument("target", nargs="*", help="targets to build")
      args = args.parse_args()

      if args.list and args.target:
            print("You specified --list and some targets to build, but I can't do both.")
            sys.exit(os.EX_USAGE)

      if args.clipboard:
            patch = ClipboardPatch()
      elif args.file:
            patch = FilePatch(args.file)
      elif args.revisions:
            patch = RevisionsPatch(args.revisions)
      elif args.gmail:
            patch = GMailPatch()
      else:
            print("No patch source selected")
            sys.exit(os.EX_USAGE)

      patches = list(patch.enumerate())
      if patches:
            print("Applying the following %d patches:" % (len(patches)))
            for p in patches:
                print(" %s" % p)
      else:
            print("No patches found to apply.")
            sys.exit(os.EX_NOINPUT)

      if args.list:
            # And we're done
            sys.exit(os.EX_OK)

      if args.target:
            # TODO Exception handling. If I control-C applyotron the
            # childs keep on running. This is bad.
            targets = args.target

            head, wip = apply_on_branch(patch)

            # Verify that the targets we've got actually build
            print("Checking targets are buildable...")
            build(["--dry-run", "--quiet"] + targets)

            checkout(head)
            wipe_sysroot()
            build(targets)

            checkout(wip)
            wipe_sysroot()
            build(targets)

            merge(head, wip)
            buildhistory()
      else:
            patch.apply()


if __name__ == "__main__":
      main()