feat(console): schedule preview endpoint — next-3-runs read-out backend

POST /v1/api/admin/schedules/preview validates {schedule_type, cron_expr,
at_time} with the same _validate_schedule_fields the CRUD path uses and
returns the next three croniter firings. Pure compute, no storage touch;
invalid input answers 200 {valid:false, error} because the schedule
editor renders it live while the user types. Registered ahead of the
{task_id} routes so the literal segment wins.
This commit is contained in:
Patrick Buckley
2026-06-09 21:00:38 -07:00
parent 8f415f9c68
commit f086e38e37
2 changed files with 117 additions and 0 deletions
+71
View File
@@ -21,6 +21,7 @@ from turnstone.console.server import (
admin_get_schedule,
admin_list_schedule_runs,
admin_list_schedules,
admin_preview_schedule,
admin_update_schedule,
)
from turnstone.core.auth import AuthResult
@@ -54,6 +55,11 @@ def client(storage):
routes=[
Route("/api/admin/schedules", admin_list_schedules),
Route("/api/admin/schedules", admin_create_schedule, methods=["POST"]),
Route(
"/api/admin/schedules/preview",
admin_preview_schedule,
methods=["POST"],
),
Route("/api/admin/schedules/{task_id}", admin_get_schedule),
Route(
"/api/admin/schedules/{task_id}",
@@ -283,3 +289,68 @@ class TestScheduleAPI:
resp = client.get(f"/v1/api/admin/schedules/{task_id}/runs?limit=abc")
assert resp.status_code == 200
assert resp.json()["runs"] == []
class TestPreviewSchedule:
"""POST /v1/api/admin/schedules/preview — the editor's NEXT RUNS read-out."""
def test_valid_cron_returns_three_ascending_runs(self, client):
resp = client.post(
"/v1/api/admin/schedules/preview",
json={"schedule_type": "cron", "cron_expr": "0 6 * * *"},
)
assert resp.status_code == 200
data = resp.json()
assert data["valid"] is True
assert data["error"] == ""
assert len(data["next"]) == 3
assert data["next"] == sorted(data["next"])
# All at 06:00 (the daily expression's only firing time)
assert all(t.endswith("T06:00:00") for t in data["next"])
def test_invalid_cron_is_a_200_with_the_message(self, client):
resp = client.post(
"/v1/api/admin/schedules/preview",
json={"schedule_type": "cron", "cron_expr": "not a cron"},
)
assert resp.status_code == 200
data = resp.json()
assert data["valid"] is False
assert "Invalid cron expression" in data["error"]
assert data["next"] == []
def test_missing_cron_expr(self, client):
resp = client.post(
"/v1/api/admin/schedules/preview",
json={"schedule_type": "cron", "cron_expr": ""},
)
data = resp.json()
assert data["valid"] is False
assert "cron_expr is required" in data["error"]
def test_at_future_echoes_the_time(self, client):
resp = client.post(
"/v1/api/admin/schedules/preview",
json={"schedule_type": "at", "at_time": "2030-01-01T12:00:00+00:00"},
)
data = resp.json()
assert data["valid"] is True
assert data["next"] == ["2030-01-01T12:00:00+00:00"]
def test_at_in_the_past_is_invalid(self, client):
resp = client.post(
"/v1/api/admin/schedules/preview",
json={"schedule_type": "at", "at_time": "2020-01-01T12:00:00+00:00"},
)
data = resp.json()
assert data["valid"] is False
assert "future" in data["error"]
def test_unknown_schedule_type(self, client):
resp = client.post(
"/v1/api/admin/schedules/preview",
json={"schedule_type": "sometimes"},
)
data = resp.json()
assert data["valid"] is False
assert "schedule_type" in data["error"]
+46
View File
@@ -5599,6 +5599,45 @@ def _validate_schedule_fields(schedule_type: str, cron_expr: str, at_time: str)
return None
async def admin_preview_schedule(request: Request) -> JSONResponse:
"""POST /v1/api/admin/schedules/preview — validate timing, return next runs.
Pure compute (no storage): powers the schedule editor's NEXT RUNS read-out,
re-queried as the user types. Invalid input is a normal preview outcome
(the read-out renders the message live), so it answers 200 with
``valid: false`` rather than a 4xx.
"""
from turnstone.core.auth import require_permission
from turnstone.core.web_helpers import read_json_or_400
err = require_permission(request, "admin.schedules")
if err:
return err
body = await read_json_or_400(request)
if isinstance(body, JSONResponse):
return body
schedule_type = str(body.get("schedule_type", "")).strip()
cron_expr = str(body.get("cron_expr", "")).strip()[:256]
at_time = str(body.get("at_time", "")).strip()[:64]
verr = _validate_schedule_fields(schedule_type, cron_expr, at_time)
if verr:
return JSONResponse({"valid": False, "error": verr, "next": []})
if schedule_type == "at":
return JSONResponse({"valid": True, "error": "", "next": [at_time]})
from datetime import UTC, datetime
from croniter import croniter
cron = croniter(cron_expr, datetime.now(UTC))
runs = [cron.get_next(datetime).strftime("%Y-%m-%dT%H:%M:%S") for _ in range(3)]
return JSONResponse({"valid": True, "error": "", "next": runs})
async def admin_list_schedules(request: Request) -> JSONResponse:
"""GET /v1/api/admin/schedules — list all scheduled tasks."""
from turnstone.core.auth import require_permission
@@ -13047,6 +13086,13 @@ def create_app(
),
Route("/api/admin/schedules", admin_list_schedules),
Route("/api/admin/schedules", admin_create_schedule, methods=["POST"]),
# Registered before the {task_id} routes so the literal
# segment wins the match.
Route(
"/api/admin/schedules/preview",
admin_preview_schedule,
methods=["POST"],
),
Route("/api/admin/schedules/{task_id}", admin_get_schedule),
Route("/api/admin/schedules/{task_id}", admin_update_schedule, methods=["PUT"]),
Route(