ci(atlasbot): fix test-case metric label escaping
This commit is contained in:
parent
6218c4593d
commit
4fa280ca8d
@ -23,6 +23,11 @@ from pathlib import Path
|
|||||||
QUALITY_SUCCESS_STATES = {"ok", "pass", "passed", "success", "compliant"}
|
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:
|
def _as_int(node: ET.Element, name: str) -> int:
|
||||||
raw = node.attrib.get(name) or "0"
|
raw = node.attrib.get(name) or "0"
|
||||||
try:
|
try:
|
||||||
@ -54,6 +59,39 @@ def _load_junit(path: Path) -> dict[str, int]:
|
|||||||
return totals
|
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:
|
def _load_coverage_percent(path: Path) -> float:
|
||||||
if not path.exists():
|
if not path.exists():
|
||||||
return 0.0
|
return 0.0
|
||||||
@ -161,7 +199,7 @@ def _post_text(url: str, payload: str) -> None:
|
|||||||
req = urllib.request.Request(
|
req = urllib.request.Request(
|
||||||
url,
|
url,
|
||||||
data=payload.encode("utf-8"),
|
data=payload.encode("utf-8"),
|
||||||
method="POST",
|
method="PUT",
|
||||||
headers={"Content-Type": "text/plain"},
|
headers={"Content-Type": "text/plain"},
|
||||||
)
|
)
|
||||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
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"))
|
junit_path = Path(os.getenv("JUNIT_PATH", "build/junit.xml"))
|
||||||
coverage_path = Path(os.getenv("COVERAGE_PATH", "build/coverage.json"))
|
coverage_path = Path(os.getenv("COVERAGE_PATH", "build/coverage.json"))
|
||||||
gate_rc_path = Path(os.getenv("QUALITY_GATE_RC_PATH", "build/quality-gate.rc"))
|
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"))
|
source_root = Path(os.getenv("SOURCE_ROOT", "atlasbot"))
|
||||||
build_dir = Path(os.getenv("BUILD_DIR", "build"))
|
build_dir = Path(os.getenv("BUILD_DIR", "build"))
|
||||||
|
|
||||||
totals = _load_junit(junit_path)
|
totals = _load_junit(junit_path)
|
||||||
|
test_cases = _load_junit_cases(junit_path)
|
||||||
coverage_pct = _load_coverage_percent(coverage_path)
|
coverage_pct = _load_coverage_percent(coverage_path)
|
||||||
gate_rc = _load_gate_rc(gate_rc_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)
|
source_lines_over_500 = _count_source_lines_over_500(source_root)
|
||||||
passed = max(totals["tests"] - totals["failures"] - totals["errors"] - totals["skipped"], 0)
|
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"
|
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",
|
"tests": "ok" if outcome == "ok" else "failed",
|
||||||
"coverage": "ok" if coverage_pct >= 95.0 else "failed",
|
"coverage": "ok" if coverage_pct >= 95.0 else "failed",
|
||||||
"loc": "ok" if source_lines_over_500 == 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",
|
"gate_glue": "ok",
|
||||||
"sonarqube": _sonarqube_check_status(build_dir),
|
"sonarqube": _sonarqube_check_status(build_dir),
|
||||||
"supply_chain": _supply_chain_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",
|
"# 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}',
|
f'platform_quality_gate_source_lines_over_500_total{{suite="{suite}"}} {source_lines_over_500}',
|
||||||
"# TYPE atlasbot_quality_gate_checks_total gauge",
|
"# TYPE atlasbot_quality_gate_checks_total gauge",
|
||||||
|
"# TYPE platform_quality_gate_test_case_result gauge",
|
||||||
]
|
]
|
||||||
) + "\n"
|
) + "\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(
|
payload += "\n".join(
|
||||||
f'atlasbot_quality_gate_checks_total{{suite="{suite}",check="{check_name}",result="{check_status}"}} 1'
|
f'atlasbot_quality_gate_checks_total{{suite="{suite}",check="{check_name}",result="{check_status}"}} 1'
|
||||||
for check_name, check_status in checks.items()
|
for check_name, check_status in checks.items()
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user