from __future__ import annotations """Parse test results and format Pushgateway-friendly metrics payloads.""" from dataclasses import dataclass import re import urllib.request import xml.etree.ElementTree as ET from pathlib import Path from typing import Iterable def _escape_label(value: str) -> str: """Escape Prometheus label values.""" return value.replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"') @dataclass(frozen=True) class RunSummary: """Aggregate counts from a collection of JUnit XML files.""" tests: int = 0 failures: int = 0 errors: int = 0 skipped: int = 0 @property def passed(self) -> int: """Return the number of passing test cases derived from the aggregate.""" return max(self.tests - self.failures - self.errors - self.skipped, 0) def load_junit_summary(paths: Iterable[Path]) -> RunSummary: """Load one or more JUnit XML files and combine their result counts. WHY: CI needs a single stable view of each suite instead of separately parsing backend, unit, component, and e2e XML files in shell. """ totals = {"tests": 0, "failures": 0, "errors": 0, "skipped": 0} 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 node in suites: for key in totals: raw = node.attrib.get(key) or "0" try: totals[key] += int(float(raw)) except ValueError: continue return RunSummary(**totals) def load_junit_cases(paths: Iterable[Path]) -> list[tuple[str, str]]: """Load per-test outcomes from one or more JUnit XML files.""" 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.""" counters: dict[str, float] = {"ok": 0.0, "failed": 0.0} for status in counters: pattern = re.compile( rf'^platform_quality_gate_runs_total\{{[^}}]*job="{re.escape(job)}"[^}}]*suite="{re.escape(suite)}"[^}}]*status="{status}"[^}}]*\}}\s+([0-9]+(?:\.[0-9]+)?)$', re.M, ) match = pattern.search(text) if match: counters[status] = float(match.group(1)) return counters def render_payload( *, suite: str, ok: int, failed: int, 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.""" 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' "# TYPE bstein_home_quality_gate_tests_total gauge\n" f'bstein_home_quality_gate_tests_total{{suite="{suite}",result="passed"}} {summary.passed}\n' f'bstein_home_quality_gate_tests_total{{suite="{suite}",result="failed"}} {summary.failures}\n' f'bstein_home_quality_gate_tests_total{{suite="{suite}",result="error"}} {summary.errors}\n' f'bstein_home_quality_gate_tests_total{{suite="{suite}",result="skipped"}} {summary.skipped}\n' "# TYPE platform_quality_gate_workspace_line_coverage_percent gauge\n" f'platform_quality_gate_workspace_line_coverage_percent{{suite="{suite}"}} {workspace_line_coverage_percent:.3f}\n' "# 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() ) payload += "# TYPE platform_quality_gate_test_case_result gauge\n" if test_cases: 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 ) else: payload += f'platform_quality_gate_test_case_result{{suite="{suite}",test="__no_test_cases__",status="skipped"}} 1\n' return payload def publish_quality_metrics( *, gateway: str, suite: str, job: str, status: str, 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.""" gateway = gateway.rstrip("/") text = urllib.request.urlopen(f"{gateway}/metrics", timeout=10).read().decode("utf-8", errors="replace") counters = read_pushgateway_counters(text, suite=suite, job=job) if status == "ok": counters["ok"] += 1 else: counters["failed"] += 1 payload = render_payload( suite=suite, ok=int(counters["ok"]), failed=int(counters["failed"]), 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="PUT", headers={"Content-Type": "text/plain"}, ) urllib.request.urlopen(req, timeout=10).read()