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
7 changes: 6 additions & 1 deletion .github/workflows/execution-report-heartbeat.yml
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,11 @@ on:
required: false
type: string
default: "36"
notify_on_success:
description: "Opt in to normal healthy-check notifications for this manual run."
required: false
type: boolean
default: false
fail_workflow_on_alert:
description: "Fail this workflow when an alert is emitted."
required: false
Expand Down Expand Up @@ -44,7 +49,7 @@ jobs:
RUNTIME_HEARTBEAT_ACCEPT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_ACCEPT_STAGES }}
RUNTIME_HEARTBEAT_REJECT_STAGES: ${{ vars.RUNTIME_HEARTBEAT_REJECT_STAGES }}
RUNTIME_HEARTBEAT_MARKET_AWARE: ${{ vars.RUNTIME_HEARTBEAT_MARKET_AWARE || 'true' }}
RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS: ${{ github.event_name != 'schedule' && vars.RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS || 'false' }}
RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS: ${{ github.event_name == 'workflow_dispatch' && inputs.notify_on_success && 'true' || 'false' }}
RUNTIME_HEARTBEAT_MARKET_CALENDAR: ${{ vars.FIRSTRADE_MARKET_CALENDAR }}
RUNTIME_HEARTBEAT_MARKET_TIMEZONE: ${{ vars.FIRSTRADE_MARKET_TIMEZONE }}
RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES: ${{ vars.RUNTIME_HEARTBEAT_PUBLICATION_GRACE_MINUTES || '30' }}
Expand Down
55 changes: 55 additions & 0 deletions tests/test_runtime_monitor_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -140,3 +140,58 @@ def test_lifecycle_publishes_read_only_observation_for_exact_service() -> None:
assert "CLOUD_RUN_SERVICES" not in publisher
assert "gcloud scheduler jobs update" not in workflow
assert "gcloud run deploy" not in workflow


def _manual_input_block(workflow: str, name: str) -> str:
import re
match = re.search(rf"(?ms)^ {name}:\n(.*?)(?=^ [a-z_]+:|^ [a-z_]+:)", workflow)
assert match is not None, f"missing workflow_dispatch input {name}"
return match.group(1)


def _evaluate_success_notify_expression(workflow: str, event: str, explicit_input, repository_value) -> str:
"""Evaluate this bounded Actions boolean/string expression without running a workflow."""
import ast
import re
expression = re.search(r"RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS: \$\{\{ (.*?) \}\}", workflow).group(1)
expression = expression.replace("github.event_name", repr(event))
expression = expression.replace("inputs.notify_on_success", repr(False if explicit_input is None else explicit_input))
expression = expression.replace("vars.RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS", repr(repository_value or ""))
expression = expression.replace("&&", " and ").replace("||", " or ")
parsed = ast.parse(expression, mode="eval")
assert all(isinstance(node, (ast.Expression, ast.BoolOp, ast.And, ast.Or, ast.Compare, ast.Eq, ast.NotEq, ast.Constant)) for node in ast.walk(parsed))
result = eval(compile(parsed, "<offline Actions expression>", "eval"), {"__builtins__": {}}, {})
return str(result).lower() if isinstance(result, bool) else str(result)


def test_manual_healthy_notify_is_typed_and_explicitly_opt_in() -> None:
workflow = (ROOT / ".github/workflows/execution-report-heartbeat.yml").read_text()
block = _manual_input_block(workflow, "notify_on_success")
assert "type: boolean" in block
assert "default: false" in block
assert "required: false" in block
line = next(line for line in workflow.splitlines() if "RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS:" in line)
assert "inputs.notify_on_success" in line
assert "github.event.inputs" not in line
assert "vars.RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS" not in line
assert "github.event_name == 'workflow_dispatch'" in line


def test_schedule_and_manual_default_never_inherit_legacy_success_variable() -> None:
workflow = (ROOT / ".github/workflows/execution-report-heartbeat.yml").read_text()
for event in ("schedule", "workflow_dispatch", "repository_dispatch"):
for explicit_input in (None, False, True):
for repository_value in (None, "false", "true"):
actual = _evaluate_success_notify_expression(workflow, event, explicit_input, repository_value)
expected = "true" if event == "workflow_dispatch" and explicit_input is True else "false"
assert actual == expected, (event, explicit_input, repository_value, actual)


def test_manual_quiet_control_preserves_existing_alert_and_report_steps() -> None:
workflow = (ROOT / ".github/workflows/execution-report-heartbeat.yml").read_text()
alert_block = _manual_input_block(workflow, "fail_workflow_on_alert")
assert 'default: "true"' in alert_block
assert "RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT: ${{ inputs.fail_workflow_on_alert || vars.RUNTIME_HEARTBEAT_FAIL_WORKFLOW_ON_ALERT || 'true' }}" in workflow
assert "uv run --no-sync python scripts/execution_report_heartbeat.py" in workflow
assert "Publish read-only runtime execution evidence" in workflow
assert 'cron: "20 22 * * *"' in workflow
Loading