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

ci(ananke): publish per-test case metrics for flaky tracking

Browse Source
This commit is contained in:
codex 2026-04-20 08:35:15 -03:00
parent b2a7d72305
commit b1f3dd1d4e
1 changed files with 30 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
scripts/publish_quality_metrics.py
View File

@ -87,6 +87,7 @@ def _build_payload(
tests_failed: int,
tests_errors: int,
tests_skipped: int,
test_cases: list[tuple[str, str]],
coverage_percent: float,
source_lines_over_500: int,
checks: dict[str, str],
@ -114,6 +115,12 @@ def _build_payload(
f'ananke_quality_gate_checks_total{{suite="{suite}",check="{check_name}",result="{check_status}"}} 1'
for check_name, check_status in checks.items()
)
if test_cases:
lines.append("# TYPE platform_quality_gate_test_case_result gauge")
lines.extend(
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
)
return "\n".join(lines) + "\n"
@ -159,6 +166,25 @@ def _parse_go_test_counts(output_path: Path) -> dict[str, int]:
}
def _parse_go_test_cases(output_path: Path) -> list[tuple[str, str]]:
"""Parse per-test status records from go test output text."""
if not output_path.exists():
return []
text = output_path.read_text(encoding="utf-8", errors="ignore")
cases: list[tuple[str, str]] = []
patterns = {
"passed": re.compile(r"^--- PASS: ([^\s(]+)", flags=re.M),
"failed": re.compile(r"^--- FAIL: ([^\s(]+)", flags=re.M),
"skipped": re.compile(r"^--- SKIP: ([^\s(]+)", flags=re.M),
}
for status, pattern in patterns.items():
for test_name in pattern.findall(text):
cleaned = str(test_name).strip()
if cleaned:
cases.append((cleaned, status))
return cases
def _read_exit_code(path: Path) -> int:
if not path.exists():
return 1
@ -288,7 +314,9 @@ def main(argv: list[str] | None = None) -> int:
resolved_failed = max(args.local_failed, remote_failed)
coverage_percent = _read_coverage_percent(args.coverage_percent_file)
source_lines_over_500 = _count_source_files_over_limit(repo_root, max_lines=500)
tests = _parse_go_test_counts(Path(os.getenv("ANANKE_QUALITY_OUTPUT_FILE", str(build_dir / "quality-gate.out"))))
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"))))
gate_failed = gate_rc != 0
@ -310,6 +338,7 @@ def main(argv: list[str] | None = None) -> int:
tests_failed=tests["failed"],
tests_errors=tests["errors"],
tests_skipped=tests["skipped"],
test_cases=test_cases,
coverage_percent=coverage_percent,
source_lines_over_500=source_lines_over_500,
checks=checks,