From 1dd0a48887eb85fd57f97a74c28155ffbbe466d7 Mon Sep 17 00:00:00 2001 From: codex Date: Mon, 20 Apr 2026 11:53:59 -0300 Subject: [PATCH] ci(bstein-home): publish per-test result metrics --- testing/ci/publish_metrics.py | 7 +++-- testing/ci/summary.py | 42 ++++++++++++++++++++++++++- testing/tests/test_publish_metrics.py | 28 +++++++++++++++++- 3 files changed, 73 insertions(+), 4 deletions(-) diff --git a/testing/ci/publish_metrics.py b/testing/ci/publish_metrics.py index a079273..c59dc3c 100644 --- a/testing/ci/publish_metrics.py +++ b/testing/ci/publish_metrics.py @@ -7,7 +7,7 @@ import json import os from pathlib import Path -from .summary import load_junit_summary, publish_quality_metrics +from .summary import load_junit_cases, load_junit_summary, publish_quality_metrics def _build_parser() -> argparse.ArgumentParser: @@ -92,7 +92,9 @@ 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) + junit_paths = [Path(path) for path in args.junit] + summary = load_junit_summary(junit_paths) + test_cases = load_junit_cases(junit_paths) coverage_percent, source_lines_over_500, checks = _load_quality_report(Path(args.quality_report)) publish_quality_metrics( gateway=args.gateway, @@ -103,6 +105,7 @@ def main(argv: list[str] | None = None) -> int: workspace_line_coverage_percent=coverage_percent, source_lines_over_500=source_lines_over_500, checks=checks, + test_cases=test_cases, ) return 0 diff --git a/testing/ci/summary.py b/testing/ci/summary.py index af81cea..cb9e6a1 100644 --- a/testing/ci/summary.py +++ b/testing/ci/summary.py @@ -10,6 +10,11 @@ from pathlib import Path from typing import Iterable +def _escape_label(value: str) -> str: + """Escape text for safe Prometheus label emission.""" + return value.replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"') + + @dataclass(frozen=True) class RunSummary: """Aggregate counts from a collection of JUnit XML files.""" @@ -48,6 +53,32 @@ def load_junit_summary(paths: Iterable[Path]) -> RunSummary: return RunSummary(**totals) +def load_junit_cases(paths: Iterable[Path]) -> list[tuple[str, str]]: + """Collect testcase-level statuses for flaky-test visibility panels.""" + cases: list[tuple[str, str]] = [] + for path in paths: + if not path.exists(): + continue + root = ET.parse(path).getroot() + suites = [root] if root.tag == "testsuite" else list(root.findall("testsuite")) if root.tag == "testsuites" else [] + for suite in suites: + for case in suite.findall("testcase"): + name = (case.attrib.get("name") or "").strip() + classname = (case.attrib.get("classname") or "").strip() + if not name: + continue + test_id = f"{classname}::{name}" if classname else name + status = "passed" + if case.find("failure") is not None: + status = "failed" + elif case.find("error") is not None: + status = "error" + elif case.find("skipped") is not None: + status = "skipped" + cases.append((test_id, status)) + return cases + + def read_pushgateway_counters(text: str, *, suite: str, job: str) -> dict[str, float]: """Read the current quality-gate counters for a suite from Pushgateway text.""" @@ -72,6 +103,7 @@ def render_payload( workspace_line_coverage_percent: float = 0.0, source_lines_over_500: int = 0, checks: dict[str, str] | None = None, + test_cases: list[tuple[str, str]] | None = None, ) -> str: """Render the Pushgateway payload for the quality-gate counters.""" payload = ( @@ -94,6 +126,12 @@ def render_payload( f'bstein_home_quality_gate_checks_total{{suite="{suite}",check="{check_name}",result="{check_status}"}} 1\n' for check_name, check_status in checks.items() ) + if test_cases: + payload += "# TYPE platform_quality_gate_test_case_result gauge\n" + payload += "".join( + f'platform_quality_gate_test_case_result{{suite="{suite}",test="{_escape_label(test_name)}",status="{_escape_label(test_status)}"}} 1\n' + for test_name, test_status in test_cases + ) return payload @@ -107,6 +145,7 @@ def publish_quality_metrics( workspace_line_coverage_percent: float = 0.0, source_lines_over_500: int = 0, checks: dict[str, str] | None = None, + test_cases: list[tuple[str, str]] | None = None, ) -> None: """Publish run and test totals to Pushgateway.""" @@ -126,11 +165,12 @@ def publish_quality_metrics( workspace_line_coverage_percent=workspace_line_coverage_percent, source_lines_over_500=source_lines_over_500, checks=checks, + test_cases=test_cases, ) req = urllib.request.Request( f"{gateway}/metrics/job/{job}/suite/{suite}", data=payload.encode("utf-8"), - method="POST", + method="PUT", headers={"Content-Type": "text/plain"}, ) urllib.request.urlopen(req, timeout=10).read() diff --git a/testing/tests/test_publish_metrics.py b/testing/tests/test_publish_metrics.py index 66925b4..d9e5fec 100644 --- a/testing/tests/test_publish_metrics.py +++ b/testing/tests/test_publish_metrics.py @@ -2,7 +2,7 @@ from __future__ import annotations from pathlib import Path -from testing.ci.summary import RunSummary, load_junit_summary, render_payload +from testing.ci.summary import RunSummary, load_junit_cases, load_junit_summary, render_payload def test_load_junit_summary_combines_suites(tmp_path: Path) -> None: @@ -19,3 +19,29 @@ def test_load_junit_summary_combines_suites(tmp_path: Path) -> None: assert 'bstein_home_quality_gate_tests_total{suite="bstein_home",result="skipped"} 1' in payload assert 'platform_quality_gate_workspace_line_coverage_percent{suite="bstein_home"} 0.000' in payload assert 'platform_quality_gate_source_lines_over_500_total{suite="bstein_home"} 0' in payload + + +def test_load_junit_cases_and_render_test_case_metrics(tmp_path: Path) -> None: + junit = tmp_path / "cases.xml" + junit.write_text( + ( + "" + '' + '' + "" + ), + encoding="utf-8", + ) + + cases = load_junit_cases([junit]) + assert ("app.health::test_ok", "passed") in cases + assert ("app.health::test_fail", "failed") in cases + + payload = render_payload( + suite="bstein_home", + ok=1, + failed=0, + summary=RunSummary(tests=2, failures=1, errors=0, skipped=0), + test_cases=cases, + ) + assert 'platform_quality_gate_test_case_result{suite="bstein_home",test="app.health::test_fail",status="failed"} 1' in payload