mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-17 16:37:30 +02:00
285 lines
12 KiB
Python
285 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Normalize coverage, JUnit, repeat, and mutation results into retained evidence."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import os
|
|
import platform
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from xml.etree import ElementTree
|
|
|
|
|
|
def _read_object(path: Path, label: str) -> dict[str, Any]:
|
|
try:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
raise ValueError(f"cannot read {label}: {exc}") from exc
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"{label} root must be an object")
|
|
return value
|
|
|
|
|
|
def _percentage(summary: dict[str, Any], covered: str, total: str, path: str) -> float | None:
|
|
covered_value = summary.get(covered)
|
|
total_value = summary.get(total)
|
|
for key, value in ((covered, covered_value), (total, total_value)):
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
raise ValueError(f"coverage report has invalid {key} for {path}")
|
|
if covered_value > total_value:
|
|
raise ValueError(f"coverage report overcounts {covered} for {path}")
|
|
return covered_value / total_value * 100 if total_value else None
|
|
|
|
|
|
def coverage_snapshot(
|
|
report: dict[str, Any],
|
|
risk_map: dict[str, Any] | None = None,
|
|
*,
|
|
allow_missing_risk_modules: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Return repository and risk-module coverage metrics."""
|
|
|
|
totals = report.get("totals")
|
|
files = report.get("files")
|
|
if not isinstance(totals, dict) or not isinstance(files, dict):
|
|
raise ValueError("coverage report requires totals and files objects")
|
|
line = totals.get("percent_statements_covered")
|
|
branch = totals.get("percent_branches_covered")
|
|
for label, value in (("line", line), ("branch", branch)):
|
|
if (
|
|
isinstance(value, bool)
|
|
or not isinstance(value, (int, float))
|
|
or not math.isfinite(value)
|
|
):
|
|
raise ValueError(f"coverage report requires numeric {label} coverage")
|
|
|
|
modules: dict[str, dict[str, float | None]] = {}
|
|
if risk_map is not None:
|
|
risk_modules = risk_map.get("modules")
|
|
if not isinstance(risk_modules, list):
|
|
raise ValueError("test risk map modules must be a list")
|
|
for item in risk_modules:
|
|
path = item.get("path") if isinstance(item, dict) else None
|
|
entry = files.get(path) if isinstance(path, str) else None
|
|
summary = entry.get("summary") if isinstance(entry, dict) else None
|
|
if not isinstance(path, str):
|
|
raise ValueError("test risk map module requires a path")
|
|
if not isinstance(summary, dict) and allow_missing_risk_modules:
|
|
modules[path] = {"line_percent": None, "branch_percent": None}
|
|
continue
|
|
if not isinstance(summary, dict):
|
|
raise ValueError(f"coverage report is missing risk module {path}")
|
|
modules[path] = {
|
|
"line_percent": _percentage(summary, "covered_lines", "num_statements", path),
|
|
"branch_percent": _percentage(
|
|
summary, "covered_branches", "num_branches", path,
|
|
),
|
|
}
|
|
return {
|
|
"line_percent": float(line),
|
|
"branch_percent": float(branch),
|
|
"risk_modules": modules,
|
|
}
|
|
|
|
|
|
def junit_snapshot(path: Path) -> dict[str, Any]:
|
|
"""Return stable counts, node IDs, and slow-test evidence from JUnit XML."""
|
|
|
|
try:
|
|
root = ElementTree.parse(path).getroot()
|
|
except (OSError, ElementTree.ParseError) as exc:
|
|
raise ValueError(f"cannot read JUnit XML: {exc}") from exc
|
|
cases = list(root.iter("testcase"))
|
|
failures: list[str] = []
|
|
errors: list[str] = []
|
|
skipped: list[str] = []
|
|
durations: list[dict[str, Any]] = []
|
|
marker_totals: dict[str, dict[str, int | float]] = {}
|
|
missing_marker_nodeids: list[str] = []
|
|
for case in cases:
|
|
nodeid = f"{case.attrib.get('classname', '<unknown>')}::{case.attrib.get('name', '<unknown>')}"
|
|
try:
|
|
duration = float(case.attrib.get("time", "0"))
|
|
except ValueError as exc:
|
|
raise ValueError(f"JUnit testcase {nodeid} has invalid time") from exc
|
|
if not math.isfinite(duration) or duration < 0:
|
|
raise ValueError(f"JUnit testcase {nodeid} has invalid time")
|
|
marker_property = None
|
|
properties = case.find("properties")
|
|
if properties is not None:
|
|
values = [
|
|
item.attrib.get("value", "")
|
|
for item in properties.findall("property")
|
|
if item.attrib.get("name") == "duration_markers"
|
|
]
|
|
if len(values) > 1:
|
|
raise ValueError(f"JUnit testcase {nodeid} repeats duration_markers")
|
|
marker_property = values[0] if values else None
|
|
if marker_property is None:
|
|
missing_marker_nodeids.append(nodeid)
|
|
markers = sorted({
|
|
marker.strip()
|
|
for marker in (marker_property or "unmarked").split(",")
|
|
if marker.strip()
|
|
})
|
|
durations.append({"nodeid": nodeid, "seconds": duration, "markers": markers})
|
|
for marker in markers:
|
|
summary = marker_totals.setdefault(marker, {"tests": 0, "duration_seconds": 0.0})
|
|
summary["tests"] = int(summary["tests"]) + 1
|
|
summary["duration_seconds"] = float(summary["duration_seconds"]) + duration
|
|
if case.find("failure") is not None:
|
|
failures.append(nodeid)
|
|
if case.find("error") is not None:
|
|
errors.append(nodeid)
|
|
if case.find("skipped") is not None:
|
|
skipped.append(nodeid)
|
|
failed = set(failures) | set(errors) | set(skipped)
|
|
durations.sort(key=lambda item: (-float(item["seconds"]), str(item["nodeid"])))
|
|
suites = list(root.iter("testsuite"))
|
|
raw_suite_duration = suites[0].attrib.get("time") if suites else None
|
|
try:
|
|
suite_duration = (
|
|
float(raw_suite_duration)
|
|
if raw_suite_duration is not None
|
|
else sum(float(item["seconds"]) for item in durations)
|
|
)
|
|
except ValueError as exc:
|
|
raise ValueError("JUnit testsuite has invalid time") from exc
|
|
if not math.isfinite(suite_duration) or suite_duration < 0:
|
|
raise ValueError("JUnit testsuite has invalid time")
|
|
for summary in marker_totals.values():
|
|
summary["duration_seconds"] = round(float(summary["duration_seconds"]), 3)
|
|
return {
|
|
"total": len(cases),
|
|
"passed": len(cases) - len(failed),
|
|
"failures": failures,
|
|
"errors": errors,
|
|
"skipped": skipped,
|
|
"duration_seconds": round(sum(float(item["seconds"]) for item in durations), 3),
|
|
"suite_duration_seconds": round(suite_duration, 3),
|
|
"durations": durations,
|
|
"marker_durations": dict(sorted(marker_totals.items())),
|
|
"marker_metadata_complete": not missing_marker_nodeids,
|
|
"missing_marker_nodeids": missing_marker_nodeids,
|
|
"slowest": durations[:20],
|
|
}
|
|
|
|
|
|
def mutation_snapshot(stats: dict[str, Any]) -> dict[str, int | float]:
|
|
killed = stats.get("killed")
|
|
total = stats.get("total")
|
|
if (
|
|
isinstance(killed, bool)
|
|
or not isinstance(killed, int)
|
|
or killed < 0
|
|
or isinstance(total, bool)
|
|
or not isinstance(total, int)
|
|
or total <= 0
|
|
or killed > total
|
|
):
|
|
raise ValueError("mutation statistics require valid killed and total counts")
|
|
return {"killed": killed, "total": total, "score_percent": killed / total * 100}
|
|
|
|
|
|
def build_evidence(
|
|
*,
|
|
head_sha: str,
|
|
base_sha: str,
|
|
python_version: str,
|
|
coverage: dict[str, Any] | None = None,
|
|
base_coverage: dict[str, Any] | None = None,
|
|
risk_map: dict[str, Any] | None = None,
|
|
junit: dict[str, Any] | None = None,
|
|
repeat: dict[str, Any] | None = None,
|
|
mutation: dict[str, Any] | None = None,
|
|
generated_at: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Build the project-owned evidence schema from already-parsed inputs."""
|
|
|
|
evidence: dict[str, Any] = {
|
|
"schema_version": 1,
|
|
"generated_at": generated_at or datetime.now(timezone.utc).isoformat(),
|
|
"repository": os.environ.get("GITHUB_REPOSITORY", "elder-plinius/OBLITERATUS"),
|
|
"head_sha": head_sha,
|
|
"base_sha": base_sha,
|
|
"python": python_version,
|
|
"platform": platform.platform(),
|
|
}
|
|
if coverage is not None:
|
|
current = coverage_snapshot(coverage, risk_map)
|
|
coverage_evidence: dict[str, Any] = {"current": current}
|
|
if base_coverage is not None:
|
|
base = coverage_snapshot(
|
|
base_coverage,
|
|
risk_map,
|
|
allow_missing_risk_modules=True,
|
|
)
|
|
coverage_evidence["base"] = base
|
|
coverage_evidence["delta"] = {
|
|
"line_percentage_points": current["line_percent"] - base["line_percent"],
|
|
"branch_percentage_points": current["branch_percent"] - base["branch_percent"],
|
|
}
|
|
evidence["coverage"] = coverage_evidence
|
|
if junit is not None:
|
|
evidence["tests"] = junit
|
|
if repeat is not None:
|
|
if repeat.get("schema_version") != 1:
|
|
raise ValueError("repeat evidence schema_version must be 1")
|
|
evidence["repeat"] = repeat
|
|
if mutation is not None:
|
|
evidence["mutation"] = mutation_snapshot(mutation)
|
|
if not any(key in evidence for key in ("coverage", "tests", "repeat", "mutation")):
|
|
raise ValueError("at least one test evidence input is required")
|
|
return evidence
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--coverage", type=Path)
|
|
parser.add_argument("--base-coverage", type=Path)
|
|
parser.add_argument("--junit", type=Path)
|
|
parser.add_argument("--repeat", type=Path)
|
|
parser.add_argument("--mutation", type=Path)
|
|
parser.add_argument("--risk-map", type=Path, default=Path("ci/test-risk-map.json"))
|
|
parser.add_argument("--head-sha", default=os.environ.get("GITHUB_SHA", "local"))
|
|
parser.add_argument("--base-sha", default="")
|
|
parser.add_argument("--python-version", default=platform.python_version())
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args()
|
|
try:
|
|
coverage = _read_object(args.coverage, "coverage report") if args.coverage else None
|
|
base_coverage = (
|
|
_read_object(args.base_coverage, "base coverage report")
|
|
if args.base_coverage else None
|
|
)
|
|
risk_map = _read_object(args.risk_map, "test risk map") if coverage else None
|
|
junit = junit_snapshot(args.junit) if args.junit else None
|
|
repeat = _read_object(args.repeat, "repeat evidence") if args.repeat else None
|
|
mutation = _read_object(args.mutation, "mutation evidence") if args.mutation else None
|
|
evidence = build_evidence(
|
|
head_sha=args.head_sha,
|
|
base_sha=args.base_sha,
|
|
python_version=args.python_version,
|
|
coverage=coverage,
|
|
base_coverage=base_coverage,
|
|
risk_map=risk_map,
|
|
junit=junit,
|
|
repeat=repeat,
|
|
mutation=mutation,
|
|
)
|
|
except ValueError as exc:
|
|
print(f"test evidence failed: {exc}")
|
|
return 1
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(evidence, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
print(f"test evidence written: {args.output}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|