#!/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")
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
)
PANIC_RE = re.compile(r"(?:^|\n)thread '([^']+)'(?: \(\d+\))? panicked at ")
MEMBER_RE = re.compile(
    r'unimplemented C# API member: ([^\n]+)|NotImplemented \{ csharp_member: "([^"]+)" \}'
)


@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 = RESULT_RE.findall(text)
    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 has_live_credentials() -> bool:
    dotenv = dotenv_values()
    return 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")):
        if re.search(r"\bpending\s*\(", path.read_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))

    by_test = defaultdict(list)
    for case in catalog["tests"]:
        by_test[case["rust_test"]].append(case["id"])
    passed_cases = sorted(
        case_id for name in log.passed_names for case_id in by_test.get(name, [])
    )
    allowed_passes = sorted(baseline["allowed_parity_passes"])
    if passed_cases != allowed_passes:
        raise ValueError(
            f"parity pass baseline changed: expected {allowed_passes}, found {passed_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 = expected_cases - len(allowed_passes) - expected_ignored
    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 = 101, args.log.read_text()
    else:
        return_code, text = run_suite()
    if return_code != 101:
        raise SystemExit(f"controlled suite returned {return_code}, expected Cargo test failure 101")
    live = has_live_credentials()
    audit = parse_log(text)
    verify(audit, live)
    print(
        f"controlled red suite is current: passed={audit.passed}, 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()