patchew-cli

#!/usr/bin/env python3
#
# Copyright 2016 Red Hat, Inc.
#
# Authors:
#     Fam Zheng <famz@redhat.com>
#
# This work is licensed under the MIT License.  Please see the LICENSE file or
# http://opensource.org/licenses/MIT.

import logging
import os
import sys
import argparse
import configparser
import json
import urllib.request, urllib.parse, urllib.error
import subprocess
import tempfile
import shutil
import traceback
import datetime
import time
import hashlib
import fcntl

TOKEN_FILENAME = os.path.expanduser("~/.patchew.token")


class APIError(Exception):
    pass


def git_head():
    a = str(os.environ.get("PATCHEW_TAG"))
    if a:
        if a.startswith("tags/"):
            a = a[len("tags/") :]
        elif a.startswith("refs/tags/"):
            a = a[len("refs/tags/") :]
    return a or subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip()


def git_fetch_into_cache(remote, head, logf=sys.stdout):
    cache_repo = os.path.join(
        os.path.expanduser("~/.cache/patchew-git-cache"),
        "".join([x for x in remote if x.isalnum()])
        + "-"
        + hashlib.sha1(remote.encode("utf-8")).hexdigest(),
    )
    if not os.path.isdir(cache_repo) or not os.listdir(cache_repo):
        # Clone upstream to local cache
        subprocess.check_call(["mkdir", "-p", cache_repo])
        subprocess.check_output(["git", "init", "--bare", cache_repo])
    remote_name = hashlib.sha1(remote.encode("utf-8")).hexdigest()
    subprocess.call(
        ["git", "remote", "remove", remote_name],
        cwd=cache_repo,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    subprocess.check_call(
        ["git", "remote", "add", remote_name, remote],
        cwd=cache_repo,
        stdout=logf,
        stderr=logf,
    )
    subprocess.check_call(
        [
            "git",
            "config",
            "remote.%s.fetch" % (remote_name,),
            "+refs/heads/*:refs/heads/*",
        ],
        cwd=cache_repo,
        stdout=logf,
        stderr=logf,
    )
    subprocess.check_call(
        ["git", "fetch", remote_name], cwd=cache_repo, stdout=logf, stderr=logf
    )
    if head.startswith("refs/tags/"):
        subprocess.check_call(
            ["git", "fetch", remote_name, "+%s:%s" % (head, head)],
            cwd=cache_repo,
            stdout=logf,
            stderr=logf,
        )
    return cache_repo


def git_clone_repo(clone, remote, head, logf=sys.stdout):
    cache_repo = git_fetch_into_cache(remote, head, logf)
    clone_cmd = ["git", "clone", "-q", cache_repo, clone]
    subprocess.check_call(clone_cmd, stderr=logf, stdout=logf)
    subprocess.check_call(
        [
            "git",
            "config",
            "user.name",
            "Patchew Applier",
        ],
        cwd=clone,
        stdout=logf,
        stderr=logf,
    )
    subprocess.check_call(
        [
            "git",
            "config",
            "user.email",
            "no-reply@patchew.org",
        ],
        cwd=clone,
        stdout=logf,
        stderr=logf,
    )
    subprocess.check_call(
        ["git", "checkout", head, "-b", "test"], stderr=logf, stdout=logf, cwd=clone
    )


def http_get(url):
    logging.debug("http get: " + url)
    return urllib.request.urlopen(url).read()


class SubCommand:
    """Base class of subcommand"""

    help = ""
    want_argv = False  # Whether the command accepts extra arguments

    def __init__(self):
        self.tokens = None

    def load_api_tokens(self):
        self.tokens = {}
        try:
            with open(TOKEN_FILENAME, "r") as f:
                for line in f:
                    try:
                        server, token = line.split()
                        self.tokens[server] = token
                    except Exception:
                        pass
        except IOError:
            pass

    def save_api_tokens(self):
        if self.tokens is None:
            self.load_api_tokens()

        new_fn = TOKEN_FILENAME + ".new"
        try:
            with open(new_fn, "w") as f:
                for server, token in self.tokens.items():
                    print(server, token, file=f)
        finally:
            try:
                os.rename(new_fn, TOKEN_FILENAME)
                return
            except IOError:
                pass
            try:
                os.remove(new_fn)
            except IOError:
                pass

    def rest_api_do(
        self,
        url_cmd,
        request_method="get",
        query=None,
        content_type=None,
        data=None,
        authenticate=None,
    ):
        if self.tokens is None:
            self.load_api_tokens()
        if authenticate is None:
            authenticate = request_method.lower() != "get"

        logging.debug("API call '%s':" % url_cmd)
        if query:
            logging.debug("query:\n%s" % query)
        if data:
            logging.debug("data:\n%s" % data)
        opener = urllib.request.build_opener()
        if url_cmd.startswith("http"):
            url = url_cmd
        else:
            url = self.base_url + "/api/v1/" + url_cmd + "/"
        if query:
            url = url + "?" + urllib.parse.urlencode(query)
        if data is None:
            post_data = ""
        else:
            post_data = data
        resp = None
        while resp is None:
            req = urllib.request.Request(
                url,
                data=bytes(post_data, encoding="utf-8"),
                method=request_method.upper(),
            )
            if content_type is not None:
                req.add_header("Content-Type", content_type)
            if authenticate and self.base_url in self.tokens:
                req.add_header("Authorization", "Token " + self.tokens[self.base_url])
            try:
                resp = opener.open(req)
            except urllib.error.HTTPError as e:
                # urllib requires confirmation from us before following a 307 with POST
                # do *not* change the method, as that's the preferred thing to do
                # even if it is allowed for 301 and 302 statuses
                if e.code in (301, 302, 307, 308):
                    url = e.headers["location"]
                else:
                    raise e
        respdata = resp.read()
        logging.debug("Server response:\n%s" % (respdata or "<empty>"))
        if respdata:
            r = json.loads(respdata.decode("utf-8"))
        else:
            r = None
        return r

    def do(self, args, argv):
        """Do command"""
        print("Not implemented")


class LoginCommand(SubCommand):
    name = "login"

    def arguments(self, parser):
        parser.add_argument("username", type=str, help="username for login")
        parser.add_argument("password", nargs="?", type=str, help="password for login")

    def do(self, args, argv):
        import getpass

        try:
            if not args.password:
                args.password = getpass.getpass("Password: ")
            if not args.password:
                return 1
            self.load_api_tokens()
            if self.base_url in self.tokens:
                del self.tokens[self.base_url]
            r = self.rest_api_do(
                url_cmd="users/login",
                request_method="post",
                content_type="application/json",
                data=json.dumps({"username": args.username, "password": args.password}),
            )
            self.tokens[self.base_url] = r["key"]
        except Exception as e:
            print("Login failed:", e)
            return 1
        self.save_api_tokens()
        return 0


class LogoutCommand(SubCommand):
    name = "logout"

    def do(self, args, argv):
        self.rest_api_do(url_cmd="users/logout", request_method="post")
        return 0


class DeleteCommand(SubCommand):
    name = "delete"
    want_argv = True

    def arguments(self, parser):
        parser.add_argument(
            "--all", "-a", action="store_true", help="Delete all messages"
        )

    def do(self, args, argv):
        if not argv and not args.all:
            print("Must specify --all to delete all patches")
            return 1
        if argv and args.all:
            print("--all cannot be specified together with search terms")
            return 1
        series_list = []
        query = {"q": " ".join(argv)} if len(argv) else None
        series_resp = self.rest_api_do("series", query=query)
        series_list = series_resp["results"]
        for series in series_list:
            url = series["resource_uri"]
            self.rest_api_do(url, request_method="delete")
        return 0


class ImportCommand(SubCommand):
    name = "import"
    want_argv = True

    def arguments(self, parser):
        parser.add_argument("file", nargs="*", type=str, help="object to list")
        parser.add_argument(
            "--known-flag-dir",
            "-k",
            type=str,
            help="""a directory to store "known flag files"
                            to mark messages as 'imported' and skip them next time""",
        )

    def do(self, args, argv):
        projects = set()
        import mailbox, email

        def call_import(mo):
            ff = None
            if args.known_flag_dir:
                ff = os.path.join(
                    args.known_flag_dir,
                    hashlib.sha1(mo["Message-ID"].encode("utf-8")).hexdigest(),
                )
                if os.path.exists(ff):
                    print("[OLD] " + mo["Subject"])
                    return
            print("[NEW] " + mo["Subject"])
            for mbox in [mo.as_string()]:
                r = self.rest_api_do(
                    url_cmd="messages",
                    request_method="post",
                    content_type="message/rfc822",
                    data=mbox,
                )
                projects_list = [
                    x["resource_uri"].split("messages")[0] for x in r["results"]
                ]
                for p in projects_list:
                    if p not in projects:
                        projects.add(p)
                        print(p)
            if ff:
                open(ff, "wb").close()

        def import_one(fn):
            if os.path.isdir(fn):
                for p in os.listdir(fn):
                    import_one(os.path.join(fn, p))
                return
            f = open(fn, "rb")
            if f.readline().startswith(b"From "):
                for m in mailbox.mbox(fn):
                    try:
                        call_import(m)
                    except KeyboardInterrupt:
                        raise
                    except Exception as e:
                        print("Error in importing:", m["subject"], str(e))
            else:
                f.seek(0)
                mo = email.message_from_bytes(f.read())
                call_import(mo)

        r = 0
        for f in args.file:
            try:
                import_one(f)
                if len(projects) == 0:
                    print(
                        "The message was not imported to any project. Perhaps you're not logged in as an importer or maintainer"
                    )
                    r = 1
            except:
                print("Error in importing:", f)
                traceback.print_exc()
                r = 1
                pass
        return r


class ProjectCommand(SubCommand):
    name = "project"
    want_argv = True

    def list_projects(self, argv):
        parser = argparse.ArgumentParser()
        parser.add_argument("--raw", action="store_true", help="Show raw json string")
        parser.add_argument(
            "--verbose", "-v", action="store_true", help="Show details about projects"
        )
        args = parser.parse_args(argv)
        r = self.rest_api_do(url_cmd="projects", request_method="get")
        if args.raw:
            print(json.dumps(r, indent=2, separators=",:"))
            return 0
        for p in r["results"]:
            print(p["name"])
            if args.verbose:
                for k, v in p.items():
                    if k == "name":
                        continue
                    print("  %s: %s" % (k, v))
        return 0

    def show_project(self, argv):
        parser = argparse.ArgumentParser()
        parser.add_argument("name", nargs="+", help="The name of project to show info")
        args = parser.parse_args(argv)
        r = self.rest_api_do(url_cmd="projects", request_method="get")
        items = ["mailing_list", "prefix_tags", "url", "git", "description", "logo"]
        for p in r["results"]:
            if not p["name"] in args.name:
                continue
            if len(args.name) > 1:
                print(p["name"])
            for k in items:
                print("  %s: %s" % (k, p[k]))
        return 0

    def add_project(self, argv):
        parser = argparse.ArgumentParser()
        parser.add_argument("name", help="Name of the project")
        parser.add_argument("--mailing-list", "-l", default="", help="Mailing list")
        parser.add_argument("--url", "-u", default="", help="Project URL")
        parser.add_argument("--git", "-g", default="", help="Project git repo")
        parser.add_argument(
            "--desc", "-d", default="", help="Project short discription"
        )
        args = parser.parse_args(argv)
        data = {
            "name": args.name,
            "mailing_list": args.mailing_list,
            "url": args.url,
            "git": args.git,
            "description": args.desc,
        }
        self.rest_api_do(
            url_cmd="projects",
            request_method="post",
            content_type="application/json",
            data=json.dumps(data),
        )

    def update_one_project(self, wd, project):
        if not project["git"]:
            logging.info("Project '%s' not configured", project["name"])
            return
        logging.info("Updating project '%s'", project["name"])
        clone = os.path.join("/var/tmp/patchew-project-update", project["name"])
        if " " in project["git"]:
            repo, branch = project["git"].split(" ", 2)
        else:
            repo, branch = project["git"], "master"
        if not os.path.isdir(clone):
            subprocess.check_output(["git", "clone", "--mirror", repo, clone])
        else:
            subprocess.check_call(
                ["git", "remote", "set-url", "origin", repo], cwd=clone
            )
            subprocess.check_call(["git", "fetch", "origin"], cwd=clone)
        new_head = (
            subprocess.check_output(["git", "rev-parse", branch], cwd=clone)
            .decode()
            .strip()
        )
        logging.debug("new head: %s", new_head)
        old_head = None
        try:
            old_head = project["mirror"]["head"].strip()
            if old_head == new_head:
                logging.info("no change since last update")
                return
            commit_range = "%s..%s" % (old_head, new_head)
        except:
            commit_range = new_head

        try:
            # TODO: relax this 100 limit, which probably need an API change.
            # using -100 would count all commits, while "^HEAD~100" counts merge
            # commits as one
            new_commits = (
                subprocess.check_output(
                    ["git", "rev-list", commit_range, "^%s~100" % new_head], cwd=clone
                )
                .decode()
                .splitlines()
            )
        except:
            new_commits = [new_head]
        logging.debug("old head: %s", old_head)
        logging.debug("new commits: \n%s" % "\n".join(["  " + x for x in new_commits]))
        output = subprocess.check_output(
            ["git", "show", "--format=%b"] + new_commits,
            cwd=clone,
        ).decode()
        msgid_trailer = 'Message-Id:'.casefold()
        lore_link_trailer = 'Link: https://lore.kernel.org/r/'.casefold()
        msgids = []
        for x in output.splitlines():
            words = x.split()
            header = x.casefold()
            if header.startswith(msgid_trailer):
                msgids.append(x[11:].strip())
            elif header.startswith(lore_link_trailer):
                msgids.append('%s' % x[32:].rstrip())
        logging.debug("message ids: \n%s" % "\n".join(["  " + x for x in msgids]))
        push_to = project["mirror"].get("pushurl")
        if push_to:
            try:
                subprocess.check_call(
                    ["git", "push", "--force", push_to, "%s:%s" % (new_head, branch)],
                    cwd=clone,
                )
            except Exception as e:
                logging.warn(
                    "Failed to push the new head to patchew mirror: %s", str(e)
                )
        url = project["resource_uri"] + "update_project_head/"
        data = {"old_head": old_head, "new_head": new_head, "message_ids": msgids}
        self.rest_api_do(
            url,
            request_method="post",
            content_type="application/json",
            data=json.dumps(data),
        )

    def update_project(self, argv):
        parser = argparse.ArgumentParser()
        parser.add_argument("--name", "-n", help="Name of the project")
        args = parser.parse_args(argv)
        r = self.rest_api_do(url_cmd="projects", request_method="get")
        projects = [
            p for p in r["results"] if p["name"] == args.name or args.name == None
        ]
        wd = tempfile.mkdtemp()
        logging.debug("TMPDIR: %s", wd)
        try:
            for p in projects:
                self.update_one_project(wd, p)
        finally:
            shutil.rmtree(wd)

    def do(self, args, argv):
        if argv:
            if argv[0] == "add":
                return self.add_project(argv[1:])
            elif argv[0] == "info":
                return self.show_project(argv[1:])
            elif argv[0] == "update":
                return self.update_project(argv[1:])
        return self.list_projects(argv)


class SearchCommand(SubCommand):
    name = "search"

    def arguments(self, parser):
        parser.add_argument(
            "--output", "-o", default="subject", type=str, help="Output fields"
        )
        parser.add_argument(
            "--raw", "-r", action="store_true", help="Output raw response"
        )
        parser.add_argument("--limit", "-l", type=int, help="Limit")
        parser.add_argument("--offset", "-O", type=int, help="Offset")
        parser.add_argument("term", nargs="*", type=str)

    def do(self, args, argv):
        kwargs = {}
        if args.limit:
            kwargs["limit"] = args.limit
        kwargs["offset"] = args.offset or 0
        kwargs["q"] = " ".join(args.term)
        while True:
            r = self.rest_api_do(
                url_cmd="series",
                request_method="get",
                query=kwargs,
            )
            if not r:
                return 0
            if args.raw:
                print(json.dumps(r["results"], indent=2, separators=",:"))
                break
            else:
                for x in r["results"]:
                    for a in args.output.split(","):
                        print(x[a])
                if r["next"] is None or args.limit:
                    break
                kwargs["offset"] += len(r["results"])
        return 0


class TesterCommand(SubCommand):
    name = "tester"
    want_argv = True

    def arguments(self, parser):
        parser.add_argument(
            "--print-capabilities",
            action="store_true",
            help="print capabilities for the current machine",
        )
        parser.add_argument(
            "--singleton",
            "-S",
            action="store_true",
            help="quit if another singleton mode tester is running",
        )
        parser.add_argument(
            "--project",
            "-p",
            required=True,
            help="comma separated project names to run test",
        )
        parser.add_argument(
            "--name",
            "-n",
            help="name of this tester (default is the logged in username)",
        )
        parser.add_argument(
            "--num", "-N", type=int, default=-1, help="max number of tests to run"
        )
        parser.add_argument(
            "--no-clean-up", action="store_true", help="skip cleaning up after finish"
        )
        parser.add_argument(
            "--no-wait", action="store_true", help="don't wait if nothing to test"
        )

    def _make_script(self, wd, name, content):
        filename = os.path.join(wd, name)
        tf = open(filename, "w", encoding="utf-8")
        tf.write(content)
        tf.close()
        subprocess.check_output(["chmod", "+x", filename])
        return filename

    def _refresh_capabilities(self, project):
        wd = tempfile.mkdtemp()
        project = self.rest_api_do("projects/by-name/%s" % project)
        if not project or "testing_probes" not in project:
            return []
        ret = []
        for name, val in project["testing_probes"].items():
            script = self._make_script(wd, "probe", val)
            try:
                if 0 == subprocess.call(
                    script, stdout=subprocess.PIPE, stderr=subprocess.PIPE
                ):
                    ret.append(name)
            except:
                pass
        return ret

    def test_one(self, name, project, no_clean_up, capabilities):
        r = self.rest_api_do(
            url_cmd="projects/by-name/%s/get-test" % project,
            request_method="post",
            content_type="application/json",
            data=json.dumps(
                {"project": project, "tester": name, "capabilities": capabilities}
            ),
        )
        if not r:
            return False
        print("Running test '%s'" % r["test"]["name"])
        wd = tempfile.mkdtemp(prefix="patchew-tester-tmp-", dir="/var/tmp/")
        print("  Workdir:", wd)
        print("  Project:", project)
        print("  Result URI:", str(r["result_uri"]))
        sys.stdout.flush()
        logf = open(
            os.path.join(wd, "log"), "w+", encoding="utf-8", newline="", errors="ignore"
        )
        test_cmd = r["test"]["script"]
        script = r["test"]["script"].strip() + "\n"
        test_script = self._make_script(wd, "run", script)
        for k, v in r["identity"].items():
            logf.write("%s: %s\n" % (k.capitalize(), v))
        logf.write("\n")
        logf.write("=== TEST SCRIPT BEGIN ===\n")
        logf.write(script)
        logf.write("=== TEST SCRIPT END ===\n")
        logf.write("\n")
        logf.flush()
        rc = 1
        is_timeout = False
        try:
            clone = os.path.join(wd, "src")
            git_clone_repo(clone, r["repo"], r["head"], logf)
            base = r["base"]
            if base:
                subprocess.check_call(
                    ["git", "branch", "base", base], cwd=clone, stdout=logf, stderr=logf
                )
                subprocess.check_call(
                    ["git", "log", "--oneline", "%s.." % (base)],
                    cwd=clone,
                    stdout=logf,
                    stderr=logf,
                )
            logf.write("\n")
            logf.write("=== OUTPUT BEGIN ===\n")
            logf.flush()
            start_time = time.time()
            env = os.environ.copy()
            env["PATCHEW_TAG"] = r["head"]
            tp = subprocess.Popen(
                ["/usr/bin/script", "-qefc", test_script + "< /dev/null", "/dev/null"],
                cwd=clone,
                stdin=subprocess.PIPE,
                stdout=logf,
                stderr=logf,
                env=env,
            )
            rc = None
            timeout = r["test"]["timeout"]
            while timeout <= 0 or time.time() - start_time < timeout:
                rc = tp.poll()
                if rc != None:
                    break
                time.sleep(0.1)
            logf.write("=== OUTPUT END ===\n")
            logf.write("\n")
            if rc == None:
                try:
                    tp.terminate()
                    start_time = time.time()
                    while tp.poll() == None and time.time() - start_time < 10:
                        time.sleep(0.1)
                    tp.kill()
                    tp.wait()
                except Exception as e:
                    traceback.print_exc(file=logf)
                logf.write("Abort: command timeout (>%d seconds)" % timeout)
                is_timeout = True
            else:
                logf.write("Test command exited with code: %d" % rc)
            logf.flush()
        except:
            traceback.print_exc(file=logf)
        finally:
            passed = rc == 0
            try:
                try:
                    logf.seek(0)
                    log = logf.read()
                except:
                    log = "N/A. Internal error while reading log file\n"
                print("  Result:", "Passed" if passed else "Failed")
                logging.debug(log)
                max_size = 100000000
                prefixed = False
                orig_log_size = len(log)
                while max_size > 100000:
                    json_data = {
                        "status": "success" if passed else "failure",
                        "data": {"head": r["head"], "is_timeout": is_timeout},
                        "log": str(log),
                    }
                    if name:
                        json_data["data"]["tester"] = name
                    try:
                        self.rest_api_do(
                            url_cmd=r["result_uri"],
                            request_method="put",
                            content_type="application/json",
                            data=json.dumps(json_data),
                        )
                    except urllib.error.HTTPError as e:
                        if e.code != 413:
                            raise e
                        if not prefixed:
                            prefixed = True
                            log = "WARNING: Log truncated!\n\n" + log
                        log = log[:max_size]
                        max_size = max_size // 10
                    else:
                        break
                if prefixed:
                    print(
                        "Log truncated from %d to %d bytes" % (orig_log_size, len(log))
                    )
                logf.close()
            finally:
                if not no_clean_up:
                    shutil.rmtree(wd)
            return True

    def _check_singleton(self, name):
        lockfile = os.path.expanduser("~/.%s.lock" % name)
        self._lockfile = open(lockfile, "w+")
        try:
            fcntl.flock(self._lockfile, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            sys.stderr.write("Cannot lock " + lockfile + "\n")
            sys.exit(1)

    def do(self, args, argv):
        # Make sure git works
        projects = [x.strip() for x in args.project.split(",") if len(x.strip())]
        if not projects:
            raise Exception("No project specified")
        if args.print_capabilities:
            if len(projects) > 1:
                print("Only one project is allowed for --print-capabilities")
                return 1
            caps = self._refresh_capabilities(projects[0])
            for i in caps:
                print(i)
            return 0
        if args.singleton:
            self._check_singleton(args.name or "patchew-tester")
        subprocess.check_output(["git", "version"])
        count = 0
        cap_refresh = 10
        capabilities = {}
        while True:
            progress = False
            for p in projects:
                if count % cap_refresh == 0:
                    for sp in projects:
                        capabilities[sp] = self._refresh_capabilities(sp)
                if self.test_one(
                    args.name, p, args.no_clean_up, capabilities=capabilities[p]
                ):
                    progress = True
                    count += 1
                if count == args.num:
                    return 0
            if not progress:
                if args.no_wait:
                    print("Nothing to test")
                    return 0
                print("No more work, having a rest...")
                time.sleep(60)
        return 0


class GitlabPipelineCheckCommand(SubCommand):
    name = "gitlab-pipeline-check"

    def arguments(self, parser):
        parser.add_argument("--project", "-p", required=True)
        parser.add_argument("--head", "-H", type=str)
        parser.add_argument(
            "--gitlab",
            "-g",
            type=str,
            default="https://gitlab.com/",
            help="gitlab server addr",
        )

    def _find_pipeline(self, pipelines, head):
        candidate = None
        for p in pipelines:
            if head in (p["sha"], p["ref"]):
                if p["status"] == "skipped":
                    candidate = candidate or p
                else:
                    return p
        return candidate

    def print(self, *msgs):
        print("[%s]" % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), *msgs)

    def do(self, args, argv):
        head = args.head or git_head()
        gitlab = args.gitlab
        if gitlab.endswith("/"):
            gitlab = gitlab[:-1]

        project_encoded = urllib.parse.quote(args.project, safe="")
        self.print("Looking up pipeline...")
        p = None
        for page in range(10):
            url = "%s/api/v4/projects/%s/pipelines/?page=%d&per_page=100" % (
                gitlab,
                project_encoded,
                page,
            )
            logging.debug(url)
            resp = json.loads(http_get(url))
            p = self._find_pipeline(resp, head)
            if p:
                break
        if not p:
            raise Exception("Pipeline not found")
        pid = p["id"]
        self.print("Found pipeline %d:\n\n%s\n" % (pid, p["web_url"]))

        self.print("Waiting for pipeline to finish...")
        cnt = 0
        while p["status"] not in [
            "success",
            "failed",
            "canceled",
            "canceled",
            "skipped",
        ]:
            time.sleep(60)
            cnt += 1
            if cnt % 15 == 0:
                self.print("Still waiting...")
            url = "%s/api/v4/projects/%s/pipelines/%d" % (gitlab, project_encoded, pid)
            p = json.loads(http_get(url))
        if p["status"] == "success":
            self.print("Pipeline succeeded")
            return 0
        self.print("Pipeline", p["status"])
        self.print_failed_jobs(gitlab, project_encoded, pid)
        return 1

    def print_failed_jobs(self, gitlab, project_encoded, pid):
        url = "%s/api/v4/projects/%s/pipelines/%d/jobs" % (gitlab, project_encoded, pid)
        p = json.loads(http_get(url))
        logging.debug(p)
        for job in p:
            if job["status"] in "success":
                continue
            self.print(
                "Job '%s' in stage '%s' is %s"
                % (job["name"], job.get("stage"), job["status"])
            )


class ApplyFailedException(Exception):
    pass


class ApplyCommand(SubCommand):
    name = "apply"

    def arguments(self, parser):
        parser.add_argument("term", nargs="*", type=str)
        parser.add_argument(
            "-C", dest="repo", default=".", help="The path to the git repo"
        )
        parser.add_argument(
            "--any",
            "-a",
            action="store_true",
            help="""Apply any applicable series (e.g. the
                            first) even if there are more than one matches""",
        )
        parser.add_argument(
            "--signoff",
            "-s",
            action="store_true",
            help="""Sign off the applied patch""",
        )
        parser.add_argument(
            "--branch",
            "-b",
            help="""Create a branch at current HEAD before
                            applying, with '%%m' replaced with the series'
                            message id""",
        )
        parser.add_argument(
            "--force-branch",
            "-B",
            help="""Create a branch, overwrite even if a branch
                            with the same name already exists, and point to
                            current HEAD before applying, with '%%m' replaced
                            with the series' message id""",
        )
        parser.add_argument(
            "--tag",
            "-t",
            help="""Create a tag after applying, with '%%m'
                            replaced with the series' message id. If a tag with
                            the same name already exists, it will be
                            updated""",
        )
        parser.add_argument(
            "--applier-mode",
            action="store_true",
            help="""Useful for an applier worker of patchew.
                            Clone and checkout a temporary repo, and apply the
                            patch there. Implies --any, conflicts with -C, -b,
                            -B and -t.""",
        )
        parser.add_argument(
            "--applier-target",
            help="""Restricts the applier to destination repositories
                            that start with the content of the argument.""",
        )

    def _get_maintainers(self, repo, fname):
        script = os.path.join(repo, "scripts/get_maintainer.pl")
        if not os.access(script, os.X_OK):
            return []
        try:
            lines = (
                subprocess.check_output(
                    [
                        script,
                        "--noroles",
                        "--norolestats",
                        "--nogit",
                        "--nogit-fallback",
                        fname,
                    ],
                    cwd=repo,
                )
                .decode("utf-8")
                .splitlines()
            )
            # Only return entries in the form of "First Last <email@address.com>",
            # excluding list addresses
            return [x for x in lines if "<" in x and ">" in x]
        except:
            return []

    def _apply(self, repo, s, branch, force_branch, tag, logf=None, signoff=False):
        msgid = s["message_id"]
        bn = None
        if force_branch:
            bn = force_branch
            bo = "-B"
        elif branch:
            bn = branch
            bo = "-b"
        if bn:
            bn = bn.replace("%m", msgid)
            subprocess.check_call(
                ["git", "checkout", bo, bn], cwd=repo, stdout=logf, stderr=logf
            )
        args = ["git", "am", "-m", "-3"]
        if signoff:
            args += ["-s"]
        maintainers = set()
        req = urllib.request.Request(s["mbox_uri"])
        resp = urllib.request.urlopen(req)
        respdata = resp.read()
        patchf = tempfile.NamedTemporaryFile()
        patchf.write(respdata)
        patchf.flush()
        if 0 != subprocess.call(
            args + [patchf.name], cwd=repo, stdout=logf, stderr=logf
        ):
            raise ApplyFailedException()
        elif tag:
            subprocess.check_call(
                ["git", "tag", "-f", tag.replace("%m", msgid)],
                cwd=repo,
                stdout=logf,
                stderr=logf,
            )
        return patchf

    def _push(self, repo, remote, tag, logf, push_options):
        subprocess.check_call(
            ["git", "remote", "add", "push_to", remote],
            cwd=repo,
            stdout=logf,
            stderr=logf,
        )
        push_options_args = ["-o", push_options] if push_options else []
        cmd = (
            ["git", "push", "-f", "push_to"]
            + push_options_args
            + ["%s:refs/tags/%s" % (tag, tag)]
        )
        logging.debug(" ".join(cmd))
        subprocess.check_call(cmd, cwd=repo, stdout=logf, stderr=logf)

    def _applier_mode(self, args):
        applier_get_args = {"limit": 1}
        if args.applier_target:
            applier_get_args["target_repo"] = args.applier_target
        toapply = self.rest_api_do("series/unapplied", query=applier_get_args)
        toapply = toapply["results"]
        if not toapply:
            print("No series need apply")
            return 3
        toapply = toapply[0]
        logf = tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8")
        push_repo = toapply["mirror"]["pushurl"]
        public_repo = toapply["mirror"]["url"]
        wd = tempfile.mkdtemp(dir="/var/tmp")
        message_id = toapply["message_id"]
        series_url = toapply["resource_uri"]
        result_url = toapply["result_uri"]
        push_options = toapply.get("push_options", None)
        data = {}
        try:
            based_on = toapply.get("based_on") or {}
            if based_on.get("repo"):
                remote, head = based_on["repo"], based_on["tag"]
                git_clone_repo(wd, remote, head)
            else:
                project_git = toapply["mirror"]["source"]
                if " " in project_git:
                    remote, head = project_git.split(" ", 2)
                else:
                    remote, head = project_git, "master"
                git_clone_repo(wd, remote, "origin/" + head)
            branch = message_id
            force_branch = None
            tag = "patchew/" + message_id
            data["base"] = (
                subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=wd)
                .decode("utf-8")
                .strip()
            )
            patchf = self._apply(wd, toapply, branch, force_branch, tag, logf)
            maintainers = self._get_maintainers(wd, patchf.name)
            if push_repo:
                self._push(wd, push_repo, tag, logf, push_options)
        except Exception as e:
            if not isinstance(e, ApplyFailedException):
                traceback.print_exc(file=logf)
            logf.seek(0)
            log = logf.read()
            if push_repo:
                log = log.replace(push_repo, public_repo)
            print(log)
            result_json = {"status": "failure", "data": data, "log": log}
            self.rest_api_do(
                url_cmd=result_url,
                request_method="put",
                content_type="application/json",
                data=json.dumps(result_json),
            )
            return 1
        finally:
            shutil.rmtree(wd)
        logf.seek(0)
        log = logf.read()
        if push_repo:
            log = log.replace(push_repo, public_repo)
        json_data = {"maintainers": maintainers}
        self.rest_api_do(
            url_cmd=series_url,
            request_method="patch",
            content_type="application/json",
            data=json.dumps(json_data),
        )
        data["repo"] = public_repo
        data["tag"] = "refs/tags/" + tag
        result_json = {"status": "success", "data": data, "log": log}
        self.rest_api_do(
            url_cmd=result_url,
            request_method="put",
            content_type="application/json",
            data=json.dumps(result_json),
        )
        return 0

    def do(self, args, argv):
        if args.applier_mode:
            if (
                args.repo not in [None, "."]
                or args.branch
                or args.force_branch
                or args.tag
            ):
                print("--applier-mode conflicts with -C, -b, -B and -t")
                return 1
            return self._applier_mode(args)
        if args.applier_target:
            print("--applier-target requires --applier-mode")
        r = self.rest_api_do(
            url_cmd="series", request_method="get", query={"q": " ".join(args.term)}
        )
        r = r["results"]
        if not r:
            print("Nothing to apply")
            return 3
        if len(r) > 1 and not args.any:
            print("More than one series matched:")
            for p in r:
                print(p["project"], p["subject"])
            return 1
        if not r[0]["is_complete"]:
            print("Series not complete")
            return 1

        def mformat(a):
            return a.replace("%m", r[0]["message-id"]) if a else a

        try:
            self._apply(
                args.repo,
                r[0],
                mformat(args.branch),
                mformat(args.force_branch),
                mformat(args.tag),
                signoff=args.signoff,
            )
            return 0
        except:
            traceback.print_exc()
            return 1


def global_args(parser):
    parser.add_argument(
        "-d", "--debug", action="store_true", default=None, help="Enable debug output"
    )
    parser.add_argument(
        "-D",
        "--nodebug",
        action="store_false",
        dest="debug",
        help="Disable debug output",
    )
    parser.add_argument("-s", "--server", help="Base URL for patchew server")
    parser.add_argument(
        "-c",
        "--config",
        help="""Config file to use. Default is to look up
                        .patchew.cfg under current directory and $HOME""",
    )


def args_merge_config(args):
    if not args.config:
        for fp in [".patchew.cfg", os.path.expanduser("~/.patchew.cfg")]:
            if os.path.isfile(fp):
                args.config = fp
                break
    if not args.config:
        return
    cp = configparser.ConfigParser()
    cp.read(args.config)
    if args.debug == None:
        args.debug = cp.getboolean("general", "debug")
    if args.server == None:
        args.server = cp.get("general", "server")


def main():
    parser = argparse.ArgumentParser()
    global_args(parser)
    subparsers = parser.add_subparsers(title="subcommands")
    for c in SubCommand.__subclasses__():
        cmd = c()
        p = subparsers.add_parser(cmd.name, help=cmd.help)
        if hasattr(cmd, "arguments"):
            cmd.arguments(p)
        p.set_defaults(cmdobj=cmd, all=False)
    args, argv = parser.parse_known_args()

    args_merge_config(args)

    if not hasattr(args, "cmdobj"):
        parser.print_usage()
        return 1
    if args.debug:
        logging.basicConfig(level=logging.DEBUG)
        logging.debug("Server: %s", args.server)
    if argv and not args.cmdobj.want_argv:
        raise Exception("Unrecognized arguments:\n" + argv[0])
    base_url = args.server or "https://patchew.org"
    if base_url.endswith("/"):
        base_url = base_url[:-1]
    args.cmdobj.base_url = base_url
    try:
        return args.cmdobj.do(args, argv)
    except KeyboardInterrupt:
        raise


if __name__ == "__main__":
    sys.exit(main())