bstein/bstein-dev-home
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ci: harden quality metrics freshness

Browse Source
This commit is contained in:
codex 2026-05-11 13:22:22 -03:00
parent f6ef942cd1
commit 65d5212072
3 changed files with 82 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
testing/ci/publish_metrics.py
View File

@ -76,6 +76,7 @@ def _load_quality_report(path: Path) -> tuple[float, int, dict[str, str]]:
except Exception:
checks["sonarqube"] = "failed"
ironbank_report = Path(os.getenv("QUALITY_GATE_IRONBANK_REPORT", "build/ironbank-compliance.json"))
ironbank_required = os.getenv("QUALITY_GATE_IRONBANK_REQUIRED", "1").strip().lower() in {"1", "true", "yes", "on"}
if ironbank_report.exists():
try:
ironbank_payload = json.loads(ironbank_report.read_text(encoding="utf-8"))
@ -85,11 +86,15 @@ def _load_quality_report(path: Path) -> tuple[float, int, dict[str, str]]:
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"
)
normalized = status.strip().lower()
if normalized in {"ok", "pass", "passed", "success", "compliant"}:
checks["supply_chain"] = "ok"
elif normalized in {"n/a", "na", "not_applicable", "not-applicable", "skipped", "skip"}:
checks["supply_chain"] = "failed" if ironbank_required else "not_applicable"
else:
checks["supply_chain"] = "failed" if ironbank_required else "not_applicable"
except Exception:
checks["supply_chain"] = "failed"
checks["supply_chain"] = "failed" if ironbank_required else "not_applicable"
return float(coverage), int(source_lines), checks

55
testing/ci/summary.py
View File

@ -3,7 +3,6 @@ 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
@ -89,17 +88,35 @@ def read_pushgateway_counters(text: str, *, suite: str, job: str) -> dict[str, f
"""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))
for line in text.splitlines():
if not line.startswith("platform_quality_gate_runs_total{"):
continue
if f'job="{job}"' not in line or f'suite="{suite}"' not in line:
continue
parts = line.split()
if len(parts) < 2:
continue
for status in counters:
if f'status="{status}"' not in line:
continue
try:
counters[status] = float(parts[1])
except ValueError:
counters[status] = 0.0
return counters
def pushgateway_series_exists(text: str, *, metric: str, labels: dict[str, str]) -> bool:
"""Return whether a labeled series already exists in Pushgateway text."""
for line in text.splitlines():
if not line.startswith(metric + "{"):
continue
if all(f'{key}="{value}"' in line for key, value in labels.items()):
return True
return False
def render_payload(
*,
suite: str,
@ -178,10 +195,22 @@ def publish_quality_metrics(
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
already_recorded = bool(build_number) and pushgateway_series_exists(
text,
metric="platform_quality_gate_build_info",
labels={
"job": job,
"suite": suite,
"branch": branch or "unknown",
"build_number": build_number or "unknown",
"jenkins_job": jenkins_job or suite,
},
)
if not already_recorded:
if status == "ok":
counters["ok"] += 1
else:
counters["failed"] += 1
payload = render_payload(
suite=suite,

32
testing/tests/test_publish_metrics.py
View File

@ -2,7 +2,14 @@ from __future__ import annotations
from pathlib import Path
from testing.ci.summary import RunSummary, load_junit_cases, load_junit_summary, render_payload
from testing.ci.summary import (
RunSummary,
load_junit_cases,
load_junit_summary,
pushgateway_series_exists,
read_pushgateway_counters,
render_payload,
)
def test_load_junit_summary_combines_suites(tmp_path: Path) -> None:
@ -46,3 +53,26 @@ def test_load_junit_cases_and_render_test_case_metrics(tmp_path: Path) -> None:
)
assert 'platform_quality_gate_test_case_result{suite="bstein_home"' in payload
assert 'test="app.health::test_fail",status="failed"} 1' in payload
def test_pushgateway_counter_parser_is_label_order_insensitive() -> None:
text = "\n".join(
[
'platform_quality_gate_runs_total{suite="bstein_home",status="ok",job="platform-quality-ci"} 10',
'platform_quality_gate_runs_total{status="failed",job="platform-quality-ci",suite="bstein_home"} 2',
'platform_quality_gate_build_info{suite="bstein_home",branch="master",build_number="274",jenkins_job="bstein-dev-home",job="platform-quality-ci"} 1',
]
)
assert read_pushgateway_counters(text, suite="bstein_home", job="platform-quality-ci") == {"ok": 10.0, "failed": 2.0}
assert pushgateway_series_exists(
text,
metric="platform_quality_gate_build_info",
labels={
"job": "platform-quality-ci",
"suite": "bstein_home",
"branch": "master",
"build_number": "274",
"jenkins_job": "bstein-dev-home",
},
)