diff --git a/backend/tests/test_schedule_e2e.py b/backend/tests/test_schedule_e2e.py new file mode 100644 index 00000000..3723041a --- /dev/null +++ b/backend/tests/test_schedule_e2e.py @@ -0,0 +1,277 @@ +"""End-to-end smoke: does a scheduled workflow actually fire when its +time hits, with the full scheduler loop running? + +Runs the real scheduler.start() loop with the executor mocked so we +don't need a live agent_manager. Then arms a workflow whose +next_run_at is one second in the future, waits, and asserts the +mocked executor was called. + +Run: + cd backend && .venv/bin/python -m pytest tests/test_schedule_e2e.py -v +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock +from zoneinfo import ZoneInfo + +import pytest + +pytestmark = pytest.mark.asyncio + + +@pytest.fixture(autouse=True) +def isolated_data_dir(monkeypatch, tmp_path): + from backend.apps.workflows import storage as _storage + from backend.apps.workflows import escalation as _escalation + from backend.apps.workflows import audit as _audit + from backend.apps.workflows import scheduler as _scheduler + monkeypatch.setattr(_storage, "DATA_DIR", str(tmp_path / "workflows")) + monkeypatch.setattr(_storage, "RUNS_DIR", str(tmp_path / "workflows" / "runs")) + monkeypatch.setattr(_storage, "PAUSED_FILE", str(tmp_path / "workflows" / "paused.json")) + monkeypatch.setattr(_storage, "_workflow_cache", {}) + monkeypatch.setattr(_storage, "_runs_cache", {}) + monkeypatch.setattr(_storage, "_cache_loaded", False) + monkeypatch.setattr(_storage, "_paused", False) + monkeypatch.setattr(_audit, "AUDIT_DIR", str(tmp_path / "workflows" / "audit")) + # Module-level scheduler state survives across tests; reset it so + # each test gets a fresh _wake Event bound to its own event loop. + _scheduler._loop_task = None + _scheduler._wake = asyncio.Event() + _escalation._tasks.clear() + _escalation._state.clear() + yield + + +def _make_wf(**overrides): + from backend.apps.workflows.models import Workflow, ScheduleConfig, WorkflowStep + base = dict( + title="smoke", + steps=[WorkflowStep(text="hi")], + schedule=ScheduleConfig( + enabled=True, repeat_unit="day", repeat_every=1, + hour=9, minute=0, timezone="America/Los_Angeles", + ), + ) + base.update(overrides) + return Workflow(**base) + + +async def test_loop_fires_due_workflow(monkeypatch): + """Arm a workflow to fire ~now and assert the executor was actually + invoked by the scheduler loop within the test window. Note the save + happens AFTER scheduler.start() so reconcile_on_startup doesn't + clobber next_run_at.""" + from backend.apps.workflows import storage, scheduler, executor + + fired = asyncio.Event() + captured: dict = {} + + async def fake_execute(wf, triggered_by="schedule", scheduled_for=None): + captured["wf_id"] = wf.id + captured["triggered_by"] = triggered_by + captured["scheduled_for"] = scheduled_for + from backend.apps.workflows.models import WorkflowRun + run = WorkflowRun( + workflow_id=wf.id, + status="success", + scheduled_for=scheduled_for, + started_at=datetime.now(timezone.utc), + finished_at=datetime.now(timezone.utc), + triggered_by=triggered_by, + ) + storage.record_run(run) + fired.set() + return run + + monkeypatch.setattr(executor, "execute", fake_execute) + + await scheduler.start() + try: + wf = _make_wf() + wf.next_run_at = datetime.now(timezone.utc) + timedelta(seconds=1) + storage.save_workflow(wf) + scheduler.kick() # force immediate tick + # Wait up to 5s for the fire to land. + await asyncio.wait_for(fired.wait(), timeout=5.0) + finally: + await scheduler.stop() + + assert captured.get("wf_id") == wf.id + assert captured.get("triggered_by") == "schedule" + runs = storage.list_runs(wf.id, limit=10) + assert len(runs) == 1 + assert runs[0].status == "success" + # Scheduler should have rolled next_run_at forward to a future slot. + after = storage.get_workflow(wf.id) + assert after.next_run_at is not None + assert after.next_run_at > datetime.now(timezone.utc) + + +async def test_disabled_workflow_does_not_fire(monkeypatch): + """Master switch off => loop never invokes the executor even if + next_run_at is in the past.""" + from backend.apps.workflows import storage, scheduler, executor + + fake = AsyncMock() + monkeypatch.setattr(executor, "execute", fake) + + wf = _make_wf() + wf.schedule.enabled = False + wf.next_run_at = datetime.now(timezone.utc) - timedelta(seconds=10) + storage.save_workflow(wf) + + await scheduler.start() + try: + scheduler.kick() + await asyncio.sleep(2.0) + finally: + await scheduler.stop() + fake.assert_not_called() + + +async def test_paused_state_blocks_all_fires(monkeypatch): + """Global pause flag wins over per-workflow enabled state.""" + from backend.apps.workflows import storage, scheduler, executor + + fake = AsyncMock() + monkeypatch.setattr(executor, "execute", fake) + + wf = _make_wf() + wf.next_run_at = datetime.now(timezone.utc) - timedelta(seconds=1) + storage.save_workflow(wf) + storage.set_paused(True) + + await scheduler.start() + try: + scheduler.kick() + await asyncio.sleep(2.0) + finally: + await scheduler.stop() + fake.assert_not_called() + storage.set_paused(False) + + +async def test_reconcile_skip_rolls_past_missed(monkeypatch): + """on_missed='skip' + a missed next_run_at => startup rolls forward + to the next future fire without queuing a catch-up.""" + from backend.apps.workflows import storage, scheduler + wf = _make_wf() + wf.schedule.on_missed = "skip" + # Stash a missed fire 6 hours ago. + wf.next_run_at = datetime.now(timezone.utc) - timedelta(hours=6) + storage.save_workflow(wf) + scheduler.reconcile_on_startup() + after = storage.get_workflow(wf.id) + assert after.next_run_at is not None + assert after.next_run_at > datetime.now(timezone.utc) + + +async def test_reconcile_run_once_keeps_missed(monkeypatch): + """on_missed='run_once' => startup leaves next_run_at in the past so + the first tick fires a catch-up.""" + from backend.apps.workflows import storage, scheduler + wf = _make_wf() + wf.schedule.on_missed = "run_once" + missed = datetime.now(timezone.utc) - timedelta(hours=6) + wf.next_run_at = missed + storage.save_workflow(wf) + scheduler.reconcile_on_startup() + after = storage.get_workflow(wf.id) + assert after.next_run_at <= datetime.now(timezone.utc) + + +async def test_create_workflow_schedules_next_fire(): + """POST-like create path: enabled schedule => next_run_at populated + by compute_next_fire.""" + from backend.apps.workflows.models import Workflow, ScheduleConfig, WorkflowStep + from backend.apps.workflows import scheduler + wf = Workflow( + title="t", + steps=[WorkflowStep(text="hi")], + schedule=ScheduleConfig( + enabled=True, repeat_unit="week", repeat_every=1, on_days=[0], + hour=9, minute=0, timezone="America/Los_Angeles", + ), + ) + nxt = scheduler.compute_next_fire(wf) + assert nxt is not None + assert nxt > datetime.now(timezone.utc) + tz = ZoneInfo("America/Los_Angeles") + local = nxt.astimezone(tz) + assert local.weekday() == 6 # Python: Sunday + assert (local.hour, local.minute) == (9, 0) + + +async def test_next_run_at_advances_after_fire(monkeypatch): + """After a fire the loop should re-compute next_run_at into the + future and persist it, so the same fire can't repeat in the same + minute.""" + from backend.apps.workflows import storage, scheduler, executor + + fired = asyncio.Event() + + async def fake_execute(wf, triggered_by="schedule", scheduled_for=None): + from backend.apps.workflows.models import WorkflowRun + run = WorkflowRun( + workflow_id=wf.id, status="success", scheduled_for=scheduled_for, + started_at=datetime.now(timezone.utc), finished_at=datetime.now(timezone.utc), + triggered_by=triggered_by, + ) + storage.record_run(run) + fired.set() + return run + + monkeypatch.setattr(executor, "execute", fake_execute) + + await scheduler.start() + try: + wf = _make_wf() + armed_at = datetime.now(timezone.utc) + timedelta(seconds=1) + wf.next_run_at = armed_at + storage.save_workflow(wf) + scheduler.kick() + await asyncio.wait_for(fired.wait(), timeout=5.0) + # Give the loop one extra tick to persist next_run_at. + await asyncio.sleep(0.2) + finally: + await scheduler.stop() + + after = storage.get_workflow(wf.id) + assert after.next_run_at is not None + assert after.next_run_at > armed_at, "scheduler did not advance next_run_at past the slot it just fired" + + +async def test_kick_wakes_loop_before_timeout(monkeypatch): + """kick() should wake the loop early so manual schedule edits don't + have to wait a full minute for the next tick boundary.""" + from backend.apps.workflows import storage, scheduler, executor + + fired = asyncio.Event() + + async def fake_execute(wf, triggered_by="schedule", scheduled_for=None): + from backend.apps.workflows.models import WorkflowRun + run = WorkflowRun( + workflow_id=wf.id, status="success", + started_at=datetime.now(timezone.utc), + finished_at=datetime.now(timezone.utc), triggered_by=triggered_by, + ) + storage.record_run(run) + fired.set() + return run + + monkeypatch.setattr(executor, "execute", fake_execute) + + await scheduler.start() + try: + wf = _make_wf() + wf.next_run_at = datetime.now(timezone.utc) - timedelta(seconds=1) + storage.save_workflow(wf) + scheduler.kick() + # Without kick(), the loop would sleep up to 60s before checking + # the freshly-saved workflow. With kick, it should fire fast. + await asyncio.wait_for(fired.wait(), timeout=3.0) + finally: + await scheduler.stop()