openqa-trigger-bisect-jobs

#!/usr/bin/env python3

import argparse
from functools import total_ordering
import hashlib
import json
import logging
import os
import re
import subprocess
import sys
from urllib.parse import urlparse, urlunparse

import requests

USER_AGENT = 'openqa-trigger-bisect-jobs (https://github.com/os-autoinst/scripts)'

logging.basicConfig()
log = logging.getLogger(sys.argv[0] if __name__ == "__main__" else __name__)
GOOD = "-"
BAD = "+"


class CustomFormatter(
    argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescriptionHelpFormatter
):
    """Preserve multi-line __doc__ and provide default arguments in help strings."""

    pass


@total_ordering
class Incident:
    def __init__(self, inc: str) -> None:
        self.incident = inc
        self._incident_id = None

    @property
    def incident_id(self):
        if self._incident_id:
            return self._incident_id

        try:
            self._incident_id = self.incident.split("/")[6]
        except IndexError:
            self._incident_id = self.incident

        return self._incident_id

    def __str__(self):
        return self.incident

    def __eq__(self, __o) -> bool:
        return self.incident_id == __o.incident_id

    def __gt__(self, __o) -> bool:
        return int(self.incident_id) > int(__o.incident_id)

    def __hash__(self) -> int:
        return int(hashlib.md5(self.incident.encode()).hexdigest(), base=16)

    def __repr__(self) -> str:
        return f"<Incident -> {self.incident}"


def parse_args():
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=CustomFormatter
    )
    parser.add_argument(
        "-v",
        "--verbose",
        help="Increase verbosity level, specify multiple times to increase verbosity",
        action="count",
        default=1,
    )
    parser.add_argument(
        "--url",
        required=True,
        help="The openQA test URL for which to trigger bisection investigation jobs",
    )
    parser.add_argument(
        "--priority-add",
        default=100,
        help="Adds the specified value to the cloned job's priority value",
    )
    parser.add_argument(
        "--dry-run", action="store_true", help="Do not do any action on openQA"
    )
    args = parser.parse_args()
    verbose_to_log = {
        0: logging.CRITICAL,
        1: logging.ERROR,
        2: logging.WARN,
        3: logging.INFO,
        4: logging.DEBUG,
    }
    logging_level = logging.DEBUG if args.verbose > 4 else verbose_to_log[args.verbose]
    log.setLevel(logging_level)
    return args


client_args = [
    "openqa-cli",
    "api",
    "--header",
    f"User-Agent: {USER_AGENT}",
]


def call(cmds, dry_run=False):
    log.debug("call: %s" % cmds)
    return subprocess.check_output(
        (["echo", "Simulating: "] if dry_run else []) + cmds
    ).decode("utf-8")


def openqa_comment(job, host, comment, dry_run):
    args = client_args + [
        "--host",
        host,
        "-X",
        "POST",
        "jobs/" + str(job) + "/comments",
        "text=" + comment,
    ]
    return call(args, dry_run)

def openqa_set_job_prio(job_id, host, prio, dry_run):
    prio_json = json.dumps({"priority": prio})
    args = client_args + [
        "--host",
        host,
        "--json",
        "--data",
        prio_json,
        "-X",
        "PUT",
        "jobs/" + str(job_id),
    ]
    return call(args, dry_run)

def openqa_clone(
    cmds,
    dry_run,
    default_opts=["--skip-chained-deps", "--json-output", "--within-instance"],
    default_cmds=["_GROUP=0"],
):
    return call(["openqa-clone-job"] + default_opts + cmds + default_cmds, dry_run)


def fetch_url(url, request_type="text"):
    try:
        content = requests.get(url, headers={'User-Agent': USER_AGENT})
        content.raise_for_status()
    except requests.exceptions.RequestException as e:
        log.error("Error while fetching %s: %s" % (url, str(e)))
        raise (e)
    raw = content.content
    if request_type == "json":
        try:
            content = content.json()
        except json.decoder.JSONDecodeError as e:
            log.error(
                "Error while decoding JSON from %s -> >>%s<<: %s"
                % (url, raw, str(e))
            )
            raise (e)
    return content


def find_changed_issues(investigation):
    changes = {}
    pattern = re.compile(
        r"(?P<diff>[+-])\s+\"(?P<key>[A-Z]+_TEST_(?:ISSUES|REPOS))\"\s*:\s*\"(?P<var>[^\"]*)\","
    )

    for line in investigation.splitlines():
        search = pattern.match(line)
        if search:
            issue_var = search.group("key")
            if not changes.get(issue_var):
                changes[issue_var] = {}
            changes[issue_var][search.group("diff")] = {
                Incident(i) for i in search.group("var").split(",")
            }

    for key in list(changes):
        if not changes[key].get(BAD) or not changes[key].get(GOOD):
            del changes[key]
            continue
        if len(changes[key][BAD]) <= 1:
            # no value in triggering single-incident bisections
            del changes[key]
            continue

    changes_repos = {
        key: value for key, value in changes.items() if key.endswith("REPOS")
    }

    return changes_repos if changes_repos else changes


def main(args):
    parsed_url = urlparse(args.url)
    base_url = urlunparse((parsed_url.scheme, parsed_url.netloc, "", "", "", ""))
    job_id = parsed_url.path.lstrip("/tests/")
    test_url = f"{base_url}/api/v1/jobs/{job_id}"
    log.debug("Retrieving job data from %s" % test_url)
    test_data = fetch_url(test_url, request_type="json")
    job = test_data["job"]
    if job['result'] == 'passed':
        log.info(
            "Job %d (%s) is passed, skipping bisection"
            % (job["id"], job["test"])
        )
        return
    search = re.search(":investigate:", job["test"])
    if search:
        log.info(
            "Job %d (%s) is already an investigation, skipping bisection"
            % (job["id"], job["test"])
        )
        return
    if job.get("clone_id") is not None:
        log.info("Job %d already has a clone, skipping bisection" % job["id"])
        return

    children = job["children"] if "children" in job else []
    parents = job["parents"] if "parents" in job else []
    if (
        "Parallel" in children
        and len(children["Parallel"])
        or "Directly chained" in children
        and len(children["Directly chained"])
        or "Parallel" in parents
        and len(parents["Parallel"])
        or "Directly chained" in parents
        and len(parents["Directly chained"])
    ):
        return

    investigation_url = f"{base_url}/tests/{job_id}/investigation_ajax"
    log.debug("Retrieving investigation info from %s" % investigation_url)
    investigation = fetch_url(investigation_url, request_type="json")
    log.debug("Received investigation info: %s" % investigation)
    if "diff_to_last_good" not in investigation:
        return
    all_changes = find_changed_issues(investigation["diff_to_last_good"])

    if not all_changes:
        return

    exclude_group_regex = os.environ.get("exclude_group_regex", "")
    if len(exclude_group_regex) > 0:
        full_group = job.get("group", "")
        if "parent_group" in job:
            full_group = "%s / %s" % (job["parent_group"], full_group)
        if re.search(exclude_group_regex, full_group):
            return

    log.debug("Received job data: %s" % test_data)
    test = job["settings"]["TEST"]
    prio = int(job["priority"]) + args.priority_add
    log.debug("Found test name '%s'" % test)

    created = ""
    added = []
    for key in all_changes:
        changes = all_changes[key]
        removed_key, added_key = list(changes[GOOD] - changes[BAD]), list(
            changes[BAD] - changes[GOOD]
        )
        log.debug("[%s] removed: %s, added: %s" % (key, removed_key, added_key))
        added += added_key

    # whole sort is to simplify testability of code
    for issue in sorted(list({i.incident_id for i in added}), key=int):
        line = {}
        log.info("Triggering one bisection job without issue '%s'" % issue)
        for key in all_changes:
            # use only VARS where is incident present in BAD
            if [i for i in all_changes[key][BAD] if i.incident_id == issue]:
                line[key] = ",".join(
                    str(i)
                    for i in sorted(list(all_changes[key][BAD]))
                    if i.incident_id != issue
                )
                log.debug("New set of %s='%s'" % (key, line[key]))

        test_name = test + ":investigate:bisect_without_%s" % issue
        params = (
            [args.url]
            + [k + "=" + v for k, v in line.items()]
            + [
                "TEST=" + test_name,
                "OPENQA_INVESTIGATE_ORIGIN=" + args.url,
                "MAINT_TEST_REPO=",
            ]
        )

        out = openqa_clone(
            params,
            args.dry_run,
        )
        created_job_ids = []
        try:
            created_job_ids = json.loads(out).values()
        except Exception as e:
            log.error("openqa-clone-job returned non-JSON output: " + out)
        for job_id in sorted(created_job_ids):
            log.info(f"Created {job_id}")
            created += f"* **{test_name}**: {base_url}/t{job_id}\n"
            openqa_set_job_prio(job_id, args.url, prio, args.dry_run)

    if len(created):
        comment = "Automatic bisect jobs:\n\n" + created
        openqa_comment(job["id"], base_url, comment, args.dry_run)


if __name__ == "__main__":
    main(parse_args())