tests-trigger

#!/usr/bin/env python3

import argparse
import sys

sys.dont_write_bytecode = True

from task import github

def trigger_pull(api, opts):
    pull = api.get("pulls/" + opts.pull)

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

    # triggering is manual, so don't prevent triggering a user that isn't on the whitelist
    # but issue a warning in case of an oversight
    login = pull["head"]["user"]["login"]
    if not opts.allow and not login in api.whitelist():
        sys.stderr.write("Pull request author '{0}' isn't whitelisted. Override with --allow.\n".format(login))
        return 1

    revision = pull['head']['sha']
    statuses = api.statuses(revision)
    if opts.context:
        contexts = [cntx for cntx in opts.context if github.known_context(cntx)]
        all = False
    else:
        contexts = set(statuses.keys())
        all = True

    ret = 0
    for context in contexts:
        status = statuses.get(context, { })
        current_status = status.get("state", all and "unknown" or "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 all:
                    sys.stderr.write("{0}: isn't in triggerable state (is: {1})\n".format(context, status["state"]))
                    ret = 1
                continue
        sys.stderr.write("{0}: triggering on pull request {1}\n".format(context, opts.pull))
        changes = { "state": "pending", "description": github.NOT_TESTED_DIRECT, "context": context }

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

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

    return ret

def main():
    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 whitelisted")
    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('pull', help='The pull request to trigger')
    parser.add_argument('context', nargs='*', help='The github task context(s) to trigger')
    opts = parser.parse_args()

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

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