ci(ananke): emit per-test case status metrics

scripts/publish_quality_metrics.py

@@ -45,7 +45,7 @@ def _post_text(url: str, payload: str, timeout_seconds: float, attempts: int, re
         req = urllib.request.Request(
             url,
             data=payload.encode("utf-8"),
-            method="POST",
+            method="PUT",
             headers={"Content-Type": "text/plain"},
         )
         try:
@@ -87,6 +87,7 @@ def _build_payload(
     tests_failed: int,
     tests_errors: int,
     tests_skipped: int,
+    test_cases: list[tuple[str, str]],
     coverage_percent: float,
     source_lines_over_500: int,
     checks: dict[str, str],
@@ -106,10 +107,15 @@ def _build_payload(
         f'platform_quality_gate_workspace_line_coverage_percent{{suite="{suite}"}} {coverage_percent:.3f}',
         "# TYPE platform_quality_gate_source_lines_over_500_total gauge",
         f'platform_quality_gate_source_lines_over_500_total{{suite="{suite}"}} {source_lines_over_500}',
+        "# TYPE platform_quality_gate_test_case_result gauge",
         "# TYPE ananke_quality_gate_checks_total gauge",
         "# TYPE ananke_quality_gate_publish_info gauge",
         f'ananke_quality_gate_publish_info{_label_str({"suite": suite, "trigger": trigger})} 1',
     ]
+    lines.extend(
+        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
+    )
     lines.extend(
         f'ananke_quality_gate_checks_total{{suite="{suite}",check="{check_name}",result="{check_status}"}} 1'
         for check_name, check_status in checks.items()
@@ -159,6 +165,18 @@ def _parse_go_test_counts(output_path: Path) -> dict[str, int]:
     }
 
 
+def _parse_go_test_cases(output_path: Path) -> list[tuple[str, str]]:
+    if not output_path.exists():
+        return []
+    text = output_path.read_text(encoding="utf-8", errors="ignore")
+    cases: list[tuple[str, str]] = []
+    for match in re.finditer(r"^---\s+(PASS|FAIL|SKIP):\s+(\S+)", text, flags=re.M):
+        raw_status, test_name = match.groups()
+        status = {"PASS": "passed", "FAIL": "failed", "SKIP": "skipped"}.get(raw_status, "error")
+        cases.append((test_name.strip(), status))
+    return cases
+
+
 def _read_exit_code(path: Path) -> int:
     if not path.exists():
         return 1
@@ -169,6 +187,17 @@ def _read_exit_code(path: Path) -> int:
         return 1
 
 
+def _read_status(path: Path, default: str = "failed") -> str:
+    if not path.exists():
+        return default
+    raw = path.read_text(encoding="utf-8").strip().lower()
+    if raw in {"ok", "pass", "passed", "success"}:
+        return "ok"
+    if raw in {"failed", "fail", "error"}:
+        return "failed"
+    return default
+
+
 def _load_json(path: Path) -> dict | None:
     if not path.exists():
         return None
@@ -277,14 +306,17 @@ def main(argv: list[str] | None = None) -> int:
     resolved_failed = max(args.local_failed, remote_failed)
     coverage_percent = _read_coverage_percent(args.coverage_percent_file)
     source_lines_over_500 = _count_source_files_over_limit(repo_root, max_lines=500)
-    tests = _parse_go_test_counts(Path(os.getenv("ANANKE_QUALITY_OUTPUT_FILE", str(build_dir / "quality-gate.out"))))
+    test_output = Path(os.getenv("ANANKE_QUALITY_OUTPUT_FILE", str(build_dir / "quality-gate.out")))
+    tests = _parse_go_test_counts(test_output)
+    test_cases = _parse_go_test_cases(test_output)
     gate_rc = _read_exit_code(Path(os.getenv("ANANKE_QUALITY_EXIT_CODE_PATH", str(build_dir / "quality-gate.rc"))))
+    docs_status = _read_status(Path(os.getenv("ANANKE_QUALITY_DOCS_STATUS_PATH", str(build_dir / "docs-naming.status"))))
     gate_failed = gate_rc != 0
     checks = {
         "tests": "failed" if gate_failed or tests["failed"] > 0 else "ok",
         "coverage": "ok" if coverage_percent >= 95.0 else "failed",
         "loc": "ok" if source_lines_over_500 == 0 else "failed",
-        "docs_naming": "not_applicable",
+        "docs_naming": docs_status,
         "gate_glue": "ok",
         "sonarqube": _sonarqube_check_status(build_dir),
         "supply_chain": _supply_chain_check_status(build_dir),
@@ -298,6 +330,7 @@ def main(argv: list[str] | None = None) -> int:
         tests_failed=tests["failed"],
         tests_errors=tests["errors"],
         tests_skipped=tests["skipped"],
+        test_cases=test_cases,
         coverage_percent=coverage_percent,
         source_lines_over_500=source_lines_over_500,
         checks=checks,