tests-status

#!/usr/bin/env python3

# This file is part of Cockpit.
#
# Copyright (C) 2020 Red Hat, Inc.
#
# Cockpit is free software; you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation; either version 2.1 of the License, or
# (at your option) any later version.
#
# Cockpit is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
# Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with Cockpit; If not, see <http://www.gnu.org/licenses/>.

import argparse
import subprocess
import sys
import time
import urllib.parse
import urllib.request
from collections.abc import Mapping

from lib import github
from lib.aio.jsonutil import JsonObject, get_str
from lib.network import host_ssl_context


def print_summary(by_state: dict[str, list[tuple[str, str]]], state: str) -> None:
    tests = by_state[state]
    print("%i tests in state %s: %s" % (
        len(tests),
        state,
        " ".join([t[0] for t in tests])))


def print_failure(context: str, url: str) -> None:
    print(context + ":")
    print("  " + url)
    if url.endswith(".html"):
        url = url[:-5]
    with urllib.request.urlopen(url, context=host_ssl_context(urllib.parse.urlparse(url).netloc)) as f:
        for line in f:
            if line.startswith(b"not ok"):
                print("  " + line.strip().decode())
    print()


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


# returns a dict of state->[(context, url)]
def sort_statuses(statuses: Mapping[str, JsonObject]) -> dict[str, list[tuple[str, str]]]:
    by_context: dict[str, tuple[str, str]] = {}  # context → (state, url)
    for context, status in statuses.items():
        # latest status wins
        if context in by_context:
            continue
        by_context[context] = (get_str(status, "state"), get_str(status, "target_url", ""))

    by_state: dict[str, list[tuple[str, str]]] = {}  # state → [(context, url), ..]
    for context, (state, url) in by_context.items():
        by_state.setdefault(state, []).append((context, url))

    return by_state


def main() -> None:
    parser = argparse.ArgumentParser(description='Summarize test status of a PR')
    parser.add_argument('--wait', action='store_true', help="Wait for all green, or one red", default=None)
    parser.add_argument('--repo', help="The repository of the PR", default=None)
    parser.add_argument('-v', '--verbose', action="store_true", default=False,
                        help="Print verbose information")
    parser.add_argument("target", help='The pull request number to inspect, '
                                       'or - for the upstream of the current branch')
    opts = parser.parse_args()

    api = github.GitHub(repo=opts.repo)

    if opts.target != '-':
        pull = api.get(f"pulls/{opts.target}")
        if not pull:
            sys.exit(f"{opts.target} is not a pull request.")
        revision = pull['head']['sha']
    else:
        revision = git('rev-parse', '@{upstream}')

    while True:
        by_state = sort_statuses(api.statuses(revision))

        if 'pending' not in by_state or 'failure' in by_state or not opts.wait:
            break

        print_summary(by_state, 'pending')
        print('waiting...\n')
        time.sleep(30)

    for state in by_state.keys():
        if state != "failure":
            print_summary(by_state, state)

    failed = by_state.get("failure")
    if not failed:
        return
    print("\nFailed tests\n============\n")
    for (context, url) in failed:
        print_failure(context, url)


if __name__ == '__main__':
    main()