ci(bstein-home): publish per-test result metrics

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
 

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()

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(
+        (
+            "<testsuite>"
+            '<testcase classname="app.health" name="test_ok" />'
+            '<testcase classname="app.health" name="test_fail"><failure/></testcase>'
+            "</testsuite>"
+        ),
+        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