#!/usr/bin/python3
#
# Runner for QEMU unit / qtest / functional / softfloat tests shipped in the
# qemu-kvm-tests RPM. Replays the meson test manifest captured at RPM build
# time (tests-manifest.json) without requiring meson, ninja, or the QEMU
# source tree at runtime.
#
# Manifest format: output of `meson introspect --tests BUILDDIR`. See
# https://mesonbuild.com/IDE-integration.html for the schema. Paths inside
# the manifest are expected to be rewritten at RPM install time to point at
# the installed tree (see qemu-kvm.spec).
#

import argparse
import collections
import concurrent.futures
import json
import os
import pathlib
import re
import signal
import subprocess
import sys
import time

def parse_args():
    script_dir = pathlib.Path(__file__).resolve().parent
    ap = argparse.ArgumentParser(
        description="Run QEMU tests from the installed qemu-kvm-tests tree.",
        formatter_class=lambda prog: argparse.HelpFormatter(prog, width=80))
    ap.add_argument("--manifest",
                    default=str(script_dir / "intro-tests.json"),
                    help="path to meson's intro-tests.json (default: "
                         "alongside this script)")
    ap.add_argument("--default-workdir",
                    default=str(script_dir / "build"),
                    help="cwd for tests whose 'workdir' is null in the "
                         "manifest (default: <script_dir>/build)")
    ap.add_argument("--suite", action="append", default=[],
                    help="run only tests belonging to this suite "
                         "(repeatable). Suite names match either the bare "
                         "form ('qtest') or the 'qemu:qtest' form.")
    ap.add_argument("--no-suite", action="append", default=[],
                    help="skip tests belonging to this suite (repeatable)")
    ap.add_argument("--name", action="append", default=[],
                    help="regex matched against test name; "
                         "repeatable, OR-combined")
    ap.add_argument("--case", action="append", default=[], metavar="PATH",
                    help="run only this GTest case path within each "
                         "selected test (repeatable; passed as -p PATH)")
    ap.add_argument("-j", "--jobs", type=int, default=os.cpu_count() or 1,
                    help="parallel jobs (default: %(default)s)")
    ap.add_argument("--dry-run", action="store_true",
                    help="print each test's command, do not execute")
    ap.add_argument("--verbose", "-v", action="store_true",
                    help="stream each test's stdout/stderr live "
                         "(implies -j 1)")
    ap.add_argument("--list", action="store_true",
                    help="list selected tests and exit")
    ap.add_argument("--list-suites", action="store_true",
                    help="list unique suite names with test counts and exit")
    ap.add_argument("--list-cases", action="store_true",
                    help="for each selected test, list its GTest subcases "
                         "(invoke the test binary with -l) and exit")
    ap.add_argument("--show-cases", action="store_true",
                    help="parse each test's TAP output and print per-subcase "
                         "ok/not ok/SKIP lines after its summary")
    return ap.parse_args()


def load_manifest(path):
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def suite_matches(test_suites, wanted):
    """True if any of the test's suites matches any 'wanted' entry.

    Accepts both 'qemu:qtest' and bare 'qtest' forms.
    """
    if not wanted:
        return True
    bare = {s.split(":", 1)[-1] for s in test_suites}
    return any(w in test_suites or w in bare for w in wanted)


def select_tests(tests, args):
    name_res = [re.compile(p) for p in args.name]
    out = []
    for t in tests:
        if not suite_matches(t["suite"], args.suite):
            continue
        if args.no_suite and suite_matches(t["suite"], args.no_suite):
            continue
        if name_res and not any(r.search(t["name"]) for r in name_res):
            continue
        out.append(t)
    return out


def build_env(test):
    # Seed with the manifest's env, then overlay the shell env so callers can
    # override any var (e.g. QTEST_QEMU_BINARY) just by exporting it.
    env = (test.get("env") or {}).copy()
    env.update(os.environ)
    return env


def classify(rc, protocol):
    """Return ('PASS'|'SKIP'|'FAIL', short_reason)."""
    # TAP "ok but skipped" exit code, per Meson's TAP harness.
    tap_skip_rc = 77

    if protocol == "tap" and rc == tap_skip_rc:
        return "SKIP", "tap skip"
    if rc == 0:
        return "PASS", ""

    return "FAIL", f"exit {rc}"


def _result(name, status, duration, reason="", output=""):
    return {"name": name, "status": status, "duration": duration,
            "reason": reason, "output": output}


def _spawn_and_wait(cmd, env, cwd, timeout, capture):
    """Spawn cmd; return (returncode, output, timed_out, duration).

    Uses Popen + communicate(timeout=...) and kills the whole process group
    on timeout so child QEMUs don't leak.
    """
    stdout = subprocess.PIPE if capture else None
    stderr = subprocess.STDOUT if capture else None
    t0 = time.monotonic()
    with subprocess.Popen(cmd, env=env, cwd=cwd,
                          stdout=stdout, stderr=stderr,
                          start_new_session=True) as proc:
        try:
            out, _ = proc.communicate(timeout=timeout)
            return proc.returncode, out, False, time.monotonic() - t0
        except subprocess.TimeoutExpired:
            try:
                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
            except (ProcessLookupError, PermissionError):
                pass
            out, _ = proc.communicate()
            return -1, out, True, time.monotonic() - t0


def run_one(test, args):
    cmd = list(test["cmd"])
    for case in args.case:
        cmd += ["-p", case]
    # Meson defaults workdir to the build dir; mimic that so tests with
    # relative paths (e.g. test-qga -> "qga/qemu-ga") still resolve.
    cwd = test.get("workdir") or args.default_workdir
    timeout = test.get("timeout") or 60

    if args.dry_run:
        return _result(test["name"], "DRYRUN", 0.0, output=" ".join(cmd))

    try:
        rc, out, timed_out, duration = _spawn_and_wait(
            cmd, build_env(test), cwd, timeout,
            capture=not args.verbose)
    except FileNotFoundError as e:
        return _result(test["name"], "FAIL", 0.0,
                       reason=f"missing binary: {e.filename}")

    output = out.decode(errors="replace") if out else ""
    if timed_out:
        return _result(test["name"], "FAIL", duration,
                       reason=f"timeout after {timeout:.0f}s", output=output)
    status, reason = classify(rc, test["protocol"])
    if status == "FAIL":
        # First failing TAP subcase, if any, helps pinpoint which case broke.
        for st, name, _d, _c in iter_tap_subcases(output):
            if st == "not ok":
                reason += f"; failed: {name}"
                break
    return _result(test["name"], status, duration, reason, output)


def fmt_result(r):
    mark = {"PASS": "ok", "FAIL": "FAIL", "SKIP": "skip",
            "DRYRUN": "dry-run"}[r["status"]]
    line = f"[{mark:>7}] {r['duration']:6.2f}s  {r['name']}"
    if r["reason"]:
        line += f"  ({r['reason']})"
    return line


def iter_tap_subcases(output):
    """Yield (status, name, directive, comment) per TAP test result line.

    The test number after "ok"/"not ok" is normally required by TAP spec but
    glib's harness sometimes omits it on bail-out failures, so it's optional.
    """
    tap_re = re.compile(
        r"^(?P<status>ok|not ok)(?:\s+\d+)?\s*-?\s*(?P<name>\S.*?)"
        r"(?:\s+#\s*(?P<directive>SKIP|TODO)\b\s*(?P<comment>.*))?$",
        re.IGNORECASE,
    )
    for line in output.splitlines():
        m = tap_re.match(line.strip())
        if m:
            yield (m.group("status"), m.group("name"),
                   (m.group("directive") or "").upper(),
                   m.group("comment") or "")


def print_subcases(result):
    """Print one indented line per TAP subcase parsed from result['output']."""
    for status, name, directive, comment in iter_tap_subcases(result["output"]):
        if directive == "SKIP":
            mark = "skip"
        elif status == "ok":
            mark = "ok"
        else:
            mark = "FAIL"
        line = f"         [{mark:>4}]  {name}"
        if directive:
            line += f"  # {directive}"
            if comment:
                line += f" {comment}"
        print(line)


def cmd_list_suites(tests):
    counts = collections.Counter(s for t in tests for s in t["suite"])
    for suite, n in sorted(counts.items()):
        print(f"{n:5d}  {suite}")


def cmd_list(selected):
    width = max(len(t["name"]) for t in selected)
    for t in selected:
        print(f"{t['name']:<{width}}  {', '.join(t['suite'])}")


def cmd_list_cases(selected, args):
    # -l is a GTest convention; only unit/qtest binaries honor it. Other tests
    # (Python functional, fp-test, qapi-schema diff scripts) interpret -l
    # differently or run the full test.
    gtest_suites = frozenset({"qemu:unit", "qemu:qtest", "qemu:qga"})

    for t in selected:
        if not gtest_suites.intersection(t["suite"]):
            print(f"\n=== {t['name']} === (skipped: not a GTest binary)")
            continue
        print(f"\n=== {t['name']} ===")
        env = build_env(t)
        cwd = t.get("workdir") or args.default_workdir
        try:
            # GLib's -l prints "# /case/path" lines but some test binaries
            # also run setup code (cdrom-test -> xorriso, modules-test ->
            # device warnings) and that noise pollutes the output. Keep only
            # the case-path lines.
            proc = subprocess.run(list(t["cmd"]) + ["-l"],
                                  env=env, cwd=cwd, check=False, timeout=10,
                                  stdout=subprocess.PIPE,
                                  stderr=subprocess.STDOUT, text=True)
            for line in proc.stdout.splitlines():
                if line.startswith("# /"):
                    print(line)
        except subprocess.TimeoutExpired:
            print("(timeout; binary does not support -l)")
        except FileNotFoundError as e:
            print(f"missing binary: {e.filename}")


def cmd_run(selected, args):
    print(f"Running {len(selected)} test(s) with -j {args.jobs}", flush=True)
    results = []
    counts = {"PASS": 0, "FAIL": 0, "SKIP": 0, "DRYRUN": 0}
    # tests are subprocesses; ThreadPoolExecutor + subprocess is fine.
    with concurrent.futures.ThreadPoolExecutor(max_workers=args.jobs) as ex:
        futs = {ex.submit(run_one, t, args): t for t in selected}
        for fut in concurrent.futures.as_completed(futs):
            r = fut.result()
            results.append(r)
            counts[r["status"]] += 1
            print(fmt_result(r), flush=True)
            if args.show_cases:
                print_subcases(r)

    failures = [r for r in results if r["status"] == "FAIL"]
    if failures and not args.verbose:
        print("\n=== Failure details ===")
        for r in failures:
            print(f"\n--- {r['name']} ({r['reason']}) ---")
            print("―" * 80)
            print(r["output"].rstrip() or "(no output)")
            print("―" * 80)

    print(f"\nSummary: {counts['PASS']} passed, {counts['FAIL']} failed, "
          f"{counts['SKIP']} skipped, of {len(selected)} run")
    sys.exit(1 if counts["FAIL"] else 0)


def main():
    args = parse_args()
    if args.verbose:
        args.jobs = 1

    try:
        tests = load_manifest(args.manifest)
    except FileNotFoundError:
        sys.exit(f"manifest not found: {args.manifest}")
    except json.JSONDecodeError as e:
        sys.exit(f"manifest {args.manifest} is not valid JSON: {e}")

    if args.list_suites:
        cmd_list_suites(tests)
        return

    selected = select_tests(tests, args)
    if not selected:
        sys.exit("no tests matched filters")

    if args.list:
        cmd_list(selected)
        return

    if args.list_cases:
        cmd_list_cases(selected, args)
        return

    cmd_run(selected, args)


if __name__ == "__main__":
    main()
