175 lines
6.7 KiB
Python
175 lines
6.7 KiB
Python
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 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 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()
|
|
)
|
|
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(
|
|
*,
|
|
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()
|