#!/usr/bin/env python3 """Create bounded, sanitized capability-baseline evidence candidates.""" from __future__ import annotations import argparse import hashlib import json import re import sys from pathlib import Path from typing import Any MAX_REPORT_BYTES = 4 * 1024 * 1024 REVISION_RE = re.compile(r"^[0-9a-f]{40}$") FLOW_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") ENVIRONMENT_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") COMMAND_IDS = { "python-tooling-tests", "rust-admin-integration", "rust-mcp-integration", "ui-build", "ui-playwright", "just-verify", "authenticated-product-smoke", } class CollectionError(Exception): def __init__(self, code: str, pointer: str) -> None: super().__init__(code) self.code = code self.pointer = pointer def reject_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in pairs: if key in result: raise CollectionError("INVALID_JSON", "/") result[key] = value return result def load_report(path: Path) -> tuple[dict[str, Any], str]: try: with path.open("rb") as file: raw = file.read(MAX_REPORT_BYTES + 1) except OSError as error: raise CollectionError("BROKEN_REPORT_LINK", "/report") from error if len(raw) > MAX_REPORT_BYTES: raise CollectionError("INPUT_TOO_LARGE", "/report") digest = hashlib.sha256(raw).hexdigest() try: value = json.loads(raw.decode("utf-8"), object_pairs_hook=reject_duplicates) except (UnicodeDecodeError, json.JSONDecodeError, ValueError, RecursionError) as error: raise CollectionError("INVALID_JSON", "/report") from error if not isinstance(value, dict): raise CollectionError("INVALID_REPORT", "/report") return value, digest def iter_tests(value: Any, title: str | None = None): if isinstance(value, dict): title = value.get("title") if isinstance(value.get("title"), str) else title tests = value.get("tests") if isinstance(tests, list): for test in tests: if isinstance(test, dict): yield test, title for key in ("suites", "specs"): children = value.get(key) if isinstance(children, list): for child in children: yield from iter_tests(child, title) elif isinstance(value, list): for item in value: yield from iter_tests(item, title) def playwright_verdict(report: dict[str, Any], required_titles: list[str]) -> tuple[str, dict[str, int]]: counts = {"passed": 0, "failed": 0, "flaky": 0, "skipped": 0, "not_run": 0} report_errors = report.get("errors", []) if not isinstance(report_errors, list): counts["failed"] = 1 return "fail", counts if report_errors: counts["failed"] = min(len(report_errors), 10_000) return "fail", counts tests = list(iter_tests(report)) if not tests: counts["not_run"] = 1 return "not_run", counts required = {title: False for title in required_titles} for test, title in tests: status = test.get("status") results = test.get("results") if isinstance(test.get("results"), list) else [] result_statuses = [result.get("status") for result in results if isinstance(result, dict)] retries = [result.get("retry", 0) for result in results if isinstance(result, dict)] if len(result_statuses) != len(results): counts["not_run"] += 1 continue if status == "flaky" or any(isinstance(retry, int) and retry > 0 for retry in retries): counts["flaky"] += 1 elif status == "skipped" or (not results and status in ("skipped", "expected")): counts["skipped"] += 1 elif status in ("unexpected", "failed", "timedOut", "interrupted") or any( result_status in ("failed", "timedOut", "interrupted") for result_status in result_statuses ): counts["failed"] += 1 elif results and all(result_status == "passed" for result_status in result_statuses): counts["passed"] += 1 if title in required: required[title] = True else: counts["not_run"] += 1 if not all(required.values()): counts["failed"] += len([title for title, passed in required.items() if not passed]) if counts["failed"]: return "fail", counts if counts["flaky"]: return "flaky", counts if counts["skipped"]: return "skipped", counts if counts["not_run"]: return "not_run", counts return "pass", counts def validate_labels(args: argparse.Namespace) -> None: if not REVISION_RE.fullmatch(args.source_revision): raise CollectionError("INVALID_ARGUMENT", "/source_revision") if not ENVIRONMENT_RE.fullmatch(args.environment_class): raise CollectionError("INVALID_ARGUMENT", "/environment_class") if not args.flow_id or any(not FLOW_RE.fullmatch(flow_id) for flow_id in args.flow_id): raise CollectionError("INVALID_ARGUMENT", "/flow_ids") def collect_playwright(args: argparse.Namespace) -> dict[str, Any]: validate_labels(args) report, digest = load_report(Path(args.report)) verdict, counts = playwright_verdict(report, args.required_test) return { "accepted": verdict == "pass", "collector": "capability-baseline-collector-v1", "command_id": "ui-playwright", "environment_class": args.environment_class, "evidence_mode": "automated", "execution_verdict": verdict, "flow_ids": sorted(set(args.flow_id)), "id": f"run-ui-playwright-{digest[:12]}", "source_report_sha256": digest, "source_revision": args.source_revision, "summary": counts, } def collect_command_report(args: argparse.Namespace) -> dict[str, Any]: validate_labels(args) report, digest = load_report(Path(args.report)) command_id = report.get("command_id") if command_id not in COMMAND_IDS: raise CollectionError("UNKNOWN_COMMAND", "/report/command_id") exit_code = report.get("exit_code") timed_out = report.get("timed_out") skipped = report.get("skipped") if type(exit_code) is not int or not isinstance(timed_out, bool) or type(skipped) is not int or skipped < 0: raise CollectionError("INVALID_REPORT", "/report") if timed_out: verdict = "blocked" elif exit_code != 0: verdict = "fail" elif skipped: verdict = "skipped" else: verdict = "pass" return { "accepted": verdict == "pass", "collector": "capability-baseline-collector-v1", "command_id": command_id, "environment_class": args.environment_class, "evidence_mode": "automated", "execution_verdict": verdict, "flow_ids": sorted(set(args.flow_id)), "id": f"run-{command_id}-{digest[:12]}", "source_report_sha256": digest, "source_revision": args.source_revision, "summary": {"exit_code": exit_code, "skipped": skipped, "timed_out": timed_out}, } def parser() -> argparse.ArgumentParser: root = argparse.ArgumentParser(description=__doc__) subparsers = root.add_subparsers(dest="mode", required=True) playwright = subparsers.add_parser("playwright") playwright.add_argument("--report", required=True) playwright.add_argument("--output", required=True) playwright.add_argument("--source-revision", required=True) playwright.add_argument("--environment-class", required=True) playwright.add_argument("--flow-id", action="append", required=True) playwright.add_argument("--required-test", action="append", default=[]) command = subparsers.add_parser("command-report") command.add_argument("--report", required=True) command.add_argument("--output", required=True) command.add_argument("--source-revision", required=True) command.add_argument("--environment-class", required=True) command.add_argument("--flow-id", action="append", required=True) return root def main() -> int: args = parser().parse_args() try: candidate = collect_playwright(args) if args.mode == "playwright" else collect_command_report(args) encoded = (json.dumps(candidate, indent=2, sort_keys=True) + "\n").encode("utf-8") if len(encoded) > 65_536: raise CollectionError("OUTPUT_TOO_LARGE", "/output") Path(args.output).write_bytes(encoded) except CollectionError as error: print(f"{error.code} pointer={error.pointer}", file=sys.stderr) return 1 except OSError: print("OUTPUT_WRITE_FAILED pointer=/output", file=sys.stderr) return 1 print(f"capability baseline candidate: verdict={candidate['execution_verdict']}") return 0 if __name__ == "__main__": raise SystemExit(main())