diff --git a/.github/scripts/pr-health-labels.py b/.github/scripts/pr-health-labels.py new file mode 100644 index 000000000..dfa9eaacc --- /dev/null +++ b/.github/scripts/pr-health-labels.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""Helpers for PR health maintenance labels.""" + +from __future__ import annotations + +import argparse +import json +import sys +from datetime import datetime, timezone +from typing import Any + +FAILING_STATES = {"FAILURE", "TIMED_OUT", "ACTION_REQUIRED", "CANCELLED", "ERROR"} + + +def _parse_timestamp(value: Any) -> datetime: + if not isinstance(value, str) or not value: + return datetime.min.replace(tzinfo=timezone.utc) + normalized = value.removesuffix("Z") + "+00:00" if value.endswith("Z") else value + try: + parsed = datetime.fromisoformat(normalized) + except ValueError: + return datetime.min.replace(tzinfo=timezone.utc) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed + + +def _check_key(check: dict[str, Any]) -> tuple[str, str]: + workflow = str(check.get("workflowName") or check.get("workflow") or "") + name = str(check.get("name") or check.get("context") or "") + return workflow, name + + +def _check_time(check: dict[str, Any]) -> datetime: + return max( + _parse_timestamp(check.get("startedAt")), + _parse_timestamp(check.get("completedAt")), + ) + + +def _state(check: dict[str, Any]) -> str: + return str(check.get("conclusion") or check.get("state") or "").upper() + + +def current_checks(payload: dict[str, Any]) -> list[dict[str, Any]]: + latest_by_key: dict[tuple[str, str], dict[str, Any]] = {} + for check in payload.get("statusCheckRollup") or []: + if not isinstance(check, dict): + continue + key = _check_key(check) + if not any(key): + continue + previous = latest_by_key.get(key) + if previous is None or _check_time(check) >= _check_time(previous): + latest_by_key[key] = check + return list(latest_by_key.values()) + + +def check_state(payload: dict[str, Any]) -> str: + for check in current_checks(payload): + if _state(check) in FAILING_STATES: + return "failing" + return "passing" + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--state-json", required=True, help="JSON from gh pr view") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv or sys.argv[1:]) + print(check_state(json.loads(args.state_json))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/.github/workflows/pr-health.yml b/.github/workflows/pr-health.yml index abfcae1c2..1b266e05c 100644 --- a/.github/workflows/pr-health.yml +++ b/.github/workflows/pr-health.yml @@ -189,14 +189,7 @@ jobs: --json isDraft,labels,mergeStateStatus,statusCheckRollup)" merge_state="$(jq -r '.mergeStateStatus // "UNKNOWN"' <<<"$data")" - check_state="$(jq -r ' - [ - (.statusCheckRollup // [])[] - | select((.conclusion // .state // "") as $s - | ["FAILURE", "TIMED_OUT", "ACTION_REQUIRED", "CANCELLED", "ERROR"] | index($s)) - ] - | if length > 0 then "failing" else "passing" end - ' <<<"$data")" + check_state="$(python3 .github/scripts/pr-health-labels.py --state-json "$data")" is_draft="$(jq -r '.isDraft' <<<"$data")" if [[ "$merge_state" == "BEHIND" ]]; then diff --git a/scripts/tests/test_pr_health_labels.py b/scripts/tests/test_pr_health_labels.py new file mode 100644 index 000000000..a7bb5ef18 --- /dev/null +++ b/scripts/tests/test_pr_health_labels.py @@ -0,0 +1,94 @@ +"""Tests for pr-health-labels.py.""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + + +def _load_module(): + script = Path(__file__).parents[2] / ".github" / "scripts" / "pr-health-labels.py" + spec = importlib.util.spec_from_file_location("pr_health_labels", script) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def test_check_state_ignores_historical_failures_when_latest_attempt_passed() -> None: + module = _load_module() + payload = { + "statusCheckRollup": [ + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "template", + "conclusion": "FAILURE", + "startedAt": "2026-06-14T13:38:35Z", + "completedAt": "2026-06-14T13:38:46Z", + }, + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "template", + "conclusion": "SUCCESS", + "startedAt": "2026-06-14T13:43:26Z", + "completedAt": "2026-06-14T13:43:35Z", + }, + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "label", + "conclusion": "CANCELLED", + "startedAt": "2026-06-14T13:43:18Z", + "completedAt": "2026-06-14T13:43:24Z", + }, + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "label", + "conclusion": "SUCCESS", + "startedAt": "2026-06-14T13:43:26Z", + "completedAt": "2026-06-14T13:43:35Z", + }, + { + "__typename": "CheckRun", + "workflowName": "", + "name": "GitGuardian Security Checks", + "conclusion": "SUCCESS", + "startedAt": "2026-06-14T13:38:32Z", + "completedAt": "2026-06-14T13:39:04Z", + }, + ] + } + + assert module.check_state(payload) == "passing" + + +def test_check_state_fails_when_latest_attempt_failed() -> None: + module = _load_module() + payload = { + "statusCheckRollup": [ + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "template", + "conclusion": "SUCCESS", + "startedAt": "2026-06-14T13:38:35Z", + "completedAt": "2026-06-14T13:38:46Z", + }, + { + "__typename": "CheckRun", + "workflowName": "PR Governance", + "name": "template", + "conclusion": "FAILURE", + "startedAt": "2026-06-14T13:43:26Z", + "completedAt": "2026-06-14T13:43:35Z", + }, + ] + } + + assert module.check_state(payload) == "failing"