bstein/ananke
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ci: count quality runs once per build

Browse Source
This commit is contained in:
codex 2026-05-11 13:22:22 -03:00
parent 42e6a244b5
commit 2fe14f69d4
2 changed files with 138 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
scripts/publish_quality_metrics.py
View File

@ -77,6 +77,17 @@ def _fetch_existing_counter(pushgateway_url: str, metric: str, labels: dict[str,
return 0.0
def _series_exists(pushgateway_url: str, metric: str, labels: dict[str, str], timeout_seconds: float) -> bool:
"""Return whether Pushgateway already has a series for this build."""
text = _read_http(f"{pushgateway_url.rstrip('/')}/metrics", timeout_seconds)
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 _build_payload(
suite: str,
trigger: str,
@ -284,17 +295,23 @@ def _sonarqube_check_status(build_dir: Path) -> str:
def _supply_chain_check_status(build_dir: Path) -> str:
required = os.getenv("QUALITY_GATE_IRONBANK_REQUIRED", "0").strip().lower() in {"1", "true", "yes", "on"}
report = _load_json(Path(os.getenv("QUALITY_GATE_IRONBANK_REPORT", str(build_dir / "ironbank-compliance.json"))))
if not report:
return "not_applicable"
return "failed" if required else "not_applicable"
compliant = report.get("compliant")
if isinstance(compliant, bool):
return "ok" if compliant else "failed"
status_candidates = [report.get("status"), report.get("result"), report.get("compliance")]
for value in status_candidates:
if isinstance(value, str):
return "ok" if value.strip().lower() in QUALITY_SUCCESS_STATES else "failed"
return "failed"
normalized = value.strip().lower()
if normalized in QUALITY_SUCCESS_STATES:
return "ok"
if normalized in {"n/a", "na", "not_applicable", "not-applicable", "skipped", "skip"}:
return "failed" if required else "not_applicable"
return "failed" if required else "not_applicable"
return "failed" if required else "not_applicable"
def parse_args(argv: list[str]) -> argparse.Namespace:
@ -338,10 +355,19 @@ def main(argv: list[str] | None = None) -> int:
args = parse_args(argv or sys.argv[1:])
repo_root = Path(__file__).resolve().parents[1]
build_dir = repo_root / "build"
gate_rc = _read_exit_code(Path(os.getenv("ANANKE_QUALITY_EXIT_CODE_PATH", str(build_dir / "quality-gate.rc"))))
current_ok = 1 if gate_rc == 0 else 0
current_failed = 0 if gate_rc == 0 else 1
branch = os.getenv("BRANCH_NAME") or os.getenv("GIT_BRANCH") or "unknown"
if branch.startswith("origin/"):
branch = branch[len("origin/") :]
build_number = os.getenv("BUILD_NUMBER", "")
jenkins_job = os.getenv("JOB_NAME", "ananke")
remote_ok = 0
remote_failed = 0
remote_error = ""
already_recorded = False
try:
remote_ok = int(
_fetch_existing_counter(
@ -359,22 +385,34 @@ def main(argv: list[str] | None = None) -> int:
args.timeout_seconds,
)
)
already_recorded = bool(build_number) and _series_exists(
args.pushgateway_url,
"platform_quality_gate_build_info",
{
"job": args.job_name,
"suite": args.suite,
"branch": branch or "unknown",
"build_number": build_number or "unknown",
"jenkins_job": jenkins_job,
},
args.timeout_seconds,
)
except Exception as exc:
remote_error = str(exc)
resolved_ok = max(args.local_ok, remote_ok)
resolved_failed = max(args.local_failed, remote_failed)
resolved_ok = remote_ok
resolved_failed = remote_failed
if remote_error:
resolved_ok = args.local_ok
resolved_failed = args.local_failed
elif not already_recorded:
resolved_ok += current_ok
resolved_failed += current_failed
coverage_percent = _read_coverage_percent(args.coverage_percent_file)
source_lines_over_500 = _count_source_files_over_limit(repo_root, max_lines=500)
branch = os.getenv("BRANCH_NAME") or os.getenv("GIT_BRANCH") or "unknown"
if branch.startswith("origin/"):
branch = branch[len("origin/") :]
build_number = os.getenv("BUILD_NUMBER", "")
jenkins_job = os.getenv("JOB_NAME", "ananke")
quality_output = Path(os.getenv("ANANKE_QUALITY_OUTPUT_FILE", str(build_dir / "quality-gate.out")))
tests = _parse_go_test_counts(quality_output)
test_cases = _parse_go_test_cases(quality_output)
gate_rc = _read_exit_code(Path(os.getenv("ANANKE_QUALITY_EXIT_CODE_PATH", str(build_dir / "quality-gate.rc"))))
docs_status = _read_status(Path(os.getenv("ANANKE_QUALITY_DOCS_STATUS_PATH", str(build_dir / "docs-naming.status"))))
unit_tests_failed = _unit_tests_failed(quality_output, coverage_percent)
checks = {

122
scripts/publish_quality_metrics_test.py
View File

@ -3,8 +3,11 @@
from __future__ import annotations
import http.server
from pathlib import Path
import socketserver
import tempfile
import threading
from unittest import mock
import unittest
import publish_quality_metrics as publisher
@ -58,7 +61,19 @@ class PublishQualityMetricsTest(unittest.TestCase):
self.server.server_close()
self.thread.join(timeout=5)
def test_publish_uses_remote_high_water_mark(self) -> None:
def _env_for_gate_status(self, status: int = 0) -> dict[str, str]:
tmp_dir = tempfile.TemporaryDirectory()
self.addCleanup(tmp_dir.cleanup)
rc_path = Path(tmp_dir.name) / "quality-gate.rc"
rc_path.write_text(f"{status}\n", encoding="utf-8")
return {
"ANANKE_QUALITY_EXIT_CODE_PATH": str(rc_path),
"ANANKE_QUALITY_COVERAGE_PERCENT_FILE": str(Path(tmp_dir.name) / "coverage.txt"),
"ANANKE_QUALITY_OUTPUT_FILE": str(Path(tmp_dir.name) / "quality-gate.out"),
"ANANKE_QUALITY_DOCS_STATUS_PATH": str(Path(tmp_dir.name) / "docs-naming.status"),
}
def test_publish_adds_current_run_to_remote_counters(self) -> None:
_GatewayHandler.metrics_text = "\n".join(
[
'# TYPE platform_quality_gate_runs_total counter',
@ -67,51 +82,92 @@ class PublishQualityMetricsTest(unittest.TestCase):
]
)
exit_code = publisher.main(
[
"--pushgateway-url",
self.base_url,
"--job-name",
"platform-quality-ci",
"--suite",
"ananke",
"--trigger",
"host",
"--local-ok",
"5",
"--local-failed",
"2",
]
)
with mock.patch.dict("os.environ", self._env_for_gate_status(0)):
exit_code = publisher.main(
[
"--pushgateway-url",
self.base_url,
"--job-name",
"platform-quality-ci",
"--suite",
"ananke",
"--trigger",
"host",
"--local-ok",
"5",
"--local-failed",
"2",
]
)
self.assertEqual(exit_code, 0)
self.assertEqual(len(_GatewayHandler.posts), 1)
path, body = _GatewayHandler.posts[0]
self.assertEqual(path, "/metrics/job/platform-quality-ci/suite/ananke")
self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="ok"} 7', body)
self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="failed"} 2', body)
self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="ok"} 8', body)
self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="failed"} 1', body)
self.assertIn('ananke_quality_gate_publish_info{suite="ananke",trigger="host"} 1', body)
self.assertIn('ananke_quality_gate_coverage_percent{suite="ananke"}', body)
self.assertIn('platform_quality_gate_workspace_line_coverage_percent{suite="ananke"}', body)
self.assertIn('platform_quality_gate_source_lines_over_500_total{suite="ananke"}', body)
def test_publish_does_not_double_count_same_build(self) -> None:
_GatewayHandler.metrics_text = "\n".join(
[
'platform_quality_gate_runs_total{job="platform-quality-ci",suite="ananke",status="ok"} 7',
'platform_quality_gate_runs_total{job="platform-quality-ci",suite="ananke",status="failed"} 1',
'platform_quality_gate_build_info{job="platform-quality-ci",suite="ananke",branch="main",build_number="78",jenkins_job="ananke"} 1',
]
)
with mock.patch.dict(
"os.environ",
{
**self._env_for_gate_status(0),
"BRANCH_NAME": "main",
"BUILD_NUMBER": "78",
"JOB_NAME": "ananke",
},
):
exit_code = publisher.main(
[
"--pushgateway-url",
self.base_url,
"--job-name",
"platform-quality-ci",
"--suite",
"ananke",
"--trigger",
"host",
"--local-ok",
"1",
"--local-failed",
"0",
]
)
self.assertEqual(exit_code, 0)
_, body = _GatewayHandler.posts[0]
self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="ok"} 7', body)
self.assertIn('platform_quality_gate_runs_total{suite="ananke",status="failed"} 1', body)
def test_publish_falls_back_to_local_counters_when_metrics_read_fails(self) -> None:
_GatewayHandler.fail_metrics_read = True
exit_code = publisher.main(
[
"--pushgateway-url",
self.base_url,
"--job-name",
"platform-quality-ci",
"--suite",
"ananke",
"--local-ok",
"11",
"--local-failed",
"3",
]
)
with mock.patch.dict("os.environ", self._env_for_gate_status(0)):
exit_code = publisher.main(
[
"--pushgateway-url",
self.base_url,
"--job-name",
"platform-quality-ci",
"--suite",
"ananke",
"--local-ok",
"11",
"--local-failed",
"3",
]
)
self.assertEqual(exit_code, 0)
self.assertEqual(len(_GatewayHandler.posts), 1)