HH-446: restore auditable quality gates from current main (#102)
* HH-446: restore auditable quality gates * test(HH-446): restore help center smoke contract --------- Co-authored-by: Rogee <rogee@ipao.vip>
This commit is contained in:
Rogee
Executable
+173
@@ -0,0 +1,173 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Validate Go test coverage, execution evidence, and elapsed time."""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def coverage(path):
|
||||
files = defaultdict(lambda: [0, 0])
|
||||
total = [0, 0]
|
||||
for line in Path(path).read_text(encoding="utf-8").splitlines():
|
||||
if not line or line.startswith("mode:"):
|
||||
continue
|
||||
fields = line.split()
|
||||
if len(fields) != 3:
|
||||
raise ValueError(f"invalid coverage line: {line}")
|
||||
file_name = fields[0].split(":", 1)[0]
|
||||
statements, count = int(fields[1]), int(fields[2])
|
||||
total[0] += statements
|
||||
total[1] += statements if count else 0
|
||||
files[file_name][0] += statements
|
||||
files[file_name][1] += statements if count else 0
|
||||
if total[0] == 0:
|
||||
raise ValueError("coverage profile has no statements")
|
||||
return total, files
|
||||
|
||||
|
||||
def test_evidence(path):
|
||||
packages = defaultdict(lambda: {"pass": 0, "fail": 0, "skip": 0})
|
||||
totals = {"pass": 0, "fail": 0, "skip": 0}
|
||||
for line in Path(path).read_text(encoding="utf-8").splitlines():
|
||||
try:
|
||||
event = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
action = event.get("Action")
|
||||
package = event.get("Package", "")
|
||||
if not event.get("Test") or action not in totals:
|
||||
continue
|
||||
packages[package][action] += 1
|
||||
totals[action] += 1
|
||||
return totals, packages
|
||||
|
||||
|
||||
def package_matches(name, spec):
|
||||
recursive = spec.endswith("/...")
|
||||
package = spec.removesuffix("/...")
|
||||
return name == package or recursive and name.startswith(package + "/")
|
||||
|
||||
|
||||
def run(argv=None):
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--coverage", required=True)
|
||||
parser.add_argument("--test-json", required=True)
|
||||
parser.add_argument("--duration", required=True)
|
||||
parser.add_argument("--report", required=True)
|
||||
parser.add_argument("--min-coverage", type=float, default=65.0)
|
||||
parser.add_argument("--max-duration", type=float, required=True)
|
||||
parser.add_argument("--test-exit-code", type=int, default=0)
|
||||
parser.add_argument("--critical-package", action="append", default=[])
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
failures = []
|
||||
try:
|
||||
total, files = coverage(args.coverage)
|
||||
total_tests, packages = test_evidence(args.test_json)
|
||||
elapsed = float(Path(args.duration).read_text(encoding="utf-8").strip())
|
||||
except (OSError, ValueError) as error:
|
||||
failures.append(str(error))
|
||||
total, files = [0, 0], {}
|
||||
total_tests, packages, elapsed = {"pass": 0, "fail": 0, "skip": 0}, {}, None
|
||||
|
||||
percent = total[1] / total[0] * 100 if total[0] else 0.0
|
||||
if percent < args.min_coverage:
|
||||
failures.append(f"overall coverage {percent:.2f}% is below {args.min_coverage:.2f}%")
|
||||
if args.test_exit_code:
|
||||
failures.append(f"go test exited with {args.test_exit_code}")
|
||||
if total_tests["fail"]:
|
||||
failures.append(f"test evidence contains {total_tests['fail']} failed test events")
|
||||
if elapsed is not None and elapsed > args.max_duration:
|
||||
failures.append(f"test duration {elapsed:.2f}s exceeds {args.max_duration:.2f}s")
|
||||
if sum(total_tests.values()) == 0:
|
||||
failures.append("test evidence contains no test results")
|
||||
|
||||
critical, critical_coverage = [], []
|
||||
for spec in args.critical_package:
|
||||
matched = {
|
||||
name: values for name, values in packages.items() if package_matches(name, spec)
|
||||
}
|
||||
item = {
|
||||
"package": spec,
|
||||
"matched": sorted(matched),
|
||||
"passed": sum(values["pass"] for values in matched.values()),
|
||||
"failed": sum(values["fail"] for values in matched.values()),
|
||||
"skipped": sum(values["skip"] for values in matched.values()),
|
||||
}
|
||||
critical.append(item)
|
||||
if not matched:
|
||||
failures.append(f"critical suite {spec} produced no package evidence")
|
||||
elif item["passed"] == 0:
|
||||
failures.append(f"critical suite {spec} was entirely skipped")
|
||||
|
||||
matching = [
|
||||
values
|
||||
for name, values in files.items()
|
||||
if package_matches(name.rsplit("/", 1)[0], spec)
|
||||
]
|
||||
statements = sum(values[0] for values in matching)
|
||||
covered = sum(values[1] for values in matching)
|
||||
coverage_item = {
|
||||
"package": spec,
|
||||
"covered_statements": covered,
|
||||
"total_statements": statements,
|
||||
"percent": round(covered / statements * 100, 2) if statements else 0.0,
|
||||
}
|
||||
critical_coverage.append(coverage_item)
|
||||
if not statements:
|
||||
failures.append(f"critical coverage {spec} has no statements")
|
||||
elif coverage_item["percent"] < args.min_coverage:
|
||||
failures.append(
|
||||
f"critical coverage {spec} {coverage_item['percent']:.2f}% "
|
||||
f"is below {args.min_coverage:.2f}%"
|
||||
)
|
||||
|
||||
report = {
|
||||
"schema_version": 1,
|
||||
"status": "failed" if failures else "passed",
|
||||
"thresholds": {
|
||||
"min_coverage_percent": args.min_coverage,
|
||||
"max_duration_seconds": args.max_duration,
|
||||
},
|
||||
"coverage": {
|
||||
"covered_statements": total[1],
|
||||
"total_statements": total[0],
|
||||
"percent": round(percent, 2),
|
||||
"critical": critical_coverage,
|
||||
"files": {
|
||||
name: {
|
||||
"covered_statements": values[1],
|
||||
"total_statements": values[0],
|
||||
"percent": round(values[1] / values[0] * 100, 2) if values[0] else 0.0,
|
||||
}
|
||||
for name, values in sorted(files.items())
|
||||
},
|
||||
},
|
||||
"tests": {**total_tests, "critical": critical},
|
||||
"duration_seconds": elapsed,
|
||||
"test_exit_code": args.test_exit_code,
|
||||
"failures": failures,
|
||||
}
|
||||
report_path = Path(args.report)
|
||||
report_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
report_path.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
print(
|
||||
json.dumps(
|
||||
{
|
||||
"status": report["status"],
|
||||
"coverage_percent": report["coverage"]["percent"],
|
||||
"tests": report["tests"],
|
||||
"duration_seconds": elapsed,
|
||||
"failures": failures,
|
||||
},
|
||||
sort_keys=True,
|
||||
)
|
||||
)
|
||||
return 1 if failures else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(run())
|
||||
Reference in New Issue
Block a user