mirror of
https://github.com/MobileGL-Dev/MobileGL
synced 2026-09-08 04:08:32 +09:00
541 lines
18 KiB
Python
541 lines
18 KiB
Python
#!/usr/bin/env python
|
|
"""Summarise any number of GL CTS suites and MobileGL backends.
|
|
|
|
Each repeatable suite specification consists of four values: backend, label,
|
|
caselist, and result directory. For example::
|
|
|
|
python cts_multi_report.py \
|
|
--suite DirectGLES gl30 gl30-main.txt runs/gles/gl30 \
|
|
--suite DirectVulkan gl30 gl30-main.txt runs/vulkan/gl30 \
|
|
--markdown cts-summary.md --json cts-summary.json
|
|
|
|
A compact comma form is accepted as well::
|
|
|
|
--suite=DirectGLES,gl31,gl31-main.txt,runs/gles/gl31
|
|
|
|
Per-suite parsing and validation deliberately delegate to
|
|
``cts_matrix_report`` so QPA ordering, sidecar upgrades, accepted statuses,
|
|
unrun handling, and duplicate-result semantics cannot drift between reports.
|
|
All conformance rates use unique expected caselist cases as their denominator.
|
|
Backend subtotals and the overall total are therefore case-weighted, not an
|
|
unweighted average of suite percentages.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
from collections import Counter
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from typing import Iterable, Optional, Sequence
|
|
|
|
try: # Direct script execution and package imports are both supported.
|
|
from . import cts_matrix_report
|
|
except ImportError: # pragma: no cover - covered through CLI-style tests
|
|
import cts_matrix_report
|
|
|
|
|
|
SUPPORTED_BACKENDS = ("DirectGLES", "DirectVulkan")
|
|
SUM_FIELDS = (
|
|
"expected",
|
|
"result",
|
|
"pass",
|
|
"accepted",
|
|
"crash",
|
|
"hang",
|
|
"unrun",
|
|
"duplicate",
|
|
)
|
|
|
|
|
|
class MultiReportInputError(ValueError):
|
|
"""Suite specifications cannot produce an unambiguous report."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SuiteSpec:
|
|
backend: str
|
|
label: str
|
|
caselist: str
|
|
result_dir: str
|
|
|
|
@property
|
|
def suite_id(self) -> str:
|
|
return f"{self.backend}/{self.label}"
|
|
|
|
|
|
def _ratio(numerator: int, denominator: int) -> float:
|
|
return numerator / denominator if denominator else 0.0
|
|
|
|
|
|
def _validation_state(items: Sequence[dict]) -> str:
|
|
states = {item["validation"]["state"] for item in items}
|
|
if "ERROR" in states:
|
|
return "ERROR"
|
|
if "INCOMPLETE" in states:
|
|
return "INCOMPLETE"
|
|
return "OK"
|
|
|
|
|
|
def aggregate_reports(items: Sequence[dict], suite_state_keys: Sequence[str]) -> dict:
|
|
"""Return an expected-case-weighted aggregate for suite reports."""
|
|
|
|
if len(items) != len(suite_state_keys):
|
|
raise MultiReportInputError("internal suite/state key count mismatch")
|
|
|
|
totals = {
|
|
field: sum(int(item[field]) for item in items)
|
|
for field in SUM_FIELDS
|
|
}
|
|
status_counts: Counter[str] = Counter()
|
|
for item in items:
|
|
status_counts.update(item["counts"])
|
|
|
|
state = _validation_state(items)
|
|
expected = totals["expected"]
|
|
result = totals["result"]
|
|
suite_states = {
|
|
key: item["validation"]["state"]
|
|
for key, item in zip(suite_state_keys, items)
|
|
}
|
|
return {
|
|
**totals,
|
|
"suite_count": len(items),
|
|
"counts": dict(sorted(status_counts.items())),
|
|
"aggregation": "weighted_by_expected_cases",
|
|
"coverage": {
|
|
"numerator": result,
|
|
"denominator": expected,
|
|
"rate": _ratio(result, expected),
|
|
},
|
|
"rates": {
|
|
"denominator": "expected",
|
|
"strict_pass_only": _ratio(totals["pass"], expected),
|
|
"conformance_accepted": _ratio(totals["accepted"], expected),
|
|
"measured_only_strict_pass": _ratio(totals["pass"], result),
|
|
"measured_only_conformance_accepted": _ratio(
|
|
totals["accepted"], result
|
|
),
|
|
},
|
|
# Keep the convenient aliases used by cts_matrix_report consumers.
|
|
"strict_pass_rate": _ratio(totals["pass"], expected),
|
|
"conformance_accepted_rate": _ratio(totals["accepted"], expected),
|
|
"validation": {
|
|
"state": state,
|
|
"ok": state == "OK",
|
|
"suite_states": suite_states,
|
|
"invariant_expected_equals_result_plus_unrun": (
|
|
expected == result + totals["unrun"]
|
|
),
|
|
},
|
|
}
|
|
|
|
|
|
def _caselist_fingerprint(path: str) -> tuple[str, int]:
|
|
cases, _duplicates = cts_matrix_report.read_caselist(path)
|
|
payload = "\n".join(cases).encode("utf-8") + b"\n"
|
|
return hashlib.sha256(payload).hexdigest(), len(cases)
|
|
|
|
|
|
def _read_provenance(
|
|
spec: SuiteSpec,
|
|
require_run_state: bool,
|
|
expected_run_identity: Optional[str],
|
|
) -> dict:
|
|
path = os.path.join(spec.result_dir, "run_state.json")
|
|
if not os.path.isfile(path):
|
|
if require_run_state or expected_run_identity is not None:
|
|
raise MultiReportInputError(
|
|
"suite result directory has no run_state.json; invocation provenance "
|
|
f"cannot be verified: {spec.result_dir}"
|
|
)
|
|
return {"state": "UNVERIFIED", "run_state": None}
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as handle:
|
|
state = json.load(handle)
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise MultiReportInputError(f"cannot read suite run identity {path}: {exc}") from exc
|
|
if not isinstance(state, dict):
|
|
raise MultiReportInputError(f"suite run identity must be a JSON object: {path}")
|
|
if state.get("backend") != spec.backend:
|
|
raise MultiReportInputError(
|
|
f"suite {spec.suite_id} is labelled {spec.backend}, but run_state.json "
|
|
f"records {state.get('backend')!r}"
|
|
)
|
|
fingerprint, case_count = _caselist_fingerprint(spec.caselist)
|
|
if state.get("caselist_sha256") != fingerprint or state.get("case_count") != case_count:
|
|
raise MultiReportInputError(
|
|
f"suite {spec.suite_id} run_state.json belongs to a different caselist"
|
|
)
|
|
invocation_identity = state.get("invocation_identity")
|
|
if (
|
|
expected_run_identity is not None
|
|
and invocation_identity != expected_run_identity
|
|
):
|
|
raise MultiReportInputError(
|
|
f"suite {spec.suite_id} run_state.json belongs to a different CTS invocation"
|
|
)
|
|
return {
|
|
"state": "VERIFIED",
|
|
"run_state": os.path.abspath(path),
|
|
"invocation_identity": invocation_identity,
|
|
}
|
|
|
|
|
|
def _validate_specs(
|
|
specs: Sequence[SuiteSpec],
|
|
require_run_state: bool,
|
|
expected_run_identity: Optional[str],
|
|
) -> dict[str, dict]:
|
|
if not specs:
|
|
raise MultiReportInputError("at least one --suite specification is required")
|
|
|
|
seen: set[tuple[str, str]] = set()
|
|
seen_result_dirs: list[tuple[str, str]] = []
|
|
provenance: dict[str, dict] = {}
|
|
for spec in specs:
|
|
if spec.backend not in SUPPORTED_BACKENDS:
|
|
raise MultiReportInputError(
|
|
f"unsupported backend {spec.backend!r}; expected one of "
|
|
+ ", ".join(SUPPORTED_BACKENDS)
|
|
)
|
|
if not spec.label.strip():
|
|
raise MultiReportInputError("suite label cannot be empty")
|
|
identity = (spec.backend, spec.label)
|
|
if identity in seen:
|
|
raise MultiReportInputError(
|
|
f"duplicate suite specification for {spec.backend}/{spec.label}"
|
|
)
|
|
seen.add(identity)
|
|
if not os.path.isdir(spec.result_dir):
|
|
raise MultiReportInputError(
|
|
f"suite result directory does not exist: {spec.result_dir}"
|
|
)
|
|
physical_result_dir = os.path.normcase(
|
|
os.path.realpath(os.path.abspath(spec.result_dir))
|
|
)
|
|
for previous_dir, previous_suite in seen_result_dirs:
|
|
try:
|
|
common_dir = os.path.commonpath(
|
|
[previous_dir, physical_result_dir]
|
|
)
|
|
except ValueError:
|
|
continue
|
|
if common_dir in (previous_dir, physical_result_dir):
|
|
raise MultiReportInputError(
|
|
f"suite {spec.suite_id} uses a result directory which overlaps "
|
|
f"{previous_suite}: {spec.result_dir}"
|
|
)
|
|
seen_result_dirs.append((physical_result_dir, spec.suite_id))
|
|
provenance[spec.suite_id] = _read_provenance(
|
|
spec, require_run_state, expected_run_identity
|
|
)
|
|
return provenance
|
|
|
|
|
|
def build_report(
|
|
specs: Sequence[SuiteSpec],
|
|
require_run_state: bool = True,
|
|
expected_run_identity: Optional[str] = None,
|
|
) -> dict:
|
|
"""Build suite, per-backend, and overall serialisable reports."""
|
|
|
|
provenance = _validate_specs(
|
|
specs, require_run_state, expected_run_identity
|
|
)
|
|
suite_reports: list[dict] = []
|
|
backend_order: list[str] = []
|
|
|
|
for spec in specs:
|
|
if spec.backend not in backend_order:
|
|
backend_order.append(spec.backend)
|
|
item = cts_matrix_report.build_version_report(
|
|
spec.label, spec.caselist, [spec.result_dir]
|
|
)
|
|
# ``version`` is the generic label argument in build_version_report;
|
|
# expose explicit multi-report terminology while retaining all of its
|
|
# validation and case-level evidence.
|
|
item.pop("version", None)
|
|
item["backend"] = spec.backend
|
|
item["label"] = spec.label
|
|
item["suite_id"] = spec.suite_id
|
|
item["provenance"] = provenance[spec.suite_id]
|
|
if item["provenance"]["state"] == "UNVERIFIED":
|
|
item["validation"]["warnings"].append(
|
|
"result directory has no run_state.json; backend provenance is unverified"
|
|
)
|
|
suite_reports.append(item)
|
|
|
|
backends: dict[str, dict] = {}
|
|
for backend in backend_order:
|
|
backend_items = [
|
|
item for item in suite_reports if item["backend"] == backend
|
|
]
|
|
labels = [item["label"] for item in backend_items]
|
|
aggregate = aggregate_reports(backend_items, labels)
|
|
aggregate["backend"] = backend
|
|
aggregate["suite_labels"] = labels
|
|
backends[backend] = aggregate
|
|
|
|
overall = aggregate_reports(
|
|
suite_reports, [item["suite_id"] for item in suite_reports]
|
|
)
|
|
overall["backend_count"] = len(backends)
|
|
overall["backends"] = backend_order
|
|
|
|
return {
|
|
"schema_version": 1,
|
|
"generated_at": datetime.now(timezone.utc).isoformat(),
|
|
"accepted_statuses": list(cts_matrix_report.ACCEPTED_STATUSES),
|
|
"rate_policy": {
|
|
"denominator": "unique expected cases from each suite caselist",
|
|
"unrun_cases": "included in the denominator and never accepted",
|
|
"backend_aggregation": "weighted by expected cases",
|
|
"overall_aggregation": "weighted by expected cases across backend-suite pairs",
|
|
"duplicate_results": "last QPA result wins, matching qpa_report.py",
|
|
},
|
|
"suites": suite_reports,
|
|
"backends": backends,
|
|
"overall": overall,
|
|
}
|
|
|
|
|
|
def _percent(numerator: int, denominator: int) -> str:
|
|
if not denominator:
|
|
return "n/a"
|
|
return f"{100.0 * numerator / denominator:.2f}% ({numerator}/{denominator})"
|
|
|
|
|
|
def _markdown_cell(value: object) -> str:
|
|
return str(value).replace("|", r"\|").replace("\r", " ").replace("\n", " ")
|
|
|
|
|
|
def _table_row(backend: str, label: str, item: dict, bold: bool = False) -> str:
|
|
values = [
|
|
backend,
|
|
label,
|
|
str(item["expected"]),
|
|
str(item["result"]),
|
|
str(item["pass"]),
|
|
str(item["accepted"]),
|
|
str(item["crash"]),
|
|
str(item["hang"]),
|
|
str(item["unrun"]),
|
|
str(item["duplicate"]),
|
|
_percent(item["result"], item["expected"]),
|
|
_percent(item["pass"], item["expected"]),
|
|
_percent(item["accepted"], item["expected"]),
|
|
item["validation"]["state"],
|
|
]
|
|
values = [_markdown_cell(value) for value in values]
|
|
if bold:
|
|
values = [f"**{value}**" for value in values]
|
|
return "| " + " | ".join(values) + " |"
|
|
|
|
|
|
def render_markdown(report: dict) -> str:
|
|
lines = [
|
|
"# GL CTS multi-suite conformance report",
|
|
"",
|
|
(
|
|
"| Backend | Suite | Expected | Result | Pass | Accepted | Crash | Hang | "
|
|
"Unrun | Duplicate | Coverage | Strict Pass-only | Conformance-accepted | Validation |"
|
|
),
|
|
"|---|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|:---:|",
|
|
]
|
|
|
|
for backend in report["backends"]:
|
|
for item in report["suites"]:
|
|
if item["backend"] == backend:
|
|
lines.append(_table_row(backend, item["label"], item))
|
|
subtotal = report["backends"][backend]
|
|
lines.append(
|
|
_table_row(backend, f"{backend} weighted subtotal", subtotal, bold=True)
|
|
)
|
|
|
|
lines.append(
|
|
_table_row(
|
|
"All backends",
|
|
"Overall weighted",
|
|
report["overall"],
|
|
bold=True,
|
|
)
|
|
)
|
|
lines.extend(
|
|
[
|
|
"",
|
|
(
|
|
"Rates use unique **Expected** caselist cases as the denominator. "
|
|
"Unrun cases remain in the denominator and are not accepted."
|
|
),
|
|
"Accepted statuses: "
|
|
+ ", ".join(
|
|
f"`{status}`" for status in report["accepted_statuses"]
|
|
)
|
|
+ ".",
|
|
(
|
|
"Duplicate is the number of extra QPA observations; the final "
|
|
"observation wins."
|
|
),
|
|
]
|
|
)
|
|
|
|
details: list[str] = []
|
|
for item in report["suites"]:
|
|
validation = item["validation"]
|
|
messages = validation["errors"] + validation["warnings"]
|
|
if messages:
|
|
details.append(
|
|
f"- **{_markdown_cell(item['suite_id'])} {validation['state']}**: "
|
|
+ "; ".join(_markdown_cell(message) for message in messages)
|
|
)
|
|
if details:
|
|
lines.extend(["", "## Validation details", "", *details])
|
|
return "\n".join(lines) + "\n"
|
|
|
|
|
|
def _write_text(path: str, contents: str) -> None:
|
|
absolute = os.path.abspath(path)
|
|
os.makedirs(os.path.dirname(absolute), exist_ok=True)
|
|
with open(absolute, "w", encoding="utf-8", newline="\n") as handle:
|
|
handle.write(contents)
|
|
|
|
|
|
def _write_json(path: str, report: dict) -> None:
|
|
_write_text(path, json.dumps(report, indent=2, sort_keys=True) + "\n")
|
|
|
|
|
|
def _normalise_compact_suite_args(argv: Sequence[str]) -> list[str]:
|
|
"""Expand ``--suite=b,l,c,r`` into the four-value argparse form."""
|
|
|
|
result: list[str] = []
|
|
index = 0
|
|
while index < len(argv):
|
|
token = argv[index]
|
|
if token.startswith("--suite="):
|
|
compact = token.split("=", 1)[1]
|
|
parts = compact.split(",", 3)
|
|
if len(parts) != 4:
|
|
raise MultiReportInputError(
|
|
"compact --suite expects backend,label,caselist,result-dir"
|
|
)
|
|
result.extend(["--suite", *parts])
|
|
index += 1
|
|
continue
|
|
if token == "--suite" and index + 1 < len(argv) and argv[index + 1].count(",") >= 3:
|
|
parts = argv[index + 1].split(",", 3)
|
|
result.extend(["--suite", *parts])
|
|
index += 2
|
|
continue
|
|
result.append(token)
|
|
index += 1
|
|
return result
|
|
|
|
|
|
def _parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--suite",
|
|
action="append",
|
|
nargs=4,
|
|
required=True,
|
|
metavar=("BACKEND", "LABEL", "CASELIST", "RESULT_DIR"),
|
|
help=(
|
|
"suite specification; repeat for every backend/suite pair "
|
|
f"(backends: {', '.join(SUPPORTED_BACKENDS)})"
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--markdown",
|
|
default="cts_multi_report.md",
|
|
help="Markdown output path (default: ./cts_multi_report.md)",
|
|
)
|
|
parser.add_argument(
|
|
"--json",
|
|
dest="json_out",
|
|
default="cts_multi_report.json",
|
|
help="JSON output path (default: ./cts_multi_report.json)",
|
|
)
|
|
parser.add_argument(
|
|
"--allow-incomplete",
|
|
action="store_true",
|
|
help="return success even when one or more suites are ERROR/INCOMPLETE",
|
|
)
|
|
parser.add_argument(
|
|
"--adopt-legacy",
|
|
dest="allow_unverified_provenance",
|
|
action="store_true",
|
|
help="accept legacy result directories without run_state.json (provenance remains unverified)",
|
|
)
|
|
parser.add_argument(
|
|
"--allow-unverified-provenance",
|
|
dest="allow_unverified_provenance",
|
|
action="store_true",
|
|
help=argparse.SUPPRESS,
|
|
)
|
|
parser.add_argument(
|
|
"--expected-run-identity",
|
|
help="require every suite run_state.json to contain this controller fingerprint",
|
|
)
|
|
return parser
|
|
|
|
|
|
def _specs_from_args(values: Sequence[Sequence[str]]) -> list[SuiteSpec]:
|
|
return [
|
|
SuiteSpec(
|
|
backend=backend.strip(),
|
|
label=label.strip(),
|
|
caselist=caselist,
|
|
result_dir=result_dir,
|
|
)
|
|
for backend, label, caselist, result_dir in values
|
|
]
|
|
|
|
|
|
def main(argv: Optional[Iterable[str]] = None) -> int:
|
|
raw_argv = list(argv) if argv is not None else sys.argv[1:]
|
|
try:
|
|
normalised = _normalise_compact_suite_args(raw_argv)
|
|
except MultiReportInputError as exc:
|
|
print(f"cts_multi_report: {exc}", file=sys.stderr)
|
|
return 2
|
|
args = _parser().parse_args(normalised)
|
|
|
|
if os.path.normcase(os.path.abspath(args.markdown)) == os.path.normcase(
|
|
os.path.abspath(args.json_out)
|
|
):
|
|
print("cts_multi_report: Markdown and JSON paths must differ", file=sys.stderr)
|
|
return 2
|
|
|
|
try:
|
|
report = build_report(
|
|
_specs_from_args(args.suite),
|
|
require_run_state=not args.allow_unverified_provenance,
|
|
expected_run_identity=args.expected_run_identity,
|
|
)
|
|
markdown = render_markdown(report)
|
|
_write_text(args.markdown, markdown)
|
|
_write_json(args.json_out, report)
|
|
except (
|
|
OSError,
|
|
MultiReportInputError,
|
|
cts_matrix_report.ReportInputError,
|
|
) as exc:
|
|
print(f"cts_multi_report: {exc}", file=sys.stderr)
|
|
return 2
|
|
|
|
print(markdown, end="")
|
|
print(f"\nMarkdown: {os.path.abspath(args.markdown)}")
|
|
print(f"JSON: {os.path.abspath(args.json_out)}")
|
|
if not args.allow_incomplete and not report["overall"]["validation"]["ok"]:
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|