Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
36 changes: 36 additions & 0 deletions scripts/runtime_workflow_heartbeat.py
Original file line number Diff line number Diff line change
Expand Up @@ -301,6 +301,19 @@ def _list_runtime_runs(
return [run for run in runs if run.get("display_title") == "Runtime · strategy"]



def _get_workflow_state(*, repository: str, workflow: str, token: str) -> str | None:
"""Return GitHub Actions workflow state (active/disabled_manually/...), or None if unknown."""
url = f"https://api.github.com/repos/{repository}/actions/workflows/{workflow}"
try:
payload = _github_request(url, token)
except Exception as exc: # noqa: BLE001
print(f"Workflow state lookup skipped: {exc}", file=sys.stderr)
return None
state = payload.get("state")
return state if isinstance(state, str) and state else None


def _run_summary(run: dict[str, Any] | None) -> dict[str, Any] | None:
if run is None:
return None
Expand Down Expand Up @@ -470,6 +483,29 @@ def main() -> int:
if not token:
raise SystemExit("GITHUB_TOKEN is required")

workflow_state = _get_workflow_state(repository=repository, workflow=workflow, token=token)
if workflow_state in {"disabled_manually", "disabled_inactivity"}:
assessment = {
"schema": _ASSESSMENT_SCHEMA,
"observed_at": dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z"),
"status": "not_applicable",
"reason": "runtime_workflow_disabled",
"query": {
"workflow": workflow,
"branch": branch,
"workflow_state": workflow_state,
"runs_returned": 0,
},
}
_write_assessment(assessment)
print(json.dumps(assessment, sort_keys=True))
print(
"Runtime workflow heartbeat skipped: "
f"workflow={workflow} state={workflow_state} "
"(no strategy runs expected while disabled)"
)
return 0

now = dt.datetime.now(dt.timezone.utc)
since = now - dt.timedelta(hours=max(lookback_hours, expected_interval_hours * (max_consecutive_misses + 1)))
try:
Expand Down
24 changes: 24 additions & 0 deletions tests/test_runtime_workflow_heartbeat.py
Original file line number Diff line number Diff line change
Expand Up @@ -188,6 +188,28 @@ def test_disabled_target_skips_github_api_lookup_and_writes_assessment(self) ->

request.assert_not_called()

def test_disabled_runtime_workflow_skips_alert_without_failing(self) -> None:
with patch.dict(
os.environ,
{
"GITHUB_TOKEN": "token-1",
"RUNTIME_TARGET_ENABLED": "true",
"RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT": "true",
},
clear=True,
):
with patch.object(
heartbeat,
"_get_workflow_state",
return_value="disabled_manually",
) as state:
with patch.object(heartbeat, "_list_runtime_runs") as list_runs:
with patch.object(heartbeat, "_send_telegram") as send_telegram:
self.assertEqual(heartbeat.main(), 0)
state.assert_called_once()
list_runs.assert_not_called()
send_telegram.assert_not_called()

def test_github_api_outage_is_reported_as_unavailable_when_non_failing(self) -> None:
with patch.dict(
os.environ,
Expand Down Expand Up @@ -352,6 +374,8 @@ def test_repository_runs_fallback_finds_recent_runtime_success(self) -> None:
def fake_github_request(url: str, token: str) -> dict[str, object]:
requested_urls.append(url)
self.assertEqual(token, "token-1")
if url.endswith("/actions/workflows/main.yml"):
return {"state": "active", "path": ".github/workflows/main.yml"}
if "/actions/workflows/main.yml/runs?" in url:
return {"workflow_runs": []}
if "/actions/runs?" in url:
Expand Down
Loading