199 lines
7.9 KiB
Python
199 lines
7.9 KiB
Python
# tests/test_scheduler.py — Scheduled tasks, type cron (#170)
|
|
"""Tests for backend.scheduler, its REST routes and AI tools.
|
|
|
|
The file store is redirected to tmp; task actions run against the TestVault
|
|
fixture vault (create/append reuse the real mutation services).
|
|
"""
|
|
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from backend import scheduler as _scheduler
|
|
from backend.tools.api import ToolContext, call_tool
|
|
|
|
|
|
@pytest.fixture
|
|
def isolated_tasks(tmp_path, monkeypatch):
|
|
"""Redirect the task store to a tmp file."""
|
|
tasks_file = tmp_path / "scheduled_tasks.json"
|
|
monkeypatch.setattr(_scheduler, "TASKS_FILE", tasks_file)
|
|
monkeypatch.setattr("backend.notify.CHANNELS_FILE", tmp_path / "notify_channels.json")
|
|
monkeypatch.setattr("backend.notify.SECRETS_FILE", tmp_path / "notify_secrets.json")
|
|
return tasks_file
|
|
|
|
|
|
def _ctx(**kwargs) -> ToolContext:
|
|
user = {"username": "tester", "role": "admin", "vaults": ["*"]}
|
|
kwargs.setdefault("audit_enabled", False)
|
|
return ToolContext(user=user, **kwargs)
|
|
|
|
|
|
class TestValidation:
|
|
def test_unknown_action_rejected(self, isolated_tasks):
|
|
with pytest.raises(ValueError):
|
|
_scheduler.create_task(
|
|
"x",
|
|
{"kind": "launch_rocket", "params": {}},
|
|
{"kind": "interval_hours", "hours": 1},
|
|
)
|
|
|
|
def test_interval_too_short_rejected(self, isolated_tasks):
|
|
with pytest.raises(ValueError):
|
|
_scheduler.create_task(
|
|
"x",
|
|
{"kind": "notify", "params": {"title": "T", "message": "M"}},
|
|
{"kind": "interval_hours", "hours": 0.1},
|
|
)
|
|
|
|
def test_bad_daily_time_rejected(self, isolated_tasks):
|
|
with pytest.raises(ValueError):
|
|
_scheduler.create_task(
|
|
"x",
|
|
{"kind": "notify", "params": {"title": "T", "message": "M"}},
|
|
{"kind": "daily_time", "at": "25:00"},
|
|
)
|
|
|
|
|
|
class TestNextRun:
|
|
def test_interval_from_now(self, isolated_tasks):
|
|
now = datetime(2026, 10, 4, 12, 0, tzinfo=timezone.utc)
|
|
nxt = _scheduler.compute_next_run(
|
|
{"schedule": {"kind": "interval_hours", "hours": 2}, "last_run_at": None}, now
|
|
)
|
|
assert nxt == now + timedelta(hours=2)
|
|
|
|
def test_daily_tomorrow_when_passed(self, isolated_tasks):
|
|
now = datetime(2026, 10, 4, 12, 0, tzinfo=timezone.utc)
|
|
nxt = _scheduler.compute_next_run({"schedule": {"kind": "daily_time", "at": "08:00"}}, now)
|
|
assert (nxt - now).total_seconds() == pytest.approx(20 * 3600)
|
|
|
|
def test_daily_today_when_upcoming(self, isolated_tasks):
|
|
now = datetime(2026, 10, 4, 7, 0, tzinfo=timezone.utc)
|
|
nxt = _scheduler.compute_next_run({"schedule": {"kind": "daily_time", "at": "08:00"}}, now)
|
|
assert (nxt - now).total_seconds() == pytest.approx(3600)
|
|
|
|
|
|
class TestExecution:
|
|
def test_create_and_run_append(self, client, test_vault_dir, isolated_tasks):
|
|
(Path(test_vault_dir) / "journal.md").write_text("# Journal\n", encoding="utf-8")
|
|
task = _scheduler.create_task(
|
|
"append",
|
|
{"kind": "append_to_file", "params": {"vault": "TestVault", "path": "journal.md", "content": "Ligne auto."}},
|
|
{"kind": "once_at", "at": "2020-01-01T00:00:00"},
|
|
)
|
|
outcome = _scheduler.run_task(task["id"], manual=True)
|
|
assert outcome["ok"] is True
|
|
assert "Ligne auto." in Path(test_vault_dir, "journal.md").read_text(encoding="utf-8")
|
|
stored = _scheduler.get_task(task["id"])
|
|
assert stored is not None and stored["last_status"] == "ok"
|
|
assert stored["run_count"] == 1
|
|
assert stored["enabled"] is False # one-shot consommé
|
|
|
|
def test_tick_runs_due_task(self, client, test_vault_dir, isolated_tasks):
|
|
task = _scheduler.create_task(
|
|
"once",
|
|
{"kind": "create_file", "params": {"vault": "TestVault", "path": "auto/tache.md", "content": "auto"}},
|
|
{"kind": "once_at", "at": "2020-01-01T00:00:00"},
|
|
)
|
|
outcomes = _scheduler.tick()
|
|
assert any(o.get("task_id") == task["id"] and o.get("ok") for o in outcomes)
|
|
assert Path(test_vault_dir, "auto", "tache.md").exists()
|
|
|
|
def test_failure_recorded(self, client, isolated_tasks):
|
|
task = _scheduler.create_task(
|
|
"fail",
|
|
{"kind": "append_to_file", "params": {"vault": "TestVault", "path": "missing/nope.md", "content": "x"}},
|
|
{"kind": "once_at", "at": "2020-01-01T00:00:00"},
|
|
)
|
|
outcome = _scheduler.run_task(task["id"], manual=True)
|
|
assert outcome["ok"] is False
|
|
stored = _scheduler.get_task(task["id"])
|
|
assert stored is not None and stored["last_status"] == "error"
|
|
assert stored["last_error"]
|
|
|
|
def test_run_unknown_raises(self, isolated_tasks):
|
|
with pytest.raises(KeyError):
|
|
_scheduler.run_task("nope", manual=True)
|
|
|
|
def test_delete(self, isolated_tasks):
|
|
task = _scheduler.create_task(
|
|
"del",
|
|
{"kind": "notify", "params": {"title": "T", "message": "M"}},
|
|
{"kind": "interval_hours", "hours": 24},
|
|
)
|
|
assert _scheduler.delete_task(task["id"]) is True
|
|
assert _scheduler.delete_task(task["id"]) is False
|
|
|
|
|
|
class TestSchedulerApi:
|
|
def test_crud_and_run(self, client, test_vault_dir, isolated_tasks):
|
|
created = client.post(
|
|
"/api/scheduler/tasks",
|
|
json={
|
|
"name": "t",
|
|
"action": {
|
|
"kind": "create_file",
|
|
"params": {"vault": "TestVault", "path": "sched/api.md", "content": "hello"},
|
|
},
|
|
"schedule": {"kind": "interval_hours", "hours": 24},
|
|
},
|
|
)
|
|
assert created.status_code == 200, created.text
|
|
task_id = created.json()["id"]
|
|
|
|
listed = client.get("/api/scheduler/tasks")
|
|
assert listed.status_code == 200
|
|
assert any(t["id"] == task_id for t in listed.json())
|
|
|
|
run = client.post(f"/api/scheduler/tasks/{task_id}/run")
|
|
assert run.status_code == 200, run.text
|
|
assert run.json()["ok"] is True
|
|
assert Path(test_vault_dir, "sched", "api.md").exists()
|
|
|
|
deleted = client.delete(f"/api/scheduler/tasks/{task_id}")
|
|
assert deleted.status_code == 200
|
|
|
|
def test_create_rejects_unknown_vault(self, client, isolated_tasks):
|
|
resp = client.post(
|
|
"/api/scheduler/tasks",
|
|
json={
|
|
"name": "t",
|
|
"action": {"kind": "create_file", "params": {"vault": "NoVault", "path": "x.md"}},
|
|
"schedule": {"kind": "interval_hours", "hours": 24},
|
|
},
|
|
)
|
|
assert resp.status_code in (400, 403, 404)
|
|
|
|
def test_run_unknown_404(self, client, isolated_tasks):
|
|
assert client.post("/api/scheduler/tasks/nope/run").status_code == 404
|
|
|
|
|
|
class TestSchedulerTools:
|
|
def test_create_list_delete_run(self, client, test_vault_dir, isolated_tasks):
|
|
created = call_tool(
|
|
"create_scheduled_task",
|
|
_ctx(confirmed=True),
|
|
{
|
|
"name": "tool-task",
|
|
"action": {
|
|
"kind": "create_file",
|
|
"params": {"vault": "TestVault", "path": "sched/tool.md", "content": "hi"},
|
|
},
|
|
"schedule": {"kind": "interval_hours", "hours": 24},
|
|
},
|
|
)
|
|
assert created.ok
|
|
task_id = created.data["id"]
|
|
|
|
listed = call_tool("list_scheduled_tasks", _ctx(), {})
|
|
assert listed.ok
|
|
assert any(t["id"] == task_id for t in listed.data)
|
|
|
|
ran = call_tool("run_scheduled_task_now", _ctx(confirmed=True), {"task_id": task_id})
|
|
assert ran.ok and ran.data["ok"] is True
|
|
|
|
deleted = call_tool("delete_scheduled_task", _ctx(confirmed=True), {"task_id": task_id})
|
|
assert deleted.ok
|