From 4fa280ca8de45ab1b6ba8d542c786464ca3699c0 Mon Sep 17 00:00:00 2001 From: jenkins Date: Mon, 20 Apr 2026 10:46:07 -0300 Subject: [PATCH] ci(atlasbot): fix test-case metric label escaping --- scripts/publish_test_metrics.py | 50 +++++++++++++++++++++++++++++++-- 1 file changed, 48 insertions(+), 2 deletions(-) diff --git a/scripts/publish_test_metrics.py b/scripts/publish_test_metrics.py index e70d91c..35f2655 100755 --- a/scripts/publish_test_metrics.py +++ b/scripts/publish_test_metrics.py @@ -23,6 +23,11 @@ from pathlib import Path QUALITY_SUCCESS_STATES = {"ok", "pass", "passed", "success", "compliant"} +def _escape_label(value: str) -> str: + """Escape Prometheus label values safely.""" + return value.replace("\\", "\\\\").replace("\n", "\\n").replace('"', '\\"') + + def _as_int(node: ET.Element, name: str) -> int: raw = node.attrib.get(name) or "0" try: @@ -54,6 +59,39 @@ def _load_junit(path: Path) -> dict[str, int]: return totals +def _load_junit_cases(path: Path) -> list[tuple[str, str]]: + 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 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 _load_coverage_percent(path: Path) -> float: if not path.exists(): return 0.0 @@ -161,7 +199,7 @@ def _post_text(url: str, payload: str) -> None: req = urllib.request.Request( url, data=payload.encode("utf-8"), - method="POST", + method="PUT", headers={"Content-Type": "text/plain"}, ) with urllib.request.urlopen(req, timeout=10) as resp: @@ -178,12 +216,15 @@ def main() -> int: junit_path = Path(os.getenv("JUNIT_PATH", "build/junit.xml")) coverage_path = Path(os.getenv("COVERAGE_PATH", "build/coverage.json")) gate_rc_path = Path(os.getenv("QUALITY_GATE_RC_PATH", "build/quality-gate.rc")) + docs_rc_path = Path(os.getenv("QUALITY_GATE_DOCS_RC_PATH", "build/docs-naming.rc")) source_root = Path(os.getenv("SOURCE_ROOT", "atlasbot")) build_dir = Path(os.getenv("BUILD_DIR", "build")) totals = _load_junit(junit_path) + test_cases = _load_junit_cases(junit_path) coverage_pct = _load_coverage_percent(coverage_path) gate_rc = _load_gate_rc(gate_rc_path) + docs_rc = _load_gate_rc(docs_rc_path) source_lines_over_500 = _count_source_lines_over_500(source_root) passed = max(totals["tests"] - totals["failures"] - totals["errors"] - totals["skipped"], 0) outcome = "ok" if totals["tests"] > 0 and totals["failures"] == 0 and totals["errors"] == 0 else "failed" @@ -193,7 +234,7 @@ def main() -> int: "tests": "ok" if outcome == "ok" else "failed", "coverage": "ok" if coverage_pct >= 95.0 else "failed", "loc": "ok" if source_lines_over_500 == 0 else "failed", - "docs_naming": "not_applicable", + "docs_naming": "ok" if docs_rc == 0 else "failed", "gate_glue": "ok", "sonarqube": _sonarqube_check_status(build_dir), "supply_chain": _supply_chain_check_status(build_dir), @@ -224,8 +265,13 @@ def main() -> int: "# 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 atlasbot_quality_gate_checks_total gauge", + "# TYPE platform_quality_gate_test_case_result gauge", ] ) + "\n" + payload += "\n".join( + 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 + ) + "\n" payload += "\n".join( f'atlasbot_quality_gate_checks_total{{suite="{suite}",check="{check_name}",result="{check_status}"}} 1' for check_name, check_status in checks.items()