#!/usr/bin/env python3
"""Turn the console output of a Connect IQ unit-test run into a verdict.

`monkeydo app.prg <device> -t` prints Garmin's "Run No Evil" report:

    Executing test smokeVersionIsSemver...
    DEBUG (14:16): version 0.1.0
    PASS
    ==============================================================================
    RESULTS
    Test:                                     Status:
    smokeVersionIsSemver                      PASS
    Ran 1 test
    PASSED (failures=0, errors=0)

monkeydo's exit code is reported to be non-zero even when every test passes,
so this script decides instead. A run passes only when all of these hold:
  * the RESULTS section and the final PASSED/FAILED line are present (a crash
    or a watchdog kill leaves the log incomplete);
  * the final line says PASSED with zero failures and zero errors;
  * "Ran N tests" matches the number of per-test rows, and N >= --min-tests;
  * every per-test row says PASS.

Usage:  ciq_test_report.py LOG --device ID [--junit OUT.xml] [--min-tests N]
Exit:   0 pass, 1 test failure, 2 log unreadable or incomplete.
"""

from __future__ import annotations

import argparse
import re
import sys
from dataclasses import dataclass, field
from pathlib import Path
from xml.sax.saxutils import escape, quoteattr

EXIT_PASS, EXIT_FAIL, EXIT_INCOMPLETE = 0, 1, 2

SUMMARY_RE = re.compile(r"^(PASSED|FAILED)\s*\(\s*failures\s*=\s*(\d+)\s*,\s*errors\s*=\s*(\d+)\s*\)", re.I)
RAN_RE = re.compile(r"^Ran\s+(\d+)\s+tests?\b", re.I)
ROW_RE = re.compile(r"^(\S+)\s+(PASS|FAIL|ERROR)\s*$", re.I)
EXECUTING_RE = re.compile(r"^Executing test\s+(\S+?)(?:\.\.\.|…)?\s*$")


@dataclass
class Report:
    """Everything parsed from one test log."""

    results: dict[str, str] = field(default_factory=dict)  # test name -> PASS/FAIL/ERROR
    output: dict[str, list[str]] = field(default_factory=dict)  # test name -> its console lines
    ran: int | None = None
    summary: str | None = None  # PASSED or FAILED
    failures: int = 0
    errors: int = 0

    def problems(self, min_tests: int) -> tuple[int, list[str]]:
        """Return (exit code, reasons) for this report."""
        if self.summary is None or self.ran is None:
            return EXIT_INCOMPLETE, ["log has no complete RESULTS section (crash, timeout or watchdog?)"]
        reasons = []
        if self.summary != "PASSED" or self.failures or self.errors:
            reasons.append(f"summary says {self.summary} (failures={self.failures}, errors={self.errors})")
        if self.ran != len(self.results):
            reasons.append(f"'Ran {self.ran}' but {len(self.results)} result rows")
        if self.ran < min_tests:
            reasons.append(f"only {self.ran} tests ran, expected at least {min_tests}")
        reasons += [f"{name}: {status}" for name, status in self.results.items() if status != "PASS"]
        return (EXIT_FAIL if reasons else EXIT_PASS), reasons


def parse(text: str) -> Report:
    """Parse a Run No Evil console log."""
    report = Report()
    in_results = False
    current: str | None = None
    for raw in text.splitlines():
        line = raw.strip()
        if not in_results:
            match = EXECUTING_RE.match(line)
            if match:
                current = match.group(1)
                report.output[current] = []
            elif line == "RESULTS":
                in_results = True
            elif current is not None and line and not line.startswith(("---", "===")):
                report.output[current].append(raw)
            continue
        if (match := ROW_RE.match(line)) and not line.lower().startswith("test:"):
            report.results[match.group(1)] = match.group(2).upper()
        elif match := RAN_RE.match(line):
            report.ran = int(match.group(1))
        elif match := SUMMARY_RE.match(line):
            report.summary = match.group(1).upper()
            report.failures, report.errors = int(match.group(2)), int(match.group(3))
    return report


def junit_xml(report: Report, device: str) -> str:
    """A minimal JUnit XML document, readable by most CI dashboards."""
    failed = sum(1 for status in report.results.values() if status != "PASS")
    cases = []
    for name, status in report.results.items():
        body = "\n".join(report.output.get(name, []))
        inner = f"<system-out>{escape(body)}</system-out>"
        if status != "PASS":
            inner = f"<failure message={quoteattr(status)}/>" + inner
        cases.append(f"  <testcase classname={quoteattr(device)} name={quoteattr(name)}>{inner}</testcase>")
    return (
        '<?xml version="1.0" encoding="UTF-8"?>\n'
        f'<testsuite name={quoteattr("connectiq." + device)} tests="{len(report.results)}" failures="{failed}">\n'
        + "\n".join(cases)
        + "\n</testsuite>\n"
    )


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Judge a Connect IQ unit-test log.")
    parser.add_argument("log", type=Path)
    parser.add_argument("--device", required=True)
    parser.add_argument("--junit", type=Path)
    parser.add_argument("--min-tests", type=int, default=1)
    args = parser.parse_args(argv)

    try:
        text = args.log.read_text(encoding="utf-8", errors="replace")
    except OSError as err:
        print(f"[{args.device}] cannot read {args.log}: {err}")
        return EXIT_INCOMPLETE

    report = parse(text)
    code, reasons = report.problems(args.min_tests)
    if args.junit:
        args.junit.parent.mkdir(parents=True, exist_ok=True)
        args.junit.write_text(junit_xml(report, args.device), encoding="utf-8")

    passed = sum(1 for status in report.results.values() if status == "PASS")
    verdict = {EXIT_PASS: "PASS", EXIT_FAIL: "FAIL", EXIT_INCOMPLETE: "INCOMPLETE"}[code]
    print(f"[{args.device}] {verdict}: {passed}/{len(report.results)} tests passed")
    for reason in reasons:
        print(f"[{args.device}]   {reason}")
    return code


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