bstein/bstein-dev-home
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ci(metrics): derive checks plus workspace coverage/loc gauges

Browse Source
This commit is contained in:
Brad Stein 2026-04-18 16:33:52 -03:00
parent 8245e1aaa7
commit f606c0543c
2 changed files with 159 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
testing/ci/publish_metrics.py
View File

@ -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

48
testing/ci/summary.py
View File

@ -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"),