3ed582a22c
Before the orchestrator builds the agenda and burns any cycles, it now runs the Quick Health Snapshot (#710) to check CI, open P0/P1 issues, test flakiness, and token economy. If the overall status is red the Daily Run aborts with a non-zero exit code so the loop doesn't waste cycles when mandatory systems are down. New CLI flags on orchestrator.py: - --skip-health-check bypass the pre-flight snapshot entirely - --force continue even when health status is red Adds 13 unit tests in tests/timmy_automations/test_orchestrator.py covering green/yellow/red paths, --force override, --skip-health-check, graceful degradation on snapshot errors, and JSON output. Fixes #923 Refs #710
271 lines
9.3 KiB
Python
271 lines
9.3 KiB
Python
"""Tests for Daily Run orchestrator — health snapshot integration.
|
|
|
|
Verifies that the orchestrator runs a pre-flight health snapshot before
|
|
any coding work begins, and aborts on red status unless --force is passed.
|
|
|
|
Refs: #923
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# Add timmy_automations to path for imports
|
|
_TA_PATH = Path(__file__).resolve().parent.parent.parent / "timmy_automations" / "daily_run"
|
|
if str(_TA_PATH) not in sys.path:
|
|
sys.path.insert(0, str(_TA_PATH))
|
|
# Also add utils path
|
|
_TA_UTILS = Path(__file__).resolve().parent.parent.parent / "timmy_automations"
|
|
if str(_TA_UTILS) not in sys.path:
|
|
sys.path.insert(0, str(_TA_UTILS))
|
|
|
|
import health_snapshot as hs
|
|
import orchestrator as orch
|
|
|
|
|
|
def _make_snapshot(overall_status: str) -> hs.HealthSnapshot:
|
|
"""Build a minimal HealthSnapshot for testing."""
|
|
return hs.HealthSnapshot(
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
overall_status=overall_status,
|
|
ci=hs.CISignal(status="pass", message="CI passing"),
|
|
issues=hs.IssueSignal(count=0, p0_count=0, p1_count=0),
|
|
flakiness=hs.FlakinessSignal(
|
|
status="healthy",
|
|
recent_failures=0,
|
|
recent_cycles=10,
|
|
failure_rate=0.0,
|
|
message="All good",
|
|
),
|
|
tokens=hs.TokenEconomySignal(status="balanced", message="Balanced"),
|
|
)
|
|
|
|
|
|
def _make_red_snapshot() -> hs.HealthSnapshot:
|
|
return hs.HealthSnapshot(
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
overall_status="red",
|
|
ci=hs.CISignal(status="fail", message="CI failed"),
|
|
issues=hs.IssueSignal(count=1, p0_count=1, p1_count=0),
|
|
flakiness=hs.FlakinessSignal(
|
|
status="critical",
|
|
recent_failures=8,
|
|
recent_cycles=10,
|
|
failure_rate=0.8,
|
|
message="High flakiness",
|
|
),
|
|
tokens=hs.TokenEconomySignal(status="unknown", message="No data"),
|
|
)
|
|
|
|
|
|
def _default_args(**overrides) -> argparse.Namespace:
|
|
"""Build an argparse Namespace with defaults matching the orchestrator flags."""
|
|
defaults = {
|
|
"review": False,
|
|
"json": False,
|
|
"max_items": None,
|
|
"skip_health_check": False,
|
|
"force": False,
|
|
}
|
|
defaults.update(overrides)
|
|
return argparse.Namespace(**defaults)
|
|
|
|
|
|
class TestRunHealthSnapshot:
|
|
"""Test run_health_snapshot() — the pre-flight check called by main()."""
|
|
|
|
def test_green_returns_zero(self, capsys):
|
|
"""Green snapshot returns 0 (proceed)."""
|
|
args = _default_args()
|
|
|
|
with patch.object(orch, "_generate_health_snapshot", return_value=_make_snapshot("green")):
|
|
rc = orch.run_health_snapshot(args)
|
|
|
|
assert rc == 0
|
|
|
|
def test_yellow_returns_zero(self, capsys):
|
|
"""Yellow snapshot returns 0 (proceed with caution)."""
|
|
args = _default_args()
|
|
|
|
with patch.object(orch, "_generate_health_snapshot", return_value=_make_snapshot("yellow")):
|
|
rc = orch.run_health_snapshot(args)
|
|
|
|
assert rc == 0
|
|
|
|
def test_red_returns_one(self, capsys):
|
|
"""Red snapshot returns 1 (abort)."""
|
|
args = _default_args()
|
|
|
|
with patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()):
|
|
rc = orch.run_health_snapshot(args)
|
|
|
|
assert rc == 1
|
|
|
|
def test_red_with_force_returns_zero(self, capsys):
|
|
"""Red snapshot with --force returns 0 (proceed anyway)."""
|
|
args = _default_args(force=True)
|
|
|
|
with patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()):
|
|
rc = orch.run_health_snapshot(args)
|
|
|
|
assert rc == 0
|
|
|
|
def test_snapshot_exception_is_skipped(self, capsys):
|
|
"""If health snapshot raises, it degrades gracefully and returns 0."""
|
|
args = _default_args()
|
|
|
|
with patch.object(orch, "_generate_health_snapshot", side_effect=RuntimeError("boom")):
|
|
rc = orch.run_health_snapshot(args)
|
|
|
|
assert rc == 0
|
|
captured = capsys.readouterr()
|
|
assert "warning" in captured.err.lower() or "skipping" in captured.err.lower()
|
|
|
|
def test_snapshot_prints_summary(self, capsys):
|
|
"""Health snapshot prints a pre-flight summary block."""
|
|
args = _default_args()
|
|
|
|
with patch.object(orch, "_generate_health_snapshot", return_value=_make_snapshot("green")):
|
|
orch.run_health_snapshot(args)
|
|
|
|
captured = capsys.readouterr()
|
|
assert "PRE-FLIGHT HEALTH CHECK" in captured.out
|
|
assert "CI" in captured.out
|
|
|
|
def test_red_prints_abort_message(self, capsys):
|
|
"""Red snapshot prints an abort message to stderr."""
|
|
args = _default_args()
|
|
|
|
with patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()):
|
|
orch.run_health_snapshot(args)
|
|
|
|
captured = capsys.readouterr()
|
|
assert "RED" in captured.err or "aborting" in captured.err.lower()
|
|
|
|
def test_p0_issues_shown_in_output(self, capsys):
|
|
"""P0 issue count is shown in the pre-flight output."""
|
|
args = _default_args()
|
|
snapshot = hs.HealthSnapshot(
|
|
timestamp="2026-01-01T00:00:00+00:00",
|
|
overall_status="red",
|
|
ci=hs.CISignal(status="pass", message="CI passing"),
|
|
issues=hs.IssueSignal(count=2, p0_count=2, p1_count=0),
|
|
flakiness=hs.FlakinessSignal(
|
|
status="healthy",
|
|
recent_failures=0,
|
|
recent_cycles=10,
|
|
failure_rate=0.0,
|
|
message="All good",
|
|
),
|
|
tokens=hs.TokenEconomySignal(status="balanced", message="Balanced"),
|
|
)
|
|
|
|
with patch.object(orch, "_generate_health_snapshot", return_value=snapshot):
|
|
orch.run_health_snapshot(args)
|
|
|
|
captured = capsys.readouterr()
|
|
assert "P0" in captured.out
|
|
|
|
|
|
class TestMainHealthCheckIntegration:
|
|
"""Test that main() runs health snapshot before any coding work."""
|
|
|
|
def _patch_gitea_unavailable(self):
|
|
return patch.object(orch.GiteaClient, "is_available", return_value=False)
|
|
|
|
def test_main_runs_health_check_before_gitea(self):
|
|
"""Health snapshot is called before Gitea client work."""
|
|
call_order = []
|
|
|
|
def fake_snapshot(*_a, **_kw):
|
|
call_order.append("health")
|
|
return _make_snapshot("green")
|
|
|
|
def fake_gitea_available(self):
|
|
call_order.append("gitea")
|
|
return False
|
|
|
|
args = _default_args()
|
|
|
|
with (
|
|
patch.object(orch, "_generate_health_snapshot", side_effect=fake_snapshot),
|
|
patch.object(orch.GiteaClient, "is_available", fake_gitea_available),
|
|
patch("sys.argv", ["orchestrator"]),
|
|
):
|
|
orch.main()
|
|
|
|
assert call_order.index("health") < call_order.index("gitea")
|
|
|
|
def test_main_aborts_on_red_before_gitea(self):
|
|
"""main() aborts with non-zero exit code when health is red."""
|
|
gitea_called = []
|
|
|
|
def fake_gitea_available(self):
|
|
gitea_called.append(True)
|
|
return True
|
|
|
|
with (
|
|
patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()),
|
|
patch.object(orch.GiteaClient, "is_available", fake_gitea_available),
|
|
patch("sys.argv", ["orchestrator"]),
|
|
):
|
|
rc = orch.main()
|
|
|
|
assert rc != 0
|
|
assert not gitea_called, "Gitea should NOT be called when health is red"
|
|
|
|
def test_main_skips_health_check_with_flag(self):
|
|
"""--skip-health-check bypasses the pre-flight snapshot."""
|
|
health_called = []
|
|
|
|
def fake_snapshot(*_a, **_kw):
|
|
health_called.append(True)
|
|
return _make_snapshot("green")
|
|
|
|
with (
|
|
patch.object(orch, "_generate_health_snapshot", side_effect=fake_snapshot),
|
|
patch.object(orch.GiteaClient, "is_available", return_value=False),
|
|
patch("sys.argv", ["orchestrator", "--skip-health-check"]),
|
|
):
|
|
orch.main()
|
|
|
|
assert not health_called, "Health snapshot should be skipped"
|
|
|
|
def test_main_force_flag_continues_despite_red(self):
|
|
"""--force allows Daily Run to continue even when health is red."""
|
|
gitea_called = []
|
|
|
|
def fake_gitea_available(self):
|
|
gitea_called.append(True)
|
|
return False # Gitea unavailable → exits early but after health check
|
|
|
|
with (
|
|
patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()),
|
|
patch.object(orch.GiteaClient, "is_available", fake_gitea_available),
|
|
patch("sys.argv", ["orchestrator", "--force"]),
|
|
):
|
|
orch.main()
|
|
|
|
# Gitea was reached despite red status because --force was passed
|
|
assert gitea_called
|
|
|
|
def test_main_json_output_on_red_includes_error(self, capsys):
|
|
"""JSON output includes error key when health is red."""
|
|
with (
|
|
patch.object(orch, "_generate_health_snapshot", return_value=_make_red_snapshot()),
|
|
patch.object(orch.GiteaClient, "is_available", return_value=True),
|
|
patch("sys.argv", ["orchestrator", "--json"]),
|
|
):
|
|
rc = orch.main()
|
|
|
|
assert rc != 0
|
|
captured = capsys.readouterr()
|
|
data = json.loads(captured.out)
|
|
assert "error" in data
|