MetaCrate/tools/audit_red_suite.py

#!/usr/bin/env python3
"""Run and verify MetaCrate's controlled red test-suite baseline."""

from __future__ import annotations

import argparse
from collections import Counter, defaultdict, deque
from dataclasses import dataclass
import json
import os
from pathlib import Path
import re
import subprocess


ROOT = Path(__file__).resolve().parents[1]
CATALOG = ROOT / "tests" / "upstream-tests.json"
BASELINE = ROOT / "tests" / "red-suite-baseline.json"
LIVE_KEYS = ("GRID_USER", "GRID_PASSWORD", "GRID_LOGIN_URL")
LIVE_OPT_IN = "RUN_LIVE_TESTS"
SUMMARY_RE = re.compile(
    r"test result: (?:ok|FAILED)\. (\d+) passed; (\d+) failed; "
    r"(\d+) ignored; (\d+) measured; (\d+) filtered out"
)
RESULT_RE = re.compile(
    r"^test (.+) \.\.\. (ok|FAILED|ignored)(?:, .*)?$", re.MULTILINE
)
RESULT_START_RE = re.compile(r"^test (.+) \.\.\. ?(.*)$")
RESULT_END_RE = re.compile(r"^(ok|FAILED|ignored)(?:, .*)?$")
PANIC_RE = re.compile(r"(?:^|\n)thread '([^']+)'(?: \(\d+\))? panicked at ")
MEMBER_RE = re.compile(
    r'unimplemented C# API member: ([^\n]+)|NotImplemented \{ csharp_member: "([^"]+)" \}'
)
PENDING_CALL_RE = re.compile(r"(?<![:.\w])pending\s*\(")
PENDING_DEFINITION_RE = re.compile(r"\bfn\s+pending\s*\(")


@dataclass(frozen=True)
class LogAudit:
    passed: int
    failed: int
    ignored: int
    measured: int
    filtered: int
    passed_names: tuple[str, ...]
    failed_names: tuple[str, ...]
    ignored_names: tuple[str, ...]
    members: Counter[str]
    nonstandard_failures: tuple[str, ...]


def parse_log(text: str) -> LogAudit:
    """Parse libtest output and associate each failed test with its member ID."""
    summaries = [tuple(map(int, match)) for match in SUMMARY_RE.findall(text)]
    if not summaries:
        raise ValueError("controlled suite log contains no libtest summaries")
    totals = tuple(sum(row[index] for row in summaries) for index in range(5))
    results: list[tuple[str, str]] = []
    pending_results: deque[str] = deque()
    for line in text.splitlines():
        complete = RESULT_RE.fullmatch(line)
        if complete:
            results.append((complete.group(1), complete.group(2)))
            continue
        start = RESULT_START_RE.match(line)
        if start:
            pending_results.append(start.group(1))
            continue
        end = RESULT_END_RE.fullmatch(line)
        if end and pending_results:
            results.append((pending_results.popleft(), end.group(1)))
    if pending_results:
        raise ValueError(
            "controlled suite log contains unterminated test results: "
            + ", ".join(pending_results)
        )
    names = {
        state: tuple(name for name, result in results if result == state)
        for state in ("ok", "FAILED", "ignored")
    }

    panic_members: dict[str, deque[tuple[str, ...]]] = defaultdict(deque)
    starts = list(PANIC_RE.finditer(text))
    for index, start in enumerate(starts):
        end = starts[index + 1].start() if index + 1 < len(starts) else len(text)
        chunk = text[start.end() : end]
        members = tuple(first or second for first, second in MEMBER_RE.findall(chunk))
        panic_members[start.group(1)].append(members)

    member_counts: Counter[str] = Counter()
    nonstandard: list[str] = []
    for name in names["FAILED"]:
        members = panic_members[name].popleft() if panic_members[name] else ()
        if members:
            member_counts[members[0]] += 1
        else:
            nonstandard.append(name)

    if totals[0] != len(names["ok"]) or totals[1] != len(names["FAILED"]):
        raise ValueError("libtest summaries do not match parsed pass/fail result lines")
    return LogAudit(
        *totals,
        names["ok"],
        names["FAILED"],
        names["ignored"],
        member_counts,
        tuple(nonstandard),
    )


def dotenv_values() -> dict[str, str]:
    values: dict[str, str] = {}
    path = ROOT / ".env"
    if not path.exists():
        return values
    for raw_line in path.read_text().splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        line = line.removeprefix("export ")
        if "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key.strip()] = value.strip().strip("'\"")
    return values


def live_tests_enabled() -> bool:
    opted_in = os.environ.get(LIVE_OPT_IN, "").strip().lower() in {
        "1",
        "true",
        "yes",
    }
    dotenv = dotenv_values()
    return opted_in and all(
        (os.environ.get(key) or dotenv.get(key, "")).strip() for key in LIVE_KEYS
    )


def pending_calls() -> list[str]:
    pending: list[str] = []
    roots = (ROOT / "tests" / "compat" / "tests", ROOT / "crates")
    for path in sorted(file for root in roots for file in root.rglob("*.rs")):
        text = PENDING_DEFINITION_RE.sub("", path.read_text())
        if PENDING_CALL_RE.search(text):
            pending.append(path.relative_to(ROOT).as_posix())
    return pending


def verify(log: LogAudit, live: bool) -> None:
    catalog = json.loads(CATALOG.read_text())
    baseline = json.loads(BASELINE.read_text())
    report = catalog["report"]
    expected_cases = baseline["reviewed_cases"]
    if catalog["upstream_commit"] != baseline["upstream_commit"]:
        raise ValueError("red-suite baseline targets a different upstream commit")
    if catalog["expected_cases"] != expected_cases or sum(report.values()) != expected_cases:
        raise ValueError("parity catalog does not contain the fixed reviewed case count")
    expected_report = baseline["parity_report"]
    if report != expected_report:
        raise ValueError(f"parity report changed: expected {expected_report}, found {report}")
    pending = pending_calls()
    if pending:
        raise ValueError("pending( remains in Rust test sources: " + ", ".join(pending))

    if baseline.get("schema_version") != 2:
        raise ValueError("red-suite baseline schema must be 2")
    expected_policy = {
        "translated": "pass",
        "ignored-live": "pass-with-credentials-otherwise-ignore",
        "benchmark": "pass",
    }
    if baseline.get("parity_expectation") != expected_policy:
        raise ValueError("red-suite parity expectation is not the reviewed green policy")

    by_test = defaultdict(list)
    for case in catalog["tests"]:
        by_test[case["rust_test"]].append(case["id"])
    def parity_cases(names: tuple[str, ...]) -> list[str]:
        case_ids = {
            case_id
            for name in names
            for case_id in by_test.get(name.rsplit("::", 1)[-1], [])
        }
        return sorted(case_ids)

    passed_cases = parity_cases(log.passed_names)
    expected_passes = sorted(
        case["id"]
        for case in catalog["tests"]
        if case["status"] in {"translated", "benchmark"}
        or (live and case["status"] == "ignored-live")
    )
    if passed_cases != expected_passes:
        raise ValueError(
            f"parity pass baseline changed: expected {expected_passes}, found {passed_cases}"
        )
    ignored_cases = parity_cases(log.ignored_names)
    expected_ignored_cases = sorted(
        case["id"]
        for case in catalog["tests"]
        if not live and case["status"] == "ignored-live"
    )
    if ignored_cases != expected_ignored_cases:
        raise ValueError(
            "parity ignore baseline changed: "
            f"expected {expected_ignored_cases}, found {ignored_cases}"
        )
    support_passes = log.passed - len(passed_cases)
    if support_passes != baseline["support_passes"]:
        raise ValueError(
            f"support pass baseline changed: expected {baseline['support_passes']}, "
            f"found {support_passes}"
        )

    expected_ignored = 0 if live else report["ignored_live"]
    expected_failed = 0
    if (log.failed, log.ignored) != (expected_failed, expected_ignored):
        raise ValueError(
            "controlled totals changed: "
            f"expected failed={expected_failed}, ignored={expected_ignored}; "
            f"found failed={log.failed}, ignored={log.ignored}"
        )
    if log.nonstandard_failures:
        raise ValueError(
            "failures without standardized member IDs: "
            + ", ".join(log.nonstandard_failures)
        )
    if sum(log.members.values()) != log.failed:
        raise ValueError("standardized member aggregation does not cover every failure")


def run_suite() -> tuple[int, str]:
    command = [
        "cargo",
        "test",
        "--workspace",
        "--no-fail-fast",
        "--",
        "--nocapture",
        "--test-threads=1",
    ]
    completed = subprocess.run(command, cwd=ROOT, capture_output=True, text=True, check=False)
    return completed.returncode, completed.stdout + completed.stderr


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--log", type=Path, help="audit an existing captured cargo-test log")
    args = parser.parse_args()
    if args.log:
        return_code, text = None, args.log.read_text()
    else:
        return_code, text = run_suite()
    if return_code not in {None, 0, 101}:
        tail = "\n".join(text.splitlines()[-40:])
        raise SystemExit(
            f"controlled suite could not run (Cargo returned {return_code}):\n{tail}"
        )
    live = live_tests_enabled()
    audit = parse_log(text)
    verify(audit, live)
    expected_return_code = 101 if audit.failed else 0
    if return_code is not None and return_code != expected_return_code:
        raise SystemExit(
            f"controlled suite returned {return_code}, expected {expected_return_code}"
        )
    print(
        f"controlled compatibility suite is current: passed={audit.passed}, "
        f"failed={audit.failed}, "
        f"ignored={audit.ignored}, live_credentials={str(live).lower()}, "
        f"standardized_members={len(audit.members)}"
    )
    for member, count in audit.members.most_common():
        print(f"{count:4} {member}")


if __name__ == "__main__":
    main()