225 lines
8.3 KiB
Python
225 lines
8.3 KiB
Python
#!/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):
|
|
if isinstance(value, dict):
|
|
tests = value.get("tests")
|
|
if isinstance(tests, list):
|
|
for test in tests:
|
|
if isinstance(test, dict):
|
|
yield test
|
|
for key in ("suites", "specs"):
|
|
children = value.get(key)
|
|
if isinstance(children, list):
|
|
for child in children:
|
|
yield from iter_tests(child)
|
|
elif isinstance(value, list):
|
|
for item in value:
|
|
yield from iter_tests(item)
|
|
|
|
|
|
def playwright_verdict(report: dict[str, Any]) -> 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
|
|
for test 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
|
|
else:
|
|
counts["not_run"] += 1
|
|
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)
|
|
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)
|
|
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())
|