mirror of
https://github.com/elder-plinius/OBLITERATUS.git
synced 2026-08-17 16:37:30 +02:00
test: add quality-depth gates and mature CPU coverage
This commit is contained in:
Joseph Magly
@@ -0,0 +1,78 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Validate machine-readable mutmut CI statistics against a score floor."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import math
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
def mutation_score(stats: dict[str, Any]) -> tuple[int, int, float]:
|
||||
"""Return killed, total, and percentage after validating mutmut statistics."""
|
||||
killed = stats.get("killed")
|
||||
total = stats.get("total")
|
||||
if isinstance(killed, bool) or not isinstance(killed, int) or killed < 0:
|
||||
raise ValueError("mutation statistics require a non-negative integer 'killed'")
|
||||
if isinstance(total, bool) or not isinstance(total, int) or total <= 0:
|
||||
raise ValueError("mutation statistics require a positive integer 'total'")
|
||||
if killed > total:
|
||||
raise ValueError("mutation statistics cannot report more killed mutants than total")
|
||||
return killed, total, killed / total * 100
|
||||
|
||||
|
||||
def validate_mutation_stats(
|
||||
stats: dict[str, Any], *, minimum: float,
|
||||
) -> list[str]:
|
||||
"""Return failures for an incomplete run or a score below the policy floor."""
|
||||
if not math.isfinite(minimum) or minimum < 0 or minimum > 100:
|
||||
return ["minimum mutation score must be between 0 and 100"]
|
||||
try:
|
||||
killed, total, score = mutation_score(stats)
|
||||
except ValueError as exc:
|
||||
return [str(exc)]
|
||||
|
||||
failures: list[str] = []
|
||||
if stats.get("check_was_interrupted_by_user"):
|
||||
failures.append("mutation run was interrupted")
|
||||
if score < minimum:
|
||||
failures.append(
|
||||
f"mutation score {score:.2f}% ({killed}/{total}) is below "
|
||||
f"the {minimum:.2f}% floor",
|
||||
)
|
||||
return failures
|
||||
|
||||
|
||||
def _parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("stats", type=Path, help="mutmut-cicd-stats.json path")
|
||||
parser.add_argument("--minimum", type=float, required=True)
|
||||
return parser
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = _parser().parse_args()
|
||||
try:
|
||||
stats = json.loads(args.stats.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError) as exc:
|
||||
print(f"mutation gate failed: cannot read statistics: {exc}")
|
||||
return 1
|
||||
if not isinstance(stats, dict):
|
||||
print("mutation gate failed: statistics root must be an object")
|
||||
return 1
|
||||
|
||||
failures = validate_mutation_stats(stats, minimum=args.minimum)
|
||||
if failures:
|
||||
for failure in failures:
|
||||
print(f"mutation gate failed: {failure}")
|
||||
return 1
|
||||
|
||||
killed, total, score = mutation_score(stats)
|
||||
print(f"mutation gate passed: {score:.2f}% ({killed}/{total})")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,195 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Validate immutable quality floors and mature CPU-scope coverage."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import math
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
BASELINE_FLOORS = {
|
||||
"repository_statement": 60.0,
|
||||
"repository_branch": 42.0,
|
||||
"changed_line": 90.0,
|
||||
"mature_cpu_statement": 80.0,
|
||||
"mature_cpu_branch": 75.0,
|
||||
"mutation_score": 70.0,
|
||||
"warning_budget": 0.0,
|
||||
}
|
||||
|
||||
|
||||
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 validate_policy(policy: dict[str, Any]) -> 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",
|
||||
)
|
||||
|
||||
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")
|
||||
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)
|
||||
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 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())
|
||||
@@ -0,0 +1,111 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Repeat deterministic tests in varied orders and emit timing evidence."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Sequence
|
||||
|
||||
|
||||
DEFAULT_TESTS = (
|
||||
"tests/test_config.py",
|
||||
"tests/test_config_properties.py",
|
||||
"tests/test_coverage_thresholds.py",
|
||||
"tests/test_property_contracts.py",
|
||||
"tests/test_advanced_metrics.py",
|
||||
"tests/test_metrics.py",
|
||||
)
|
||||
HASH_SEEDS = ("0", "1", "8675309")
|
||||
|
||||
|
||||
def test_orders(paths: Sequence[str]) -> list[list[str]]:
|
||||
"""Return stable forward, reverse, and interleaved orders."""
|
||||
forward = list(paths)
|
||||
reverse = list(reversed(paths))
|
||||
interleaved = forward[::2] + forward[1::2]
|
||||
return [forward, reverse, interleaved]
|
||||
|
||||
|
||||
def run_repeat_gate(
|
||||
paths: Sequence[str], *, output: Path, python: str = sys.executable,
|
||||
) -> int:
|
||||
"""Run three deterministic passes, always writing a JSON evidence record."""
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
results: list[dict[str, Any]] = []
|
||||
started = time.monotonic()
|
||||
exit_code = 0
|
||||
for index, (order, hash_seed) in enumerate(
|
||||
zip(test_orders(paths), HASH_SEEDS, strict=True),
|
||||
start=1,
|
||||
):
|
||||
command = [python, "-m", "pytest", "--no-cov", "-q", *order]
|
||||
environment = os.environ.copy()
|
||||
environment["PYTHONHASHSEED"] = hash_seed
|
||||
pass_started = time.monotonic()
|
||||
completed = subprocess.run(
|
||||
command,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=environment,
|
||||
check=False,
|
||||
)
|
||||
duration = time.monotonic() - pass_started
|
||||
results.append({
|
||||
"pass": index,
|
||||
"python_hash_seed": hash_seed,
|
||||
"tests": order,
|
||||
"duration_seconds": round(duration, 3),
|
||||
"return_code": completed.returncode,
|
||||
"stdout": completed.stdout[-4000:],
|
||||
"stderr": completed.stderr[-4000:],
|
||||
})
|
||||
if completed.returncode != 0:
|
||||
exit_code = completed.returncode
|
||||
break
|
||||
|
||||
evidence = {
|
||||
"schema_version": 1,
|
||||
"status": "passed" if exit_code == 0 else "failed",
|
||||
"total_duration_seconds": round(time.monotonic() - started, 3),
|
||||
"passes": results,
|
||||
}
|
||||
output.write_text(json.dumps(evidence, indent=2) + "\n", encoding="utf-8")
|
||||
if exit_code:
|
||||
failed = results[-1]
|
||||
print(
|
||||
f"repeat gate failed on pass {failed['pass']} "
|
||||
f"with PYTHONHASHSEED={failed['python_hash_seed']}",
|
||||
)
|
||||
if failed["stdout"]:
|
||||
print(failed["stdout"])
|
||||
if failed["stderr"]:
|
||||
print(failed["stderr"], file=sys.stderr)
|
||||
return exit_code
|
||||
|
||||
print(
|
||||
f"repeat gate passed: {len(results)} orders in "
|
||||
f"{evidence['total_duration_seconds']:.3f}s",
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
def _parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("tests", nargs="*", default=list(DEFAULT_TESTS))
|
||||
parser.add_argument("--output", type=Path, required=True)
|
||||
return parser
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = _parser().parse_args()
|
||||
return run_repeat_gate(args.tests, output=args.output)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user