tests-trigger

#!/usr/bin/env python3

import argparse
import subprocess
import sys

from lib import ALLOWLIST, testmap
from lib.aio.jsonutil import get_str
from task import github

sys.dont_write_bytecode = True


def git(*args: str) -> str:
    return subprocess.check_output(('git', *args), encoding='utf-8').strip()


def trigger_pull(api: github.GitHub, opts: argparse.Namespace) -> int:
    if opts.target != '-':
        pull = api.get("pulls/" + opts.target)

        if not pull:
            sys.stderr.write(f"{opts.target} is not a pull request.\n")
            return 1

        # triggering is manual, so don't prevent triggering a user that does not have push access
        # nor isn't in the 'Contributors' group, but issue a warning in case of an oversight
        login = pull["head"]["user"]["login"]
        if not opts.allow and login not in ALLOWLIST:
            sys.stderr.write(f"Pull request author '{login}' isn't allowed. Override with --allow.\n")
            return 1

        revision = pull['head']['sha']
        target = f'pull request {opts.target}'
    else:
        revision = git('rev-parse', '@{upstream}')
        target = git('rev-parse', '--abbrev-ref', '@{upstream}')

    statuses = api.statuses(revision)
    if opts.context:
        contexts = set[str]()
        for cntx in opts.context:
            if cntx.startswith("image:"):
                contexts.update(testmap.tests_for_image(cntx.split(':', 1)[1]))
            elif testmap.is_valid_context(cntx, api.repo):
                if opts.bots_pr:
                    cntx += "@bots#" + opts.bots_pr
                contexts.add(cntx)
            else:
                sys.stderr.write(f"ignoring unknown context {cntx}\n")
        everything = False
    else:
        contexts = set(statuses.keys())
        everything = True

    ret = 0
    for context in contexts:
        status = statuses.get(context, {})
        current_status = status.get("state", "unknown" if everything else "empty")

        if current_status not in ["empty", "error", "failure"]:
            # allow changing if manual testing required
            # otherwise "pending" state indicates that testing is in progress
            manual_testing = current_status == "pending" and status.get("description", None) == github.NO_TESTING
            queued = current_status == "pending" and status.get("description", None) == github.NOT_TESTED_DIRECT
            # also allow override with --force or --requeue
            if not (manual_testing or opts.force) and not (queued and opts.requeue):
                if not everything:
                    sys.stderr.write(f"{context}: isn't in triggerable state (is: {status['state']})\n")
                    ret = 1
                continue
        sys.stderr.write(f"{context}: triggering on {target}\n")
        if opts.dry_run:
            continue

        changes = {"state": "pending", "description": github.NOT_TESTED_DIRECT, "context": context}

        # Keep the old link for reference, until testing starts again
        if link := get_str(status, "target_url", None):
            changes["target_url"] = link

        api.post("statuses/" + revision, changes)

    return ret


def main() -> int:
    parser = argparse.ArgumentParser(description='Manually trigger CI Robots')
    parser.add_argument('-f', '--force', action="store_true",
                        help="Force setting the status even if the program logic thinks it shouldn't be done")
    parser.add_argument('-a', '--allow', action='store_true', dest='allow',
                        help="Allow triggering for users that aren't allowed")
    parser.add_argument('--requeue', action="store_true",
                        help='Re-queue pending test requests (workaround for occasionally ignored webhook events)')
    parser.add_argument('--repo', help="The repository to trigger the robots in", default=None)
    parser.add_argument('--bots-pr', help="Use the bots code from this PR instead of main")
    parser.add_argument('-n', '--dry-run', action="store_true",
                        help="Only show which contexts would be triggered")
    parser.add_argument('target', help='The pull request number to trigger, '
                                       'or - for the upstream of the current branch')
    parser.add_argument('context', nargs='*', help='The github task context(s) to trigger; can also be '
                                                   '"image:<imagename>" to trigger all tests related to that image')
    opts = parser.parse_args()

    api = github.GitHub(repo=opts.repo)
    return trigger_pull(api, opts)


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