mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-17 16:37:30 +02:00
651 lines
29 KiB
Python
651 lines
29 KiB
Python
#!/usr/bin/env python3
|
|
"""Validate immutable quality floors and mature CPU-scope coverage."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import re
|
|
from datetime import date, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
BASELINE_FLOORS = {
|
|
"repository_statement": 75.0,
|
|
"repository_branch": 60.0,
|
|
"changed_line": 95.0,
|
|
"mature_cpu_statement": 92.0,
|
|
"mature_cpu_branch": 80.0,
|
|
"mutation_score": 85.0,
|
|
"warning_budget": 0.0,
|
|
}
|
|
|
|
SUPPORTED_PYTHON = {"3.10", "3.11", "3.12"}
|
|
|
|
|
|
def _valid_exception(exceptions: Any, name: str, current: float) -> bool:
|
|
if not isinstance(exceptions, list):
|
|
return False
|
|
for exception in exceptions:
|
|
if not isinstance(exception, dict) or exception.get("threshold") != name:
|
|
continue
|
|
return (
|
|
exception.get("new_value") == current
|
|
and isinstance(exception.get("reason"), str)
|
|
and bool(exception["reason"].strip())
|
|
and isinstance(exception.get("approved_issue"), str)
|
|
and exception["approved_issue"].startswith(
|
|
"https://github.com/elder-plinius/OBLITERATUS/issues/",
|
|
)
|
|
and isinstance(exception.get("expires"), str)
|
|
and bool(exception["expires"].strip())
|
|
)
|
|
return False
|
|
|
|
|
|
def _policy_date(value: Any, label: str, failures: list[str]) -> date | None:
|
|
if not isinstance(value, str):
|
|
failures.append(f"{label} must be an ISO date")
|
|
return None
|
|
try:
|
|
return date.fromisoformat(value)
|
|
except ValueError:
|
|
failures.append(f"{label} must be an ISO date")
|
|
return None
|
|
|
|
|
|
def _validate_test_evidence(
|
|
policy: dict[str, Any], *, today: date,
|
|
) -> list[str]:
|
|
failures: list[str] = []
|
|
evidence = policy.get("test_evidence")
|
|
if not isinstance(evidence, dict):
|
|
return ["quality policy is missing the test_evidence object"]
|
|
|
|
numeric: dict[str, int] = {}
|
|
for key, minimum in (
|
|
("retention_days", 30),
|
|
("flake_window_days", 30),
|
|
("maximum_quarantine_days", 1),
|
|
):
|
|
value = evidence.get(key)
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
|
|
failures.append(f"test evidence {key} must be an integer at least {minimum}")
|
|
else:
|
|
numeric[key] = value
|
|
|
|
history = evidence.get("flake_history")
|
|
quarantines = evidence.get("quarantines")
|
|
if not isinstance(history, list):
|
|
failures.append("test evidence flake_history must be a list")
|
|
history = []
|
|
if not isinstance(quarantines, list):
|
|
failures.append("test evidence quarantines must be a list")
|
|
quarantines = []
|
|
|
|
observations: dict[str, list[date]] = {}
|
|
history_keys: set[tuple[str, str, str]] = set()
|
|
for index, entry in enumerate(history):
|
|
label = f"flake history {index}"
|
|
if not isinstance(entry, dict):
|
|
failures.append(f"{label} must be an object")
|
|
continue
|
|
nodeid = entry.get("nodeid")
|
|
head_sha = entry.get("head_sha")
|
|
gate = entry.get("gate")
|
|
if not isinstance(nodeid, str) or not nodeid.strip():
|
|
failures.append(f"{label} requires a non-empty nodeid")
|
|
continue
|
|
if not isinstance(head_sha, str) or re.fullmatch(r"[0-9a-f]{40}", head_sha) is None:
|
|
failures.append(f"{label} requires a 40-character head_sha")
|
|
if not isinstance(gate, str) or not gate.strip():
|
|
failures.append(f"{label} requires a non-empty gate")
|
|
observed = _policy_date(entry.get("observed_on"), f"{label} observed_on", failures)
|
|
if observed is not None:
|
|
if observed > today:
|
|
failures.append(f"{label} observed_on cannot be in the future")
|
|
observations.setdefault(nodeid, []).append(observed)
|
|
key = (nodeid, str(entry.get("observed_on")), str(head_sha))
|
|
if key in history_keys:
|
|
failures.append(f"duplicate flake history entry for {nodeid}")
|
|
history_keys.add(key)
|
|
|
|
active_quarantines: set[str] = set()
|
|
quarantine_nodeids: set[str] = set()
|
|
max_days = numeric.get("maximum_quarantine_days", 30)
|
|
for index, entry in enumerate(quarantines):
|
|
label = f"test quarantine {index}"
|
|
if not isinstance(entry, dict):
|
|
failures.append(f"{label} must be an object")
|
|
continue
|
|
nodeid = entry.get("nodeid")
|
|
if not isinstance(nodeid, str) or not nodeid.strip():
|
|
failures.append(f"{label} requires a non-empty nodeid")
|
|
continue
|
|
if nodeid in quarantine_nodeids:
|
|
failures.append(f"duplicate test quarantine for {nodeid}")
|
|
quarantine_nodeids.add(nodeid)
|
|
owner = entry.get("owner")
|
|
if not isinstance(owner, str) or not owner.startswith("@"):
|
|
failures.append(f"{label} requires an @owner")
|
|
if not isinstance(entry.get("reason"), str) or not entry["reason"].strip():
|
|
failures.append(f"{label} requires a non-empty reason")
|
|
issue = entry.get("issue")
|
|
if not isinstance(issue, str) or not issue.startswith(
|
|
"https://github.com/elder-plinius/OBLITERATUS/issues/",
|
|
):
|
|
failures.append(f"{label} requires an OBLITERATUS issue URL")
|
|
opened = _policy_date(entry.get("opened"), f"{label} opened", failures)
|
|
expires = _policy_date(entry.get("expires"), f"{label} expires", failures)
|
|
if opened is not None and expires is not None:
|
|
if opened > today:
|
|
failures.append(f"{label} cannot open in the future")
|
|
if expires <= opened:
|
|
failures.append(f"{label} must expire after it opens")
|
|
elif expires - opened > timedelta(days=max_days):
|
|
failures.append(f"{label} exceeds the {max_days}-day maximum")
|
|
elif expires < today:
|
|
failures.append(f"{label} expired on {expires.isoformat()}")
|
|
else:
|
|
active_quarantines.add(nodeid)
|
|
|
|
window = numeric.get("flake_window_days", 30)
|
|
cutoff = today - timedelta(days=window - 1)
|
|
for nodeid, dates in observations.items():
|
|
recent = [observed for observed in dates if cutoff <= observed <= today]
|
|
if len(recent) >= 2 and nodeid not in active_quarantines:
|
|
failures.append(
|
|
f"test {nodeid} flaked {len(recent)} times in {window} days "
|
|
"without an active quarantine",
|
|
)
|
|
failures.extend(_validate_duration_policy(evidence, today=today))
|
|
return failures
|
|
|
|
|
|
def _positive_number(value: Any) -> bool:
|
|
return (
|
|
not isinstance(value, bool)
|
|
and isinstance(value, (int, float))
|
|
and math.isfinite(value)
|
|
and value > 0
|
|
)
|
|
|
|
|
|
def _validate_duration_policy(evidence: dict[str, Any], *, today: date) -> list[str]:
|
|
"""Validate owned suite, marker, testcase, and repeat duration budgets."""
|
|
failures: list[str] = []
|
|
budgets = evidence.get("duration_budgets")
|
|
if not isinstance(budgets, dict):
|
|
return ["test evidence duration_budgets must be an object"]
|
|
|
|
mandatory = budgets.get("mandatory_cpu")
|
|
if not isinstance(mandatory, dict):
|
|
failures.append("duration budget mandatory_cpu must be an object")
|
|
else:
|
|
if not isinstance(mandatory.get("owner"), str) or not mandatory["owner"].startswith("@"):
|
|
failures.append("duration budget mandatory_cpu requires an @owner")
|
|
suite_budgets = mandatory.get("max_suite_seconds_by_python")
|
|
if not isinstance(suite_budgets, dict):
|
|
failures.append(
|
|
"duration budget mandatory_cpu max_suite_seconds_by_python must be an object",
|
|
)
|
|
else:
|
|
versions = set(suite_budgets)
|
|
if versions != SUPPORTED_PYTHON:
|
|
failures.append(
|
|
"duration budget mandatory_cpu must cover exactly Python 3.10, 3.11, and 3.12",
|
|
)
|
|
for version, value in suite_budgets.items():
|
|
if not _positive_number(value):
|
|
failures.append(
|
|
"duration budget mandatory_cpu "
|
|
f"max_suite_seconds_by_python.{version} must be positive",
|
|
)
|
|
testcase_budget = mandatory.get("max_testcase_seconds")
|
|
if not _positive_number(testcase_budget):
|
|
failures.append(
|
|
"duration budget mandatory_cpu max_testcase_seconds must be positive",
|
|
)
|
|
marker_budgets = mandatory.get("max_marker_seconds")
|
|
if not isinstance(marker_budgets, dict) or not marker_budgets:
|
|
failures.append(
|
|
"duration budget mandatory_cpu max_marker_seconds must be a non-empty object",
|
|
)
|
|
else:
|
|
if "unmarked" not in marker_budgets:
|
|
failures.append("duration budget mandatory_cpu must own the unmarked layer")
|
|
for marker, value in marker_budgets.items():
|
|
if not isinstance(marker, str) or not marker.strip() or not _positive_number(value):
|
|
failures.append(
|
|
f"duration budget mandatory_cpu has invalid marker budget {marker!r}",
|
|
)
|
|
|
|
review_days = mandatory.get("maximum_owner_days")
|
|
if (
|
|
isinstance(review_days, bool)
|
|
or not isinstance(review_days, int)
|
|
or not 1 <= review_days <= 365
|
|
):
|
|
failures.append(
|
|
"duration budget mandatory_cpu maximum_owner_days must be an integer from 1 to 365",
|
|
)
|
|
review_days = 90
|
|
owners = mandatory.get("owned_slow_tests")
|
|
if not isinstance(owners, list):
|
|
failures.append("duration budget mandatory_cpu owned_slow_tests must be a list")
|
|
owners = []
|
|
seen: set[str] = set()
|
|
for index, entry in enumerate(owners):
|
|
label = f"owned slow test {index}"
|
|
if not isinstance(entry, dict):
|
|
failures.append(f"{label} must be an object")
|
|
continue
|
|
nodeid = entry.get("nodeid")
|
|
if not isinstance(nodeid, str) or not nodeid.strip():
|
|
failures.append(f"{label} requires a non-empty nodeid")
|
|
continue
|
|
if nodeid in seen:
|
|
failures.append(f"duplicate owned slow test {nodeid}")
|
|
seen.add(nodeid)
|
|
if not isinstance(entry.get("owner"), str) or not entry["owner"].startswith("@"):
|
|
failures.append(f"{label} requires an @owner")
|
|
if not isinstance(entry.get("reason"), str) or not entry["reason"].strip():
|
|
failures.append(f"{label} requires a non-empty reason")
|
|
issue = entry.get("issue")
|
|
if not isinstance(issue, str) or not issue.startswith(
|
|
"https://github.com/elder-plinius/OBLITERATUS/issues/",
|
|
):
|
|
failures.append(f"{label} requires an OBLITERATUS issue URL")
|
|
maximum = entry.get("max_seconds")
|
|
if not _positive_number(maximum):
|
|
failures.append(f"{label} max_seconds must be positive")
|
|
elif _positive_number(testcase_budget) and maximum <= testcase_budget:
|
|
failures.append(
|
|
f"{label} max_seconds must exceed the default testcase budget",
|
|
)
|
|
opened = _policy_date(entry.get("opened"), f"{label} opened", failures)
|
|
expires = _policy_date(entry.get("expires"), f"{label} expires", failures)
|
|
if opened is not None and expires is not None:
|
|
if opened > today:
|
|
failures.append(f"{label} cannot open in the future")
|
|
if expires <= opened:
|
|
failures.append(f"{label} must expire after it opens")
|
|
elif expires - opened > timedelta(days=review_days):
|
|
failures.append(f"{label} exceeds the {review_days}-day review window")
|
|
elif expires < today:
|
|
failures.append(f"{label} expired on {expires.isoformat()}")
|
|
|
|
repeat = budgets.get("repeat_gate")
|
|
if not isinstance(repeat, dict):
|
|
failures.append("duration budget repeat_gate must be an object")
|
|
else:
|
|
if not isinstance(repeat.get("owner"), str) or not repeat["owner"].startswith("@"):
|
|
failures.append("duration budget repeat_gate requires an @owner")
|
|
total = repeat.get("max_total_seconds")
|
|
per_pass = repeat.get("max_pass_seconds")
|
|
if not _positive_number(total):
|
|
failures.append("duration budget repeat_gate max_total_seconds must be positive")
|
|
if not _positive_number(per_pass):
|
|
failures.append("duration budget repeat_gate max_pass_seconds must be positive")
|
|
if _positive_number(total) and _positive_number(per_pass) and per_pass > total:
|
|
failures.append(
|
|
"duration budget repeat_gate max_pass_seconds cannot exceed max_total_seconds",
|
|
)
|
|
return failures
|
|
|
|
|
|
def _duration_value(value: Any, label: str, failures: list[str]) -> float | None:
|
|
if not _positive_number(value) and value != 0:
|
|
failures.append(f"{label} must be a non-negative finite number")
|
|
return None
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
failures.append(f"{label} must be a non-negative finite number")
|
|
return None
|
|
return float(value)
|
|
|
|
|
|
def validate_duration_evidence(
|
|
trend: dict[str, Any], policy: dict[str, Any],
|
|
) -> list[str]:
|
|
"""Enforce mandatory CPU and repeat budgets on normalized trend evidence."""
|
|
failures: list[str] = []
|
|
budgets = policy["test_evidence"]["duration_budgets"]
|
|
tests = trend.get("tests")
|
|
if tests is not None:
|
|
if not isinstance(tests, dict):
|
|
failures.append("test trend tests must be an object")
|
|
else:
|
|
mandatory = budgets["mandatory_cpu"]
|
|
python_version = trend.get("python")
|
|
suite_budgets = mandatory["max_suite_seconds_by_python"]
|
|
if python_version not in suite_budgets:
|
|
failures.append(f"test trend has unsupported Python version {python_version!r}")
|
|
suite_duration = _duration_value(
|
|
tests.get("suite_duration_seconds"),
|
|
"test trend suite_duration_seconds",
|
|
failures,
|
|
)
|
|
if suite_duration is not None and python_version in suite_budgets:
|
|
maximum = float(suite_budgets[python_version])
|
|
if suite_duration > maximum:
|
|
failures.append(
|
|
f"test trend Python {python_version} suite duration "
|
|
f"{suite_duration:.3f}s exceeds {maximum:.3f}s",
|
|
)
|
|
|
|
durations = tests.get("durations")
|
|
if not isinstance(durations, list) or not durations:
|
|
failures.append("test trend durations must be a non-empty list")
|
|
durations = []
|
|
missing_markers = tests.get("missing_marker_nodeids")
|
|
if not isinstance(missing_markers, list) or any(
|
|
not isinstance(nodeid, str) or not nodeid.strip()
|
|
for nodeid in missing_markers
|
|
):
|
|
failures.append("test trend missing_marker_nodeids must be a string list")
|
|
missing_markers = []
|
|
if tests.get("marker_metadata_complete") is not True or missing_markers:
|
|
failures.append(
|
|
"test trend duration marker metadata is incomplete"
|
|
+ (f" for {len(missing_markers)} testcase(s)" if missing_markers else ""),
|
|
)
|
|
owners = {
|
|
entry["nodeid"]: entry
|
|
for entry in mandatory["owned_slow_tests"]
|
|
}
|
|
default_max = float(mandatory["max_testcase_seconds"])
|
|
seen_nodeids: set[str] = set()
|
|
computed_markers: dict[str, dict[str, int | float]] = {}
|
|
for index, entry in enumerate(durations):
|
|
label = f"test duration {index}"
|
|
if not isinstance(entry, dict):
|
|
failures.append(f"{label} must be an object")
|
|
continue
|
|
nodeid = entry.get("nodeid")
|
|
if not isinstance(nodeid, str) or not nodeid.strip():
|
|
failures.append(f"{label} requires a non-empty nodeid")
|
|
continue
|
|
if nodeid in seen_nodeids:
|
|
failures.append(f"test trend repeats duration for {nodeid}")
|
|
seen_nodeids.add(nodeid)
|
|
seconds = _duration_value(entry.get("seconds"), f"{label} seconds", failures)
|
|
markers = entry.get("markers")
|
|
if not isinstance(markers, list) or not markers or any(
|
|
not isinstance(marker, str) or not marker.strip() for marker in markers
|
|
):
|
|
failures.append(f"{label} markers must be a non-empty string list")
|
|
markers = []
|
|
if seconds is not None:
|
|
if seconds > default_max:
|
|
owner = owners.get(nodeid)
|
|
if owner is None:
|
|
failures.append(
|
|
f"test {nodeid} took {seconds:.3f}s above the "
|
|
f"{default_max:.3f}s default and has no owned slow-test budget",
|
|
)
|
|
elif seconds > float(owner["max_seconds"]):
|
|
failures.append(
|
|
f"owned slow test {nodeid} took {seconds:.3f}s above its "
|
|
f"{float(owner['max_seconds']):.3f}s budget",
|
|
)
|
|
for marker in set(markers):
|
|
summary = computed_markers.setdefault(
|
|
marker, {"tests": 0, "duration_seconds": 0.0},
|
|
)
|
|
summary["tests"] = int(summary["tests"]) + 1
|
|
summary["duration_seconds"] = (
|
|
float(summary["duration_seconds"]) + seconds
|
|
)
|
|
|
|
total = tests.get("total")
|
|
if isinstance(total, bool) or not isinstance(total, int) or total != len(durations):
|
|
failures.append("test trend total must equal the number of duration records")
|
|
marker_evidence = tests.get("marker_durations")
|
|
if not isinstance(marker_evidence, dict):
|
|
failures.append("test trend marker_durations must be an object")
|
|
marker_evidence = {}
|
|
marker_budgets = mandatory["max_marker_seconds"]
|
|
for marker, computed in computed_markers.items():
|
|
if marker not in marker_budgets:
|
|
failures.append(f"test marker {marker} has no duration budget")
|
|
continue
|
|
observed = _duration_value(
|
|
computed["duration_seconds"],
|
|
f"test marker {marker} duration",
|
|
failures,
|
|
)
|
|
if observed is not None and observed > float(marker_budgets[marker]):
|
|
failures.append(
|
|
f"test marker {marker} duration {observed:.3f}s exceeds "
|
|
f"{float(marker_budgets[marker]):.3f}s",
|
|
)
|
|
recorded = marker_evidence.get(marker)
|
|
expected = {
|
|
"tests": computed["tests"],
|
|
"duration_seconds": round(float(computed["duration_seconds"]), 3),
|
|
}
|
|
if recorded != expected:
|
|
failures.append(f"test marker {marker} duration summary is inconsistent")
|
|
extra_markers = sorted(set(marker_evidence) - set(computed_markers))
|
|
for marker in extra_markers:
|
|
failures.append(f"test marker {marker} has summary without duration records")
|
|
|
|
repeat = trend.get("repeat")
|
|
if repeat is not None:
|
|
if not isinstance(repeat, dict):
|
|
failures.append("test trend repeat must be an object")
|
|
else:
|
|
repeat_budget = budgets["repeat_gate"]
|
|
total = _duration_value(
|
|
repeat.get("total_duration_seconds"),
|
|
"repeat gate total duration",
|
|
failures,
|
|
)
|
|
if total is not None and total > float(repeat_budget["max_total_seconds"]):
|
|
failures.append(
|
|
f"repeat gate total duration {total:.3f}s exceeds "
|
|
f"{float(repeat_budget['max_total_seconds']):.3f}s",
|
|
)
|
|
passes = repeat.get("passes")
|
|
if not isinstance(passes, list) or not passes:
|
|
failures.append("repeat gate passes must be a non-empty list")
|
|
passes = []
|
|
for index, entry in enumerate(passes, start=1):
|
|
if not isinstance(entry, dict):
|
|
failures.append(f"repeat gate pass {index} must be an object")
|
|
continue
|
|
duration = _duration_value(
|
|
entry.get("duration_seconds"),
|
|
f"repeat gate pass {index} duration",
|
|
failures,
|
|
)
|
|
if duration is not None and duration > float(repeat_budget["max_pass_seconds"]):
|
|
failures.append(
|
|
f"repeat gate pass {index} duration {duration:.3f}s exceeds "
|
|
f"{float(repeat_budget['max_pass_seconds']):.3f}s",
|
|
)
|
|
if tests is None and repeat is None:
|
|
failures.append("test trend contains neither tests nor repeat duration evidence")
|
|
return failures
|
|
|
|
|
|
def validate_policy(policy: dict[str, Any], *, today: date | None = None) -> list[str]:
|
|
"""Validate policy structure, immutable floors, and exclusion traceability."""
|
|
failures: list[str] = []
|
|
minimums = policy.get("minimums")
|
|
if not isinstance(minimums, dict):
|
|
return ["quality policy is missing the minimums object"]
|
|
|
|
exceptions = policy.get("threshold_exceptions", [])
|
|
for name, baseline in BASELINE_FLOORS.items():
|
|
value = minimums.get(name)
|
|
if (
|
|
isinstance(value, bool)
|
|
or not isinstance(value, (int, float))
|
|
or not math.isfinite(value)
|
|
):
|
|
failures.append(f"quality policy requires numeric minimum {name}")
|
|
elif value < baseline and not _valid_exception(exceptions, name, float(value)):
|
|
failures.append(
|
|
f"quality minimum {name} cannot move below {baseline:g} "
|
|
"without an explicit reviewed exception",
|
|
)
|
|
|
|
critical_paths = policy.get("critical_cpu_paths")
|
|
if not isinstance(critical_paths, list) or not critical_paths:
|
|
failures.append("quality policy requires non-empty critical_cpu_paths")
|
|
else:
|
|
seen_critical: set[str] = set()
|
|
for path in critical_paths:
|
|
if not isinstance(path, str) or not path.startswith("obliteratus/"):
|
|
failures.append(f"invalid critical CPU path: {path!r}")
|
|
elif path in seen_critical:
|
|
failures.append(f"critical CPU path is duplicated: {path}")
|
|
else:
|
|
seen_critical.add(path)
|
|
|
|
scope = policy.get("mature_cpu_scope")
|
|
exclusions = scope.get("exclusions") if isinstance(scope, dict) else None
|
|
if not isinstance(exclusions, list):
|
|
failures.append("quality policy is missing mature_cpu_scope.exclusions")
|
|
return failures
|
|
|
|
paths: set[str] = set()
|
|
for index, exclusion in enumerate(exclusions):
|
|
label = f"mature CPU exclusion {index}"
|
|
if not isinstance(exclusion, dict):
|
|
failures.append(f"{label} must be an object")
|
|
continue
|
|
path = exclusion.get("path")
|
|
if not isinstance(path, str) or not path.startswith("obliteratus/"):
|
|
failures.append(f"{label} requires an obliteratus source path")
|
|
elif path in paths:
|
|
failures.append(f"mature CPU exclusion path is duplicated: {path}")
|
|
else:
|
|
paths.add(path)
|
|
for key in ("boundary", "rationale", "conditional_gate"):
|
|
if not isinstance(exclusion.get(key), str) or not exclusion[key].strip():
|
|
failures.append(f"{label} requires non-empty {key}")
|
|
issue = exclusion.get("conditional_issue")
|
|
if issue != "https://github.com/elder-plinius/OBLITERATUS/issues/71":
|
|
failures.append(f"{label} must link the conditional-test issue #71")
|
|
failures.extend(_validate_test_evidence(policy, today=today or date.today()))
|
|
return failures
|
|
|
|
|
|
def measure_mature_cpu_scope(
|
|
report: dict[str, Any], policy: dict[str, Any],
|
|
) -> tuple[dict[str, float | int], list[str]]:
|
|
"""Measure coverage after removing only documented environment boundaries."""
|
|
files = report.get("files")
|
|
if not isinstance(files, dict):
|
|
return {}, ["coverage report is missing the files object"]
|
|
exclusions = policy["mature_cpu_scope"]["exclusions"]
|
|
excluded_paths = {entry["path"] for entry in exclusions}
|
|
missing = sorted(excluded_paths - files.keys())
|
|
if missing:
|
|
return {}, [f"coverage report is missing excluded source file {path}" for path in missing]
|
|
|
|
statements = covered_lines = branches = covered_branches = 0
|
|
for path, entry in files.items():
|
|
if path in excluded_paths or not path.startswith("obliteratus/"):
|
|
continue
|
|
summary = entry.get("summary") if isinstance(entry, dict) else None
|
|
if not isinstance(summary, dict):
|
|
return {}, [f"coverage report is missing summary for {path}"]
|
|
statements += int(summary.get("num_statements", 0))
|
|
covered_lines += int(summary.get("covered_lines", 0))
|
|
branches += int(summary.get("num_branches", 0))
|
|
covered_branches += int(summary.get("covered_branches", 0))
|
|
|
|
measurement: dict[str, float | int] = {
|
|
"statements": statements,
|
|
"covered_lines": covered_lines,
|
|
"line_percent": covered_lines / statements * 100 if statements else 100.0,
|
|
"branches": branches,
|
|
"covered_branches": covered_branches,
|
|
"branch_percent": covered_branches / branches * 100 if branches else 100.0,
|
|
}
|
|
return measurement, []
|
|
|
|
|
|
def validate_mature_cpu_scope(
|
|
report: dict[str, Any], policy: dict[str, Any],
|
|
) -> tuple[dict[str, float | int], list[str]]:
|
|
measurement, failures = measure_mature_cpu_scope(report, policy)
|
|
if failures:
|
|
return measurement, failures
|
|
minimums = policy["minimums"]
|
|
for label in ("line", "branch"):
|
|
value = float(measurement[f"{label}_percent"])
|
|
floor = float(minimums[f"mature_cpu_{'statement' if label == 'line' else 'branch'}"])
|
|
if value < floor:
|
|
failures.append(
|
|
f"mature CPU {label} coverage {value:.2f}% is below the {floor:.2f}% floor",
|
|
)
|
|
return measurement, failures
|
|
|
|
|
|
def _parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--policy", type=Path, required=True)
|
|
parser.add_argument("--coverage", type=Path)
|
|
parser.add_argument("--evidence", type=Path)
|
|
return parser
|
|
|
|
|
|
def main() -> int:
|
|
args = _parser().parse_args()
|
|
try:
|
|
policy = json.loads(args.policy.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
print(f"quality policy failed: cannot read policy: {exc}")
|
|
return 1
|
|
if not isinstance(policy, dict):
|
|
print("quality policy failed: policy root must be an object")
|
|
return 1
|
|
|
|
failures = validate_policy(policy)
|
|
measurement = None
|
|
if args.coverage is not None and not failures:
|
|
try:
|
|
report = json.loads(args.coverage.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
failures.append(f"cannot read coverage report: {exc}")
|
|
else:
|
|
if not isinstance(report, dict):
|
|
failures.append("coverage report root must be an object")
|
|
else:
|
|
measurement, coverage_failures = validate_mature_cpu_scope(report, policy)
|
|
failures.extend(coverage_failures)
|
|
if args.evidence is not None and not failures:
|
|
try:
|
|
trend = json.loads(args.evidence.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError) as exc:
|
|
failures.append(f"cannot read test trend evidence: {exc}")
|
|
else:
|
|
if not isinstance(trend, dict):
|
|
failures.append("test trend evidence root must be an object")
|
|
else:
|
|
failures.extend(validate_duration_evidence(trend, policy))
|
|
|
|
if failures:
|
|
for failure in failures:
|
|
print(f"quality policy failed: {failure}")
|
|
return 1
|
|
if measurement is None:
|
|
print("quality policy passed")
|
|
else:
|
|
print(
|
|
"quality policy passed: mature CPU "
|
|
f"line={measurement['line_percent']:.2f}% "
|
|
f"branch={measurement['branch_percent']:.2f}%",
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|