ci(pegasus): publish per-test case metrics for flaky tracking

scripts/publish_test_metrics.py

@@ -72,6 +72,37 @@ def _load_junit(path: Path) -> dict[str, int]:
     return totals
 
 
+def _load_junit_cases(path: Path) -> list[tuple[str, str]]:
+    """Return per-test outcomes from a JUnit report."""
+    if not path.exists():
+        return []
+    tree = ET.parse(path)
+    root = tree.getroot()
+    suites: list[ET.Element]
+    if root.tag == "testsuite":
+        suites = [root]
+    elif root.tag == "testsuites":
+        suites = list(root.findall("testsuite"))
+    else:
+        suites = []
+
+    cases: list[tuple[str, str]] = []
+    for suite in suites:
+        for test_case in suite.findall("testcase"):
+            case_name = (test_case.attrib.get("name") or "").strip()
+            class_name = (test_case.attrib.get("classname") or "").strip()
+            if not case_name:
+                continue
+            full_name = f"{class_name}.{case_name}" if class_name else case_name
+            status = "passed"
+            if test_case.find("failure") is not None or test_case.find("error") is not None:
+                status = "failed"
+            elif test_case.find("skipped") is not None:
+                status = "skipped"
+            cases.append((full_name, status))
+    return cases
+
+
 def _load_backend_coverage_percent(path: Path) -> float:
     if not path.exists():
         return 0.0
@@ -228,6 +259,9 @@ def main() -> int:
 
     b = _load_junit(backend_junit)
     f = _load_junit(frontend_junit)
+    backend_cases = _load_junit_cases(backend_junit)
+    frontend_cases = _load_junit_cases(frontend_junit)
+    test_cases = backend_cases + frontend_cases
     totals = {
         "tests": b["tests"] + f["tests"],
         "failures": b["failures"] + f["failures"],
@@ -324,6 +358,16 @@ def main() -> int:
         f'pegasus_quality_gate_checks_total{{suite="{suite}",check="{check_name}",result="{check_status}"}} 1'
         for check_name, check_status in checks.items()
     )
+    if test_cases:
+        payload_lines.extend(
+            [
+                "# TYPE platform_quality_gate_test_case_result gauge",
+                *[
+                    f'platform_quality_gate_test_case_result{{suite="{suite}",test="{_escape_label(test_name)}",status="{_escape_label(test_status)}"}} 1'
+                    for test_name, test_status in test_cases
+                ],
+            ]
+        )
     payload = "\n".join(payload_lines) + "\n"
 
     push_url = f"{pushgateway_url.rstrip('/')}/metrics/job/{job_name}/suite/{suite}"