From f606c0543ca181ad448df49602f173f98e835b0a Mon Sep 17 00:00:00 2001 From: Brad Stein Date: Sat, 18 Apr 2026 16:33:52 -0300 Subject: [PATCH] ci(metrics): derive checks plus workspace coverage/loc gauges --- testing/ci/publish_metrics.py | 114 ++++++++++++++++++++++++++++++++++ testing/ci/summary.py | 50 +++++++++++++-- 2 files changed, 159 insertions(+), 5 deletions(-) diff --git a/testing/ci/publish_metrics.py b/testing/ci/publish_metrics.py index 8b3959b..4d7f90d 100644 --- a/testing/ci/publish_metrics.py +++ b/testing/ci/publish_metrics.py @@ -3,11 +3,98 @@ from __future__ import annotations """Command-line entry point for publishing CI test metrics.""" import argparse +import json from pathlib import Path +import re +import xml.etree.ElementTree as ET from .summary import load_junit_summary, publish_quality_metrics +def _parse_check_args(raw_checks: list[str]) -> dict[str, str]: + checks: dict[str, str] = {} + for item in raw_checks: + if ":" not in item: + continue + name, status = item.split(":", 1) + normalized_name = name.strip() + normalized_status = status.strip().lower() + if not normalized_name or normalized_status not in {"ok", "failed"}: + continue + checks[normalized_name] = normalized_status + return checks + + +def _derive_quality_report(path: Path) -> tuple[int, dict[str, str]]: + if not path.exists(): + return 0, {} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return 0, {} + issues = payload.get("issues") if isinstance(payload, dict) else [] + if not isinstance(issues, list): + return 0, {} + over_500 = 0 + checks = {"docs": "ok", "loc": "ok", "coverage": "ok"} + for issue in issues: + if not isinstance(issue, dict): + continue + check_name = str(issue.get("check", "")) + if check_name == "docstring": + checks["docs"] = "failed" + elif check_name == "loc": + checks["loc"] = "failed" + over_500 += 1 + elif check_name == "coverage": + checks["coverage"] = "failed" + return over_500, checks + + +def _load_backend_coverage_percent(path: Path) -> float: + if not path.exists(): + return 0.0 + try: + root = ET.parse(path).getroot() + except ET.ParseError: + return 0.0 + raw = root.attrib.get("line-rate") + if raw is None: + return 0.0 + try: + return float(raw) * 100.0 + except ValueError: + return 0.0 + + +def _load_frontend_coverage_percent(path: Path) -> float: + if not path.exists(): + return 0.0 + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return 0.0 + total = payload.get("total") if isinstance(payload, dict) else {} + lines = total.get("lines") if isinstance(total, dict) else {} + pct = lines.get("pct") if isinstance(lines, dict) else None + if isinstance(pct, (int, float)): + return float(pct) + if isinstance(pct, str): + match = re.search(r"[0-9]+(?:\.[0-9]+)?", pct) + if match: + return float(match.group(0)) + return 0.0 + + +def _workspace_coverage_percent(backend_path: Path, frontend_path: Path) -> float: + backend = _load_backend_coverage_percent(backend_path) + frontend = _load_frontend_coverage_percent(frontend_path) + values = [value for value in (backend, frontend) if value > 0] + if not values: + return 0.0 + return sum(values) / len(values) + + def _build_parser() -> argparse.ArgumentParser: """Build the CLI parser for the metrics publisher.""" @@ -17,6 +104,16 @@ def _build_parser() -> argparse.ArgumentParser: parser.add_argument("--job", default="platform-quality-ci", help="Pushgateway job label") parser.add_argument("--status", choices=("ok", "failed"), required=True, help="Gate outcome") parser.add_argument("--junit", nargs="*", default=(), help="JUnit XML files to aggregate") + parser.add_argument("--quality-report", default="build/quality-gate.json", help="Quality-gate JSON report") + parser.add_argument("--backend-coverage", default="build/backend-coverage.xml", help="Backend coverage XML") + parser.add_argument( + "--frontend-coverage", + default="frontend/coverage/coverage-summary.json", + help="Frontend coverage summary JSON", + ) + parser.add_argument("--coverage-percent", type=float, help="Override workspace coverage percent") + parser.add_argument("--source-lines-over-500", type=int, help="Override count of source files over 500 LOC") + parser.add_argument("--check", action="append", default=[], help="check_name:ok|failed") return parser @@ -26,12 +123,29 @@ def main(argv: list[str] | None = None) -> int: parser = _build_parser() args = parser.parse_args(argv) summary = load_junit_summary(Path(path) for path in args.junit) + report_over_500, report_checks = _derive_quality_report(Path(args.quality_report)) + source_lines_over_500 = args.source_lines_over_500 if args.source_lines_over_500 is not None else report_over_500 + coverage_percent = ( + args.coverage_percent + if args.coverage_percent is not None + else _workspace_coverage_percent(Path(args.backend_coverage), Path(args.frontend_coverage)) + ) + checks = { + "tests": "ok" if summary.tests > 0 and summary.failures == 0 and summary.errors == 0 else "failed", + "coverage": "ok" if coverage_percent >= 95.0 else "failed", + "loc": "ok" if source_lines_over_500 == 0 else "failed", + } + checks.update(report_checks) + checks.update(_parse_check_args(args.check)) publish_quality_metrics( gateway=args.gateway, suite=args.suite, job=args.job, status=args.status, summary=summary, + coverage_percent=coverage_percent, + source_lines_over_500=source_lines_over_500, + checks=checks, ) return 0 diff --git a/testing/ci/summary.py b/testing/ci/summary.py index 335be54..6ff7e71 100644 --- a/testing/ci/summary.py +++ b/testing/ci/summary.py @@ -63,10 +63,19 @@ def read_pushgateway_counters(text: str, *, suite: str, job: str) -> dict[str, f return counters -def render_payload(*, suite: str, ok: int, failed: int, summary: RunSummary) -> str: +def render_payload( + *, + suite: str, + ok: int, + failed: int, + summary: RunSummary, + coverage_percent: float = 0.0, + source_lines_over_500: int = 0, + checks: dict[str, str] | None = None, +) -> str: """Render the Pushgateway payload for the quality-gate counters.""" - return ( + lines = [ "# TYPE platform_quality_gate_runs_total counter\n" f'platform_quality_gate_runs_total{{suite="{suite}",status="ok"}} {ok}\n' f'platform_quality_gate_runs_total{{suite="{suite}",status="failed"}} {failed}\n' @@ -75,10 +84,33 @@ def render_payload(*, suite: str, ok: int, failed: int, summary: RunSummary) -> f'bstein_home_quality_gate_tests_total{{suite="{suite}",result="failed"}} {summary.failures}\n' f'bstein_home_quality_gate_tests_total{{suite="{suite}",result="error"}} {summary.errors}\n' f'bstein_home_quality_gate_tests_total{{suite="{suite}",result="skipped"}} {summary.skipped}\n' - ) + "# TYPE platform_quality_gate_workspace_line_coverage_percent gauge\n" + f'platform_quality_gate_workspace_line_coverage_percent{{suite="{suite}"}} {coverage_percent:.3f}\n' + "# TYPE platform_quality_gate_source_lines_over_500_total gauge\n" + f'platform_quality_gate_source_lines_over_500_total{{suite="{suite}"}} {max(source_lines_over_500, 0)}\n' + "# TYPE bstein_home_quality_gate_checks_total gauge\n" + ] + merged_checks = checks or {} + for check_name, check_status in merged_checks.items(): + if check_status not in {"ok", "failed"}: + continue + lines.append( + f'bstein_home_quality_gate_checks_total{{suite="{suite}",check="{check_name}",result="{check_status}"}} 1\n' + ) + return "".join(lines) -def publish_quality_metrics(*, gateway: str, suite: str, job: str, status: str, summary: RunSummary) -> None: +def publish_quality_metrics( + *, + gateway: str, + suite: str, + job: str, + status: str, + summary: RunSummary, + coverage_percent: float = 0.0, + source_lines_over_500: int = 0, + checks: dict[str, str] | None = None, +) -> None: """Publish run and test totals to Pushgateway.""" gateway = gateway.rstrip("/") @@ -89,7 +121,15 @@ def publish_quality_metrics(*, gateway: str, suite: str, job: str, status: str, else: counters["failed"] += 1 - payload = render_payload(suite=suite, ok=int(counters["ok"]), failed=int(counters["failed"]), summary=summary) + payload = render_payload( + suite=suite, + ok=int(counters["ok"]), + failed=int(counters["failed"]), + summary=summary, + coverage_percent=coverage_percent, + source_lines_over_500=source_lines_over_500, + checks=checks, + ) req = urllib.request.Request( f"{gateway}/metrics/job/{job}/suite/{suite}", data=payload.encode("utf-8"),