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

Jenkinsfile

@@ -80,8 +80,10 @@ spec:
     BACK_IMAGE  = "${REGISTRY}/bstein-dev-home-backend"
     VERSION_TAG = 'dev'
     SEMVER = 'dev'
-    SUITE_NAME = 'bstein-home'
+    SUITE_NAME = 'bstein_home'
     PUSHGATEWAY_URL = 'http://platform-quality-gateway.monitoring.svc.cluster.local:9091'
+    QUALITY_GATE_SONARQUBE_REPORT = 'build/sonarqube-quality-gate.json'
+    QUALITY_GATE_IRONBANK_REPORT = 'build/ironbank-compliance.json'
   }
   options {
     disableConcurrentBuilds()
@@ -97,6 +99,73 @@ spec:
       }
     }
 
+    stage('Collect SonarQube evidence') {
+      steps {
+        container('tester') {
+          sh '''
+            set -euo pipefail
+            mkdir -p build
+            python3 - <<'PY'
+import base64
+import json
+import os
+import urllib.parse
+import urllib.request
+
+host = os.getenv('SONARQUBE_HOST_URL', '').strip().rstrip('/')
+project_key = os.getenv('SONARQUBE_PROJECT_KEY', '').strip()
+token = os.getenv('SONARQUBE_TOKEN', '').strip()
+report_path = os.getenv('QUALITY_GATE_SONARQUBE_REPORT', 'build/sonarqube-quality-gate.json')
+payload = {"status": "ERROR", "note": "missing SONARQUBE_HOST_URL and/or SONARQUBE_PROJECT_KEY"}
+if host and project_key:
+    query = urllib.parse.urlencode({"projectKey": project_key})
+    request = urllib.request.Request(f"{host}/api/qualitygates/project_status?{query}", method="GET")
+    if token:
+        encoded = base64.b64encode(f"{token}:".encode("utf-8")).decode("utf-8")
+        request.add_header("Authorization", f"Basic {encoded}")
+    try:
+        with urllib.request.urlopen(request, timeout=12) as response:
+            payload = json.loads(response.read().decode("utf-8"))
+    except Exception as exc:  # noqa: BLE001
+        payload = {"status": "ERROR", "error": str(exc)}
+with open(report_path, "w", encoding="utf-8") as handle:
+    json.dump(payload, handle, indent=2, sort_keys=True)
+    handle.write("\\n")
+PY
+          '''
+        }
+      }
+    }
+
+    stage('Collect Supply Chain evidence') {
+      steps {
+        container('tester') {
+          sh '''
+            set -euo pipefail
+            mkdir -p build
+            python3 - <<'PY'
+import json
+import os
+from pathlib import Path
+
+report_path = Path(os.getenv('QUALITY_GATE_IRONBANK_REPORT', 'build/ironbank-compliance.json'))
+if report_path.exists():
+    raise SystemExit(0)
+status = os.getenv('IRONBANK_COMPLIANCE_STATUS', '').strip()
+compliant = os.getenv('IRONBANK_COMPLIANT', '').strip().lower()
+payload = {"status": status or "unknown", "compliant": compliant in {"1", "true", "yes", "on"} if compliant else None}
+payload = {k: v for k, v in payload.items() if v is not None}
+if "status" not in payload:
+    payload["status"] = "unknown"
+payload["note"] = "Set IRONBANK_COMPLIANCE_STATUS/IRONBANK_COMPLIANT or write build/ironbank-compliance.json in image-building repos."
+report_path.parent.mkdir(parents=True, exist_ok=True)
+report_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\\n", encoding="utf-8")
+PY
+          '''
+        }
+      }
+    }
+
     stage('Prep toolchain') {
       steps {
         container('builder') {
@@ -193,30 +262,67 @@ spec:
     stage('Frontend tests') {
       steps {
         container('frontend') {
-          sh '''
-            set -euo pipefail
-            mkdir -p build
-            cd frontend
-            npm ci
-            npm run lint
-            npm run test:unit
-            npm run test:component
-            npm run test:e2e
-          '''
+          sh(script: '''#!/usr/bin/env bash
+set -euo pipefail
+mkdir -p build
+cd frontend
+npm ci
+npm run lint
+npm run test:unit
+npm run test:component
+npm run test:e2e
+''')
         }
       }
     }
 
-    stage('Unified quality gate') {
+    stage('Run quality gate') {
       steps {
         container('tester') {
           sh '''
             set -euo pipefail
             export PYTHONPATH="${WORKSPACE}:${PYTHONPATH:-}"
+            set +e
             python -m testing.ci.quality_gate \
               --backend-coverage build/backend-coverage.xml \
               --frontend-coverage frontend/coverage/coverage-summary.json \
               --report build/quality-gate.json
+            gate_rc=$?
+            set -e
+            printf '%s\n' "${gate_rc}" > build/quality-gate.rc
+          '''
+        }
+      }
+    }
+
+    stage('Publish test metrics') {
+      steps {
+        container('tester') {
+          sh '''
+            set -euo pipefail
+            gate_rc="$(cat build/quality-gate.rc 2>/dev/null || echo 1)"
+            status="ok"
+            if [ "${gate_rc}" -ne 0 ]; then
+              status="failed"
+            fi
+            python -m testing.ci.publish_metrics \
+              --gateway "${PUSHGATEWAY_URL}" \
+              --suite "${SUITE_NAME}" \
+              --job platform-quality-ci \
+              --status "${status}" \
+              --quality-report build/quality-gate.json \
+              --junit build/junit-backend.xml build/junit-frontend-unit.xml build/junit-frontend-component.xml build/junit-frontend-e2e.xml
+          '''
+        }
+      }
+    }
+
+    stage('Enforce quality gate') {
+      steps {
+        container('tester') {
+          sh '''
+            set -euo pipefail
+            test "$(cat build/quality-gate.rc 2>/dev/null || echo 1)" -eq 0
           '''
         }
       }
@@ -260,34 +366,6 @@ spec:
   }
 
   post {
-    success {
-      container('tester') {
-        sh '''
-          set -euo pipefail
-          python -m testing.ci.publish_metrics \
-            --gateway "${PUSHGATEWAY_URL}" \
-            --suite "${SUITE_NAME}" \
-            --job platform-quality-ci \
-            --status ok \
-            --quality-report build/quality-gate.json \
-            --junit build/junit-backend.xml build/junit-frontend-unit.xml build/junit-frontend-component.xml build/junit-frontend-e2e.xml
-        '''
-      }
-    }
-    failure {
-      container('tester') {
-        sh '''
-          set -euo pipefail
-          python -m testing.ci.publish_metrics \
-            --gateway "${PUSHGATEWAY_URL}" \
-            --suite "${SUITE_NAME}" \
-            --job platform-quality-ci \
-            --status failed \
-            --quality-report build/quality-gate.json \
-            --junit build/junit-backend.xml build/junit-frontend-unit.xml build/junit-frontend-component.xml build/junit-frontend-e2e.xml
-        '''
-      }
-    }
     always {
       script {
         def props = fileExists('build.env') ? readProperties(file: 'build.env') : [:]

testing/ci/publish_metrics.py

@@ -4,9 +4,10 @@ from __future__ import annotations
 
 import argparse
 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:
@@ -22,11 +23,19 @@ def _build_parser() -> argparse.ArgumentParser:
     return parser
 
 
-def _load_quality_report(path: Path) -> tuple[float, int]:
+def _load_quality_report(path: Path) -> tuple[float, int, dict[str, str]]:
     """Read workspace coverage/LOC summary from the quality gate JSON output."""
 
     if not path.exists():
-        return 0.0, 0
+        return 0.0, 0, {
+            "tests": "not_applicable",
+            "coverage": "not_applicable",
+            "loc": "not_applicable",
+            "docs_naming": "not_applicable",
+            "gate_glue": "ok",
+            "sonarqube": "not_applicable",
+            "supply_chain": "not_applicable",
+        }
     payload = json.loads(path.read_text(encoding="utf-8"))
     coverage = payload.get("workspace_line_coverage_percent")
     if not isinstance(coverage, (int, float)):
@@ -34,7 +43,48 @@ def _load_quality_report(path: Path) -> tuple[float, int]:
     source_lines = payload.get("source_lines_over_500")
     if not isinstance(source_lines, int):
         source_lines = 0
-    return float(coverage), int(source_lines)
+    issue_checks = [item.get("check") for item in payload.get("issues", []) if isinstance(item, dict)]
+    docs_failed = any(str(check).lower() in {"docstring", "docs", "naming"} for check in issue_checks)
+    coverage_failed = any(str(check).lower() == "coverage" for check in issue_checks)
+    loc_failed = any(str(check).lower() in {"loc", "smell"} for check in issue_checks) or source_lines > 0
+    checks = {
+        "tests": "ok" if payload.get("issue_count", 0) == 0 else "failed",
+        "coverage": "failed" if coverage_failed or float(coverage) < 95.0 else "ok",
+        "loc": "failed" if loc_failed else "ok",
+        "docs_naming": "failed" if docs_failed else "ok",
+        "gate_glue": "ok",
+        "sonarqube": "not_applicable",
+        "supply_chain": "not_applicable",
+    }
+    sonarqube_report = Path(os.getenv("QUALITY_GATE_SONARQUBE_REPORT", "build/sonarqube-quality-gate.json"))
+    if sonarqube_report.exists():
+        try:
+            sonarqube_payload = json.loads(sonarqube_report.read_text(encoding="utf-8"))
+            status = (
+                sonarqube_payload.get("status")
+                or (sonarqube_payload.get("projectStatus") or {}).get("status")
+                or (sonarqube_payload.get("qualityGate") or {}).get("status")
+            )
+            if isinstance(status, str):
+                checks["sonarqube"] = "ok" if status.strip().lower() in {"ok", "pass", "passed", "success"} else "failed"
+        except Exception:
+            checks["sonarqube"] = "failed"
+    ironbank_report = Path(os.getenv("QUALITY_GATE_IRONBANK_REPORT", "build/ironbank-compliance.json"))
+    if ironbank_report.exists():
+        try:
+            ironbank_payload = json.loads(ironbank_report.read_text(encoding="utf-8"))
+            compliant = ironbank_payload.get("compliant")
+            if isinstance(compliant, bool):
+                checks["supply_chain"] = "ok" if compliant else "failed"
+            else:
+                status = ironbank_payload.get("status") or ironbank_payload.get("result")
+                if isinstance(status, str):
+                    checks["supply_chain"] = (
+                        "ok" if status.strip().lower() in {"ok", "pass", "passed", "success", "compliant"} else "failed"
+                    )
+        except Exception:
+            checks["supply_chain"] = "failed"
+    return float(coverage), int(source_lines), checks
 
 
 def main(argv: list[str] | None = None) -> int:
@@ -42,8 +92,10 @@ 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)
-    coverage_percent, source_lines_over_500 = _load_quality_report(Path(args.quality_report))
+    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,
         suite=args.suite,
@@ -52,6 +104,8 @@ def main(argv: list[str] | None = None) -> int:
         summary=summary,
         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."""
 
@@ -71,10 +102,11 @@ def render_payload(
     summary: RunSummary,
     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."""
-
-    return (
+    payload = (
         "# 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'
@@ -88,6 +120,19 @@ def render_payload(
         "# TYPE platform_quality_gate_source_lines_over_500_total gauge\n"
         f'platform_quality_gate_source_lines_over_500_total{{suite="{suite}"}} {int(source_lines_over_500)}\n'
     )
+    if checks:
+        payload += "# TYPE bstein_home_quality_gate_checks_total gauge\n"
+        payload += "".join(
+            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
 
 
 def publish_quality_metrics(
@@ -99,6 +144,8 @@ def publish_quality_metrics(
     summary: RunSummary,
     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."""
 
@@ -117,11 +164,13 @@ def publish_quality_metrics(
         summary=summary,
         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:
@@ -14,8 +14,34 @@ def test_load_junit_summary_combines_suites(tmp_path: Path) -> None:
     summary = load_junit_summary([junit])
 
     assert summary == RunSummary(tests=3, failures=1, errors=0, skipped=1)
-    payload = render_payload(suite="bstein-home", ok=2, failed=0, summary=summary)
-    assert 'platform_quality_gate_runs_total{suite="bstein-home",status="ok"} 2' in payload
-    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
+    payload = render_payload(suite="bstein_home", ok=2, failed=0, summary=summary)
+    assert 'platform_quality_gate_runs_total{suite="bstein_home",status="ok"} 2' in payload
+    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