From c284a8967f489118d5aae0c5d31209c3b9b00eb4 Mon Sep 17 00:00:00 2001 From: abccodes Date: Tue, 23 Jun 2026 03:29:40 -0700 Subject: [PATCH] [aidan] test/schedule: cover executor pipeline, storage durability, and recurrence gaps --- backend/tests/conftest.py | 137 +++++++++++++++++ backend/tests/test_executor_pipeline.py | 178 ++++++++++++++++++++++ backend/tests/test_schedule_recovery.py | 59 +++++++ backend/tests/test_schedule_recurrence.py | 121 +++++++++++++++ backend/tests/test_workflows_api.py | 115 ++++++++++++++ backend/tests/test_workflows_storage.py | 120 +++++++++++++++ 6 files changed, 730 insertions(+) create mode 100644 backend/tests/test_executor_pipeline.py create mode 100644 backend/tests/test_schedule_recovery.py create mode 100644 backend/tests/test_schedule_recurrence.py create mode 100644 backend/tests/test_workflows_api.py create mode 100644 backend/tests/test_workflows_storage.py diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index e17c9835..09be8e5e 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -4,10 +4,19 @@ Isolate the persistent browser-skill store (and metrics) into throwaway temp dirs for the whole test session, so tests never write skills/metrics into the real ~/Library/Application Support/OpenSwarm/data tree (which would pollute the dev machine and let a stale persisted skill leak across test runs). + +The workflow fixtures below (isolated_workflows_data, reset_scheduler_state, +make_wf, fake_agent_manager) are deliberately NOT autouse: only the scheduled- +workflows suites opt in, so they never touch unrelated browser/service tests. +The two pre-existing workflow suites still carry their own local copies; new +suites lean on these shared ones instead of re-pasting the boilerplate. """ +import asyncio import os import tempfile +from datetime import datetime, timezone +from types import SimpleNamespace import pytest @@ -38,3 +47,131 @@ def _isolate_browser_state(monkeypatch): _reset() yield _reset() + + +@pytest.fixture +def isolated_workflows_data(monkeypatch, tmp_path): + """Point the workflow store at a throwaway tmpdir and start every test with + empty in-memory caches, so nothing leaks into a real install or across + tests. Mirrors the local fixtures in test_workflows_semantics.py and + test_schedule_e2e.py; opt in by requesting this fixture (or depend on it + from a file-local autouse).""" + from backend.apps.workflows import storage as p_storage + from backend.apps.workflows import audit as p_audit + from backend.apps.workflows import escalation as p_escalation + monkeypatch.setattr(p_storage, "DATA_DIR", str(tmp_path / "workflows")) + monkeypatch.setattr(p_storage, "RUNS_DIR", str(tmp_path / "workflows" / "runs")) + monkeypatch.setattr(p_storage, "PAUSED_FILE", str(tmp_path / "workflows" / "paused.json")) + monkeypatch.setattr(p_storage, "MISSED_FILE", str(tmp_path / "workflows" / "missed.json")) + monkeypatch.setattr(p_storage, "_workflow_cache", {}) + monkeypatch.setattr(p_storage, "_runs_cache", {}) + monkeypatch.setattr(p_storage, "_missed_cache", []) + monkeypatch.setattr(p_storage, "_cache_loaded", False) + monkeypatch.setattr(p_storage, "_paused", False) + monkeypatch.setattr(p_audit, "AUDIT_DIR", str(tmp_path / "workflows" / "audit")) + p_escalation._tasks.clear() + p_escalation._state.clear() + yield + + +@pytest.fixture +def reset_scheduler_state(monkeypatch): + """Reset the scheduler + executor module globals that otherwise survive + across tests and cause order-dependent flakes: a stale _wake Event bound to + a dead loop, a lingering _loop_task, a cached host tz, or a leftover entry + in the _running / control maps.""" + from backend.apps.workflows import scheduler as p_scheduler + from backend.apps.workflows import executor as p_executor + p_scheduler._loop_task = None + p_scheduler._wake = asyncio.Event() + monkeypatch.setattr(p_scheduler, "_host_tz_cache", None) + p_executor._running.clear() + p_executor._run_control.clear() + p_executor._run_pause_override.clear() + yield + p_executor._running.clear() + p_executor._run_control.clear() + p_executor._run_pause_override.clear() + + +@pytest.fixture +def make_wf(): + """Factory for a Workflow with a sane default daily-9am-LA schedule; + override any field via kwargs.""" + def p_build(**overrides): + from backend.apps.workflows.models import Workflow, ScheduleConfig, WorkflowStep + base = dict( + title="t", + 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) + return p_build + + +class FakeAgentManager: + """Stand-in for agent_manager so executor.execute() drives a full run + without a live LLM. Records launched configs + every step prompt sent, and + lets a test script the per-step terminal status (the value + _await_session_idle reads): default 'completed' (-> advance), or 'error' / + 'stopped' to exercise the failure paths. + """ + + def __init__(self): + self.sessions: dict[str, SimpleNamespace] = {} + self.tasks: dict[str, object] = {} + self.launched_configs: list[object] = [] + self.sent_messages: list[str] = [] + # statuses[i] is the session status after the i-th send_message; absent + # entries default to 'completed'. cost_usd lands on the run. + self.statuses: list[str] = [] + self.cost_usd: float = 0.0 + + async def launch_agent(self, config): + self.launched_configs.append(config) + sid = f"sess-{len(self.sessions)}" + sess = SimpleNamespace(id=sid, status="completed", cost_usd=self.cost_usd, messages=[]) + self.sessions[sid] = sess + return sess + + async def send_message(self, session_id, text, hidden=False): + i = len(self.sent_messages) + self.sent_messages.append(text) + sess = self.sessions.get(session_id) + if sess is not None: + sess.status = self.statuses[i] if i < len(self.statuses) else "completed" + + async def close_session(self, session_id): + self.sessions.pop(session_id, None) + + +@pytest.fixture +def fake_agent_manager(monkeypatch): + """Patch the agent_manager seam (plus the ws + notifier side effects) that + executor.execute imports lazily, and hand back the FakeAgentManager so the + test can script statuses and assert on launched configs / sent steps.""" + from backend.apps.agents import agent_manager as p_am + from backend.apps.agents.core.ws_manager import ws_manager as p_ws + from backend.apps.workflows import notifier as p_notifier + + fake = FakeAgentManager() + monkeypatch.setattr(p_am.agent_manager, "sessions", fake.sessions) + monkeypatch.setattr(p_am.agent_manager, "tasks", fake.tasks) + monkeypatch.setattr(p_am.agent_manager, "launch_agent", fake.launch_agent) + monkeypatch.setattr(p_am.agent_manager, "send_message", fake.send_message) + monkeypatch.setattr(p_am.agent_manager, "close_session", fake.close_session) + + async def p_noop_async(*args, **kwargs): + return None + + monkeypatch.setattr(p_am, "set_workflow_approval_memory", lambda *a, **k: None) + monkeypatch.setattr(p_am, "set_workflow_approval_step", lambda *a, **k: None) + monkeypatch.setattr(p_am, "clear_workflow_approval_memory", lambda *a, **k: None) + monkeypatch.setattr(p_am, "get_workflow_step_usage", lambda *a, **k: {}) + monkeypatch.setattr(p_ws, "broadcast_global", p_noop_async) + monkeypatch.setattr(p_notifier, "notify_run_complete", p_noop_async) + return fake diff --git a/backend/tests/test_executor_pipeline.py b/backend/tests/test_executor_pipeline.py new file mode 100644 index 00000000..99869716 --- /dev/null +++ b/backend/tests/test_executor_pipeline.py @@ -0,0 +1,178 @@ +"""Executor pipeline integration: drive executor.execute() end to end with a +faked agent_manager (the fake_agent_manager fixture) and real on-disk storage. + +The existing semantics suite covers the cost-cap skip, p_ran_late in isolation, +and the _persist_run_fields merge/delete races. This fills the gap those leave: +the full run lifecycle through a scripted agent, the terminal-status matrix, +runs_count accounting, step ordering, and the _running lock release on a persist +failure (the crash-hardening fix). + +Run: + cd backend && .venv/bin/python -m pytest tests/test_executor_pipeline.py -v +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timedelta, timezone + +import pytest + + +@pytest.fixture(autouse=True) +def _wf_env(isolated_workflows_data, reset_scheduler_state): + yield + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +# --- terminal-status matrix -------------------------------------------------- + +def test_successful_run_records_success(make_wf, fake_agent_manager): + from backend.apps.workflows import storage, executor + wf = make_wf() + storage.save_workflow(wf) + run = _run(executor.execute(wf, triggered_by="schedule")) + assert run.status == "success" + assert run.finished_at is not None + stored = storage.list_runs(wf.id, limit=10) + assert stored[0].status == "success" + assert storage.get_workflow(wf.id).last_run_status == "success" + + +def test_agent_error_marks_failure(make_wf, fake_agent_manager): + from backend.apps.workflows import storage, executor + fake_agent_manager.statuses = ["error"] + wf = make_wf() + storage.save_workflow(wf) + run = _run(executor.execute(wf, triggered_by="schedule")) + assert run.status == "failure" + assert run.error == "Agent session entered error state" + + +def test_scheduled_run_late_start_marks_ran_late(make_wf, fake_agent_manager): + """A scheduled fire whose slot is already >5min in the past when it starts + is classified ran_late, not success, even though the agent succeeds.""" + from backend.apps.workflows import storage, executor + wf = make_wf() + storage.save_workflow(wf) + slot = datetime.now(timezone.utc) - timedelta(minutes=10) + run = _run(executor.execute(wf, triggered_by="schedule", scheduled_for=slot)) + assert run.status == "ran_late" + assert storage.get_workflow(wf.id).last_run_status == "ran_late" + + +def test_overlap_skipped_when_already_running(make_wf, fake_agent_manager): + """A second fire while a run for the same workflow holds _running is skipped + with a clear reason, never launching a second agent.""" + from backend.apps.workflows import storage, executor + wf = make_wf() + storage.save_workflow(wf) + executor._running[wf.id] = "in-flight-run" + run = _run(executor.execute(wf, triggered_by="schedule")) + assert run.status == "skipped" + assert run.error == "Previous run still active" + assert fake_agent_manager.launched_configs == [] + + +# --- runs_count accounting --------------------------------------------------- + +def test_schedule_run_bumps_runs_count(make_wf, fake_agent_manager): + from backend.apps.workflows import storage, executor + wf = make_wf() + storage.save_workflow(wf) + _run(executor.execute(wf, triggered_by="schedule")) + assert storage.get_workflow(wf.id).schedule.runs_count == 1 + + +def test_manual_run_does_not_bump_runs_count(make_wf, fake_agent_manager): + """Manual runs are free presses of the Run button; they must not count + against max_runs.""" + from backend.apps.workflows import storage, executor + wf = make_wf() + storage.save_workflow(wf) + _run(executor.execute(wf, triggered_by="manual")) + assert storage.get_workflow(wf.id).schedule.runs_count == 0 + + +# --- step iteration ---------------------------------------------------------- + +def test_steps_sent_in_order(make_wf, fake_agent_manager): + from backend.apps.workflows import storage, executor + from backend.apps.workflows.models import WorkflowStep + wf = make_wf(steps=[WorkflowStep(text="one"), WorkflowStep(text="two"), WorkflowStep(text="three")]) + storage.save_workflow(wf) + _run(executor.execute(wf, triggered_by="schedule")) + assert fake_agent_manager.sent_messages == ["one", "two", "three"] + + +def test_disabled_and_blank_steps_are_skipped(make_wf, fake_agent_manager): + from backend.apps.workflows import storage, executor + from backend.apps.workflows.models import WorkflowStep + wf = make_wf(steps=[ + WorkflowStep(text="run me"), + WorkflowStep(text="muted", enabled=False), + WorkflowStep(text=" "), + WorkflowStep(text="me too"), + ]) + storage.save_workflow(wf) + _run(executor.execute(wf, triggered_by="schedule")) + assert fake_agent_manager.sent_messages == ["run me", "me too"] + + +def test_mid_sequence_error_halts_remaining_steps(make_wf, fake_agent_manager): + """An error on step 2 stops the run; step 3 is never dispatched.""" + from backend.apps.workflows import storage, executor + from backend.apps.workflows.models import WorkflowStep + fake_agent_manager.statuses = ["completed", "error"] + wf = make_wf(steps=[WorkflowStep(text="s1"), WorkflowStep(text="s2"), WorkflowStep(text="s3")]) + storage.save_workflow(wf) + run = _run(executor.execute(wf, triggered_by="schedule")) + assert run.status == "failure" + assert fake_agent_manager.sent_messages == ["s1", "s2"] + + +def test_no_runnable_steps_fails(make_wf, fake_agent_manager): + from backend.apps.workflows import storage, executor + from backend.apps.workflows.models import WorkflowStep + wf = make_wf(steps=[WorkflowStep(text="", enabled=True)]) + storage.save_workflow(wf) + run = _run(executor.execute(wf, triggered_by="schedule")) + assert run.status == "failure" + assert "no steps" in (run.error or "").lower() + assert fake_agent_manager.launched_configs == [] + + +# --- crash hardening: the lock must always be released ----------------------- + +def test_running_lock_released_on_persist_failure(make_wf, fake_agent_manager, monkeypatch): + """If persisting run fields throws after _running is claimed, the finally + must still release the lock; otherwise the workflow is wedged 'running' + forever and every future fire is skipped. Guards the fix that moved the + running-persist inside the try whose finally frees _running.""" + from backend.apps.workflows import storage, executor + wf = make_wf() + storage.save_workflow(wf) + + def p_boom(*args, **kwargs): + raise RuntimeError("disk gone") + + monkeypatch.setattr(executor, "_persist_run_fields", p_boom) + with pytest.raises(RuntimeError): + _run(executor.execute(wf, triggered_by="schedule")) + assert wf.id not in executor._running + + +def test_resolved_config_uses_workflow_model_and_tools(make_wf, fake_agent_manager): + """The launched AgentConfig reflects the workflow's model and, when frozen, + its configured tool set rather than the default surface.""" + from backend.apps.workflows import storage, executor + from backend.apps.workflows.models import ActionsConfig + wf = make_wf(model="opus", actions=ActionsConfig(freeze=True, configured_sets=["Read", "Grep"])) + storage.save_workflow(wf) + _run(executor.execute(wf, triggered_by="schedule")) + config = fake_agent_manager.launched_configs[0] + assert config.model == "opus" + assert config.allowed_tools == ["Read", "Grep"] diff --git a/backend/tests/test_schedule_recovery.py b/backend/tests/test_schedule_recovery.py new file mode 100644 index 00000000..31a7996a --- /dev/null +++ b/backend/tests/test_schedule_recovery.py @@ -0,0 +1,59 @@ +"""Recovery + missed-run edges the e2e suite leaves uncovered. + +test_schedule_e2e already covers reconcile capture, the over-cap collapse, and +the friendly stuck-run message. This adds the boundaries it skips: the summary +heal on a stuck run, and the exactly-at-cap / empty cases of _capture_missed. + +Run: + cd backend && .venv/bin/python -m pytest tests/test_schedule_recovery.py -v +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest + + +@pytest.fixture(autouse=True) +def _wf_env(isolated_workflows_data, reset_scheduler_state): + yield + + +def test_stuck_run_heals_workflow_summary(make_wf): + """When the dead 'running' run is also the workflow's last run, the reaper + must fix the summary too, not just the run row, or the detail header keeps + showing a spinner forever.""" + from backend.apps.workflows import storage, scheduler + from backend.apps.workflows.models import WorkflowRun + wf = make_wf() + run = WorkflowRun(workflow_id=wf.id, status="running") + wf.last_run_id = run.id + wf.last_run_status = "running" + storage.save_workflow(wf) + storage.record_run(run) + + scheduler._mark_stuck_runs_failed() + + healed = storage.get_workflow(wf.id) + assert healed.last_run_status == "failure" + + +def test_capture_missed_exactly_at_cap_keeps_all_no_skipped(make_wf): + from backend.apps.workflows import storage, scheduler + wf = make_wf() + storage.save_workflow(wf) + base = datetime(2026, 1, 1, tzinfo=timezone.utc) + missed = [base + timedelta(minutes=15 * i) for i in range(scheduler.PER_WORKFLOW_MISSED_CAP)] + scheduler._capture_missed(wf, missed) + assert len(storage.list_missed()) == scheduler.PER_WORKFLOW_MISSED_CAP + assert [r for r in storage.list_runs(wf.id, limit=50) if r.status == "skipped"] == [] + + +def test_capture_missed_empty_is_noop(make_wf): + from backend.apps.workflows import storage, scheduler + wf = make_wf() + storage.save_workflow(wf) + scheduler._capture_missed(wf, []) + assert storage.list_missed() == [] + assert storage.list_runs(wf.id) == [] diff --git a/backend/tests/test_schedule_recurrence.py b/backend/tests/test_schedule_recurrence.py new file mode 100644 index 00000000..154a66d4 --- /dev/null +++ b/backend/tests/test_schedule_recurrence.py @@ -0,0 +1,121 @@ +"""Pure recurrence + validator units not already covered by the semantics +suite (which owns the DST, monthly, daily/weekly-interval, and occurrence +timezone/end-condition cases). These functions take explicit reference +datetimes and touch no disk, so they need no scaffolding. + +Run: + cd backend && .venv/bin/python -m pytest tests/test_schedule_recurrence.py -v +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest + + +def _sched(**overrides): + from backend.apps.workflows.models import ScheduleConfig + base = dict(enabled=True, repeat_unit="day", repeat_every=1, hour=9, minute=0, timezone="UTC") + base.update(overrides) + return ScheduleConfig(**base) + + +# --- minute / hour intervals (day/week/month intervals live in semantics) ---- + +def test_minute_interval_steps_by_repeat_every_anchored(): + from backend.apps.workflows.scheduler import _next_fire_after + sched = _sched(repeat_unit="minute", repeat_every=30) + anchor = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc) + # Asking from 00:10 on the same grid -> next 30-min point is 00:30. + ref = datetime(2026, 6, 1, 0, 10, tzinfo=timezone.utc) + assert _next_fire_after(sched, ref, anchor) == datetime(2026, 6, 1, 0, 30, tzinfo=timezone.utc) + + +def test_minute_repeat_every_floored_at_15(): + """A sub-15 interval would be a token-burning loop; the floor lifts it to + 15 (enforced both by the validator and the scheduler's own max(15, ...)).""" + from backend.apps.workflows.scheduler import _next_fire_after + sched = _sched(repeat_unit="minute", repeat_every=5) + anchor = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc) + ref = datetime(2026, 6, 1, 0, 1, tzinfo=timezone.utc) + assert _next_fire_after(sched, ref, anchor) == datetime(2026, 6, 1, 0, 15, tzinfo=timezone.utc) + + +def test_hourly_interval_fires_at_configured_minute(): + from backend.apps.workflows.scheduler import _next_fire_after + sched = _sched(repeat_unit="hour", repeat_every=3, minute=20) + anchor = datetime(2026, 6, 1, 0, 20, tzinfo=timezone.utc) + ref = datetime(2026, 6, 1, 1, 0, tzinfo=timezone.utc) + assert _next_fire_after(sched, ref, anchor) == datetime(2026, 6, 1, 3, 20, tzinfo=timezone.utc) + + +# --- p_first_after grid math ------------------------------------------------- + +def test_p_first_after_before_anchor_returns_anchor(): + from backend.apps.workflows.scheduler import p_first_after + anchor = datetime(2026, 6, 1, 12, 0, tzinfo=timezone.utc) + ref = datetime(2026, 6, 1, 9, 0, tzinfo=timezone.utc) + assert p_first_after(anchor, ref, timedelta(hours=1)) == anchor + + +def test_p_first_after_is_strict_on_exact_grid_point(): + from backend.apps.workflows.scheduler import p_first_after + anchor = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc) + ref = datetime(2026, 6, 1, 2, 0, tzinfo=timezone.utc) # exactly on the grid + # strictly-after means we advance to the next point, not return ref. + assert p_first_after(anchor, ref, timedelta(hours=1)) == datetime(2026, 6, 1, 3, 0, tzinfo=timezone.utc) + + +# --- fires_in_window (backs the cost estimate) ------------------------------- + +def test_fires_in_window_zero_when_disabled(): + from backend.apps.workflows import scheduler + from backend.apps.workflows.models import Workflow, WorkflowStep + wf = Workflow(title="t", steps=[WorkflowStep(text="hi")], schedule=_sched(enabled=False)) + assert scheduler.fires_in_window(wf, days=30) == 0 + + +def test_fires_in_window_capped_by_remaining_max_runs(): + """With 2 runs left on a daily schedule, a 30-day window projects exactly 2 + fires, not 30.""" + from backend.apps.workflows import scheduler + from backend.apps.workflows.models import Workflow, WorkflowStep + wf = Workflow(title="t", steps=[WorkflowStep(text="hi")], + schedule=_sched(max_runs=10, runs_count=8)) + assert scheduler.fires_in_window(wf, days=30) == 2 + + +# --- occurrences_between cap ------------------------------------------------- + +def test_occurrences_between_respects_cap(): + from backend.apps.workflows import scheduler + from backend.apps.workflows.models import Workflow, WorkflowStep + wf = Workflow(title="t", steps=[WorkflowStep(text="hi")], schedule=_sched(repeat_unit="day")) + wf.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc) + fires = scheduler.occurrences_between( + wf, + datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc), + datetime(2026, 6, 30, 0, 0, tzinfo=timezone.utc), + cap=5, + ) + assert len(fires) == 5 + + +# --- ScheduleConfig validators ----------------------------------------------- + +def test_clean_on_days_drops_out_of_range_and_dedupes_preserving_order(): + from backend.apps.workflows.models import ScheduleConfig + sched = ScheduleConfig(repeat_unit="week", on_days=[3, 7, -1, 3, 0, 6]) + assert sched.on_days == [3, 0, 6] + + +def test_interval_bounds_clamp_minute_unit_to_15_1440(): + from backend.apps.workflows.models import ScheduleConfig + assert ScheduleConfig(repeat_unit="minute", repeat_every=5).repeat_every == 15 + assert ScheduleConfig(repeat_unit="minute", repeat_every=99999).repeat_every == 1440 + + +def test_interval_bounds_clamp_other_units_to_365(): + from backend.apps.workflows.models import ScheduleConfig + assert ScheduleConfig(repeat_unit="day", repeat_every=99999).repeat_every == 365 diff --git a/backend/tests/test_workflows_api.py b/backend/tests/test_workflows_api.py new file mode 100644 index 00000000..700d0ee0 --- /dev/null +++ b/backend/tests/test_workflows_api.py @@ -0,0 +1,115 @@ +"""Route-handler behavior for the trash lifecycle and the global pause switch, +called as coroutines (the established pattern in test_workflows_semantics.py). +CRUD create/patch + If-Match + the calendar endpoint already live in the +semantics suite; this covers the soft-delete -> restore -> purge flow and +pause-all / resume-all, including their 404 guards. + +Run: + cd backend && .venv/bin/python -m pytest tests/test_workflows_api.py -v +""" + +from __future__ import annotations + +import asyncio +from datetime import datetime, timedelta, timezone + +import pytest + + +@pytest.fixture(autouse=True) +def _wf_env(isolated_workflows_data, reset_scheduler_state, monkeypatch): + # Silence the ws fan-out the handlers fire; nothing is listening in-test. + from backend.apps.agents.core.ws_manager import ws_manager + + async def p_noop(*args, **kwargs): + return None + + monkeypatch.setattr(ws_manager, "broadcast_global", p_noop) + yield + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def test_soft_delete_hides_and_disables_schedule(make_wf): + from backend.apps.workflows import storage + from backend.apps.workflows.workflows import delete_workflow + wf = make_wf() + wf.next_run_at = datetime.now(timezone.utc) + timedelta(hours=1) + storage.save_workflow(wf) + + res = _run(delete_workflow(wf.id)) + assert res == {"ok": True} + after = storage.get_workflow(wf.id) + assert after.deleted_at is not None + assert after.schedule.enabled is False + assert after.next_run_at is None + assert wf.id not in {w.id for w in storage.list_workflows()} + assert wf.id in {w.id for w in storage.list_deleted_workflows()} + + +def test_soft_delete_drops_pending_missed(make_wf): + from backend.apps.workflows import storage + from backend.apps.workflows.workflows import delete_workflow + from backend.apps.workflows.models import MissedRun + wf = make_wf() + storage.save_workflow(wf) + storage.add_missed(MissedRun(workflow_id=wf.id, scheduled_for=datetime.now(timezone.utc))) + _run(delete_workflow(wf.id)) + assert storage.list_missed() == [] + + +def test_restore_brings_back_with_schedule_still_off(make_wf): + from backend.apps.workflows import storage + from backend.apps.workflows.workflows import delete_workflow, restore_workflow + wf = make_wf() + storage.save_workflow(wf) + _run(delete_workflow(wf.id)) + enriched = _run(restore_workflow(wf.id)) + assert enriched["id"] == wf.id + after = storage.get_workflow(wf.id) + assert after.deleted_at is None + assert after.schedule.enabled is False # restore is deliberate; user re-arms + assert wf.id in {w.id for w in storage.list_workflows()} + + +def test_purge_only_from_trash_and_removes_record(make_wf): + from backend.apps.workflows import storage + from backend.apps.workflows.workflows import delete_workflow, purge_workflow + wf = make_wf() + storage.save_workflow(wf) + _run(delete_workflow(wf.id)) + res = _run(purge_workflow(wf.id)) + assert res == {"ok": True} + assert storage.get_workflow(wf.id) is None + + +def test_delete_restore_purge_guards_404(make_wf): + from fastapi import HTTPException + from backend.apps.workflows import storage + from backend.apps.workflows.workflows import delete_workflow, restore_workflow, purge_workflow + wf = make_wf() + storage.save_workflow(wf) + + # restore / purge refuse a workflow that isn't in trash. + for fn in (restore_workflow, purge_workflow): + with pytest.raises(HTTPException) as exc: + _run(fn(wf.id)) + assert exc.value.status_code == 404 + + # double-delete is a 404 (already gone). + _run(delete_workflow(wf.id)) + with pytest.raises(HTTPException) as exc: + _run(delete_workflow(wf.id)) + assert exc.value.status_code == 404 + + +def test_pause_all_and_resume_all_flip_global_flag(make_wf): + from backend.apps.workflows import storage + from backend.apps.workflows.workflows import pause_all_schedules, resume_all_schedules + assert storage.get_paused() is False + assert _run(pause_all_schedules()) == {"paused": True} + assert storage.get_paused() is True + assert _run(resume_all_schedules()) == {"paused": False} + assert storage.get_paused() is False diff --git a/backend/tests/test_workflows_storage.py b/backend/tests/test_workflows_storage.py new file mode 100644 index 00000000..877d8333 --- /dev/null +++ b/backend/tests/test_workflows_storage.py @@ -0,0 +1,120 @@ +"""Storage durability: the crash-safe write path and the bounded caches. + +These guard properties the rest of the suite assumes but never exercises: a +power-off mid-write must not corrupt or orphan files, a record that does end up +corrupt must be skipped rather than crash the loader, and the per-workflow run +log + missed-run list must stay bounded. + +Run: + cd backend && .venv/bin/python -m pytest tests/test_workflows_storage.py -v +""" + +from __future__ import annotations + +import json +import os +from datetime import datetime, timedelta, timezone + +import pytest + + +@pytest.fixture(autouse=True) +def _wf_env(isolated_workflows_data): + yield + + +# --- atomic write ------------------------------------------------------------ + +def test_atomic_write_round_trips(tmp_path): + from backend.apps.workflows import storage + storage._ensure_dirs() + path = os.path.join(storage.DATA_DIR, "thing.json") + storage.p_atomic_write_json(path, {"a": 1, "b": [2, 3]}) + with open(path) as f: + assert json.load(f) == {"a": 1, "b": [2, 3]} + + +def test_atomic_write_leaves_no_temp_and_preserves_old_on_failure(monkeypatch): + """A failure mid-write must keep the previous complete file intact and drop + no .tmp sibling behind (the loader would never read a .tmp, but a litter of + them is its own bug).""" + from backend.apps.workflows import storage + storage._ensure_dirs() + path = os.path.join(storage.DATA_DIR, "keep.json") + storage.p_atomic_write_json(path, {"v": "original"}) + + def p_boom(*args, **kwargs): + raise RuntimeError("write died") + + monkeypatch.setattr(storage.json, "dump", p_boom) + with pytest.raises(RuntimeError): + storage.p_atomic_write_json(path, {"v": "new"}) + + with open(path) as f: + assert json.load(f) == {"v": "original"} + leftovers = [n for n in os.listdir(storage.DATA_DIR) if n.endswith(".tmp")] + assert leftovers == [] + + +# --- corrupt-record resilience ----------------------------------------------- + +def test_corrupt_workflow_record_is_skipped_not_fatal(make_wf): + """A truncated .json must not take down the whole load; the bad record + silently drops out and the good ones still come back.""" + from backend.apps.workflows import storage + good = make_wf(title="good") + storage.save_workflow(good) + storage._ensure_dirs() + with open(os.path.join(storage.DATA_DIR, "broken.json"), "w") as f: + f.write('{"id": "broken", "title": "trunc') # deliberately unterminated + + storage._cache_loaded = False + ids = {w.id for w in storage.list_workflows()} + assert good.id in ids + assert "broken" not in ids + + +def test_corrupt_runs_file_yields_empty_history(make_wf): + from backend.apps.workflows import storage + wf = make_wf() + storage.save_workflow(wf) + storage._ensure_dirs() + with open(os.path.join(storage.RUNS_DIR, f"{wf.id}.json"), "w") as f: + f.write("not json at all") + + storage._cache_loaded = False + assert storage.list_runs(wf.id) == [] + + +# --- bounded caches ---------------------------------------------------------- + +def test_missed_cache_capped_to_newest(make_wf): + """add_missed past MAX_MISSED keeps the newest by scheduled_for so a card + the user never acts on can't grow the file without bound.""" + from backend.apps.workflows import storage + from backend.apps.workflows.models import MissedRun + wf = make_wf() + storage.save_workflow(wf) + base = datetime(2026, 1, 1, tzinfo=timezone.utc) + total = storage.MAX_MISSED + 25 + for i in range(total): + storage.add_missed(MissedRun(workflow_id=wf.id, scheduled_for=base + timedelta(minutes=i))) + kept = storage.list_missed() + assert len(kept) == storage.MAX_MISSED + # The oldest 25 fell off; the kept set starts at minute 25. + earliest = min(m.scheduled_for for m in kept) + assert earliest == base + timedelta(minutes=25) + + +def test_run_history_bounded_per_workflow(make_wf): + from backend.apps.workflows import storage + from backend.apps.workflows.models import WorkflowRun + wf = make_wf() + storage.save_workflow(wf) + over = storage.RUNS_PER_WORKFLOW + 10 + for i in range(over): + storage.record_run(WorkflowRun( + workflow_id=wf.id, status="success", + started_at=datetime(2026, 1, 1, tzinfo=timezone.utc) + timedelta(minutes=i), + )) + assert len(storage._runs_cache[wf.id]) == storage.RUNS_PER_WORKFLOW