From d4c24e3f6f245fec7ee10dbeceecd27790e480c1 Mon Sep 17 00:00:00 2001 From: rayhpeng Date: Tue, 28 Jul 2026 14:46:13 +0800 Subject: [PATCH] feat(schedule): implement the persistence ports in SQL Two secondary adapters plus the spec mapping they share, and the first real payoff of the port boundary: test_schedule_fakes.py becomes a contract suite that runs all 31 cases against both the in-memory doubles and the SQL adapters on a real sqlite file. A rule stated in a port docstring now has to hold for both, and a divergence is a failure rather than a surprise in production. The queries come over unchanged. The claim statement's FOR UPDATE SKIP LOCKED and both protect_terminal conditional writes are this module's concurrency contract, not style, and the IntegrityError translation in add() is what lets the service collapse a lost active-slot race into the same outcome as its own non-atomic fast path. Two things the adapters own that the domain deliberately does not. The claiming process's identity is generated here, because who claimed a task is an identity rather than a rule and nothing reads it back. And `Unsupported schedule_type` is raised by the mapping, because ScheduleSpec only accepts the enum -- structural checks belong to the boundary, value rules to __post_init__, and both surface as the same domain error so the router maps one family. _to_domain introduces a failure mode the legacy repository did not have: a row whose stored schedule no longer parses. Single-row reads let it propagate; list reads skip and log, so one corrupt row cannot 500 an entire listing. The contract suite reaches past the port in exactly one place -- seeding a task that already carries a claim, a shape claim_due would never produce -- and says so where it does. --- .../infra/persistence/scheduled_task_runs.py | 181 ++++++++ .../app/infra/persistence/scheduled_tasks.py | 281 ++++++++++++ backend/app/infra/schedule/__init__.py | 1 + backend/app/infra/schedule/spec_mapping.py | 68 +++ backend/tests/test_schedule_fakes.py | 412 +++++++++++------- backend/tests/test_schedule_spec_mapping.py | 78 ++++ 6 files changed, 870 insertions(+), 151 deletions(-) create mode 100644 backend/app/infra/persistence/scheduled_task_runs.py create mode 100644 backend/app/infra/persistence/scheduled_tasks.py create mode 100644 backend/app/infra/schedule/__init__.py create mode 100644 backend/app/infra/schedule/spec_mapping.py create mode 100644 backend/tests/test_schedule_spec_mapping.py diff --git a/backend/app/infra/persistence/scheduled_task_runs.py b/backend/app/infra/persistence/scheduled_task_runs.py new file mode 100644 index 000000000..c2736cd8b --- /dev/null +++ b/backend/app/infra/persistence/scheduled_task_runs.py @@ -0,0 +1,181 @@ +"""SQL adapter for the schedule context's execution-record repository. + +Secondary adapter implementing `ScheduledRunRepository`. Queries are migrated +unchanged from the legacy repository. + +The load-bearing piece is the `IntegrityError` translation in `add`: the +partial unique index `uq_scheduled_task_run_active` is the atomic arbiter of +"at most one active run per task", and turning its rejection into +`ActiveRunConflictError` is what lets the service collapse a lost race into +exactly the same outcome as its own non-atomic fast path. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.domain.schedule.model import ( + ACTIVE_RUN_STATUSES, + TERMINAL_RUN_STATUSES, + ActiveRunConflictError, + RunStatus, + ScheduledRun, + TriggerKind, +) +from deerflow.domain.schedule.ports import ScheduledRunRepository + +# Transitional: the ORM row stays in the harness until engine, models, and +# migrations move into app/infra together. +from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow + +_ACTIVE_STATUS_VALUES = tuple(str(status) for status in ACTIVE_RUN_STATUSES) + + +def _tz_aware(value: datetime | None) -> datetime | None: + """SQLite drops tzinfo on read; stored values are always UTC.""" + return value if value is None or value.tzinfo is not None else value.replace(tzinfo=UTC) + + +class SqlScheduledRunRepository(ScheduledRunRepository): + """SQL implementation of the `ScheduledRunRepository` port.""" + + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._sf = session_factory + + @staticmethod + def _to_domain(row: ScheduledTaskRunRow) -> ScheduledRun: + return ScheduledRun( + record_id=row.id, + task_id=row.task_id, + thread_id=row.thread_id, + scheduled_for=_tz_aware(row.scheduled_for), + trigger=TriggerKind(row.trigger), + status=RunStatus(row.status), + run_id=row.run_id, + error=row.error, + started_at=_tz_aware(row.started_at), + finished_at=_tz_aware(row.finished_at), + created_at=_tz_aware(row.created_at), + ) + + async def add(self, run: ScheduledRun) -> ScheduledRun: + row = ScheduledTaskRunRow( + id=run.record_id, + task_id=run.task_id, + thread_id=run.thread_id, + run_id=run.run_id, + scheduled_for=run.scheduled_for, + trigger=str(run.trigger), + status=str(run.status), + error=run.error, + started_at=run.started_at, + finished_at=run.finished_at, + created_at=run.created_at, + ) + async with self._sf() as session: + session.add(row) + try: + await session.commit() + except IntegrityError: + await session.rollback() + # Only an active-status insert can trip the partial unique + # index; a terminal row (e.g. a skip tombstone) is outside its + # predicate and cannot conflict, so an IntegrityError there is + # a genuine fault and is re-raised untranslated. + if run.is_active: + raise ActiveRunConflictError(f"scheduled task {run.task_id!r} already has an active run") from None + raise + await session.refresh(row) + return self._to_domain(row) + + async def list_by_task(self, task_id: str, *, limit: int = 50, offset: int = 0) -> list[ScheduledRun]: + stmt = ( + select(ScheduledTaskRunRow) + .where(ScheduledTaskRunRow.task_id == task_id) + .order_by( + ScheduledTaskRunRow.created_at.desc(), + ScheduledTaskRunRow.id.desc(), + ) + .limit(limit) + .offset(offset) + ) + async with self._sf() as session: + result = await session.execute(stmt) + return [self._to_domain(row) for row in result.scalars()] + + async def count_active(self) -> int: + """Global count of active rows, used to bound cross-task concurrency.""" + stmt = select(func.count()).select_from(ScheduledTaskRunRow).where(ScheduledTaskRunRow.status.in_(_ACTIVE_STATUS_VALUES)) + async with self._sf() as session: + result = await session.execute(stmt) + return int(result.scalar() or 0) + + async def has_active(self, task_id: str) -> bool: + stmt = ( + select(ScheduledTaskRunRow.id) + .where( + ScheduledTaskRunRow.task_id == task_id, + ScheduledTaskRunRow.status.in_(_ACTIVE_STATUS_VALUES), + ) + .limit(1) + ) + async with self._sf() as session: + result = await session.execute(stmt) + return result.scalars().first() is not None + + async def update_status( + self, + record_id: str, + *, + status: RunStatus, + run_id: str | None = None, + error: str | None = None, + started_at: datetime | None = None, + finished_at: datetime | None = None, + protect_terminal: bool = False, + ) -> None: + async with self._sf() as session: + row = await session.get(ScheduledTaskRunRow, record_id) + if row is None: + return + if protect_terminal and RunStatus(row.status) in TERMINAL_RUN_STATUSES: + # The launch-path "running" write lost the race against the + # completion hook; keep the terminal status/error and only + # backfill bookkeeping the completion write could not know. + if row.run_id is None and run_id is not None: + row.run_id = run_id + if row.started_at is None and started_at is not None: + row.started_at = started_at + await session.commit() + return + row.status = str(status) + row.run_id = run_id + row.error = error + if started_at is not None: + row.started_at = started_at + if finished_at is not None: + row.finished_at = finished_at + await session.commit() + + async def mark_stale_active(self, *, error: str) -> int: + """Fail-fast bookkeeping for runs orphaned by a process crash. + + Agent runs execute in-process, so any active row found at scheduler + startup belongs to a run whose process is gone. Only valid under the + single-scheduler-instance assumption. + """ + stmt = select(ScheduledTaskRunRow).where(ScheduledTaskRunRow.status.in_(_ACTIVE_STATUS_VALUES)) + now = datetime.now(UTC) + async with self._sf() as session: + result = await session.execute(stmt) + rows = list(result.scalars()) + for row in rows: + row.status = str(RunStatus.INTERRUPTED) + row.error = error + row.finished_at = now + await session.commit() + return len(rows) diff --git a/backend/app/infra/persistence/scheduled_tasks.py b/backend/app/infra/persistence/scheduled_tasks.py new file mode 100644 index 000000000..b4ac3a774 --- /dev/null +++ b/backend/app/infra/persistence/scheduled_tasks.py @@ -0,0 +1,281 @@ +"""SQL adapter for the schedule context's task repository. + +Secondary adapter implementing `ScheduledTaskRepository`. SQL/ORM vocabulary +stops at this file: methods exchange domain objects and normalize SQLite's +tz-naive reads. + +**The queries are migrated unchanged from the legacy repository.** The claim +statement's `FOR UPDATE SKIP LOCKED` and the `protect_terminal` conditional +write are the module's concurrency contract, not style choices -- rewriting +them is how the invariants silently break. +""" + +from __future__ import annotations + +import logging +import socket +import uuid +from datetime import UTC, datetime, timedelta + +from sqlalchemy import and_, or_, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from app.infra.schedule.spec_mapping import spec_to_domain, spec_to_wire +from deerflow.domain.schedule.model import ( + TERMINAL_TASK_STATUSES, + ContextMode, + InvalidScheduleError, + ScheduledTask, + TaskStatus, +) +from deerflow.domain.schedule.ports import ScheduledTaskRepository + +# Transitional: the ORM row stays in the harness until engine, models, and +# migrations move into app/infra together. +from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow + +logger = logging.getLogger(__name__) + + +def _tz_aware(value: datetime | None) -> datetime | None: + """SQLite drops tzinfo on read; stored values are always UTC.""" + return value if value is None or value.tzinfo is not None else value.replace(tzinfo=UTC) + + +class SqlScheduledTaskRepository(ScheduledTaskRepository): + """SQL implementation of the `ScheduledTaskRepository` port. + + Explicit inheritance is a readability aid only: a missing method would + still instantiate fine (Protocol bodies are inherited), so the contract + test suite must cover every port method. + """ + + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._sf = session_factory + # Diagnostics only. The domain deliberately does not carry this -- who + # claimed a task is an identity, not a rule, and nothing reads it back. + self._lease_owner = f"{socket.gethostname()}:{uuid.uuid4().hex}" + + # ------------------------------------------------------------ conversion + + @staticmethod + def _to_domain(row: ScheduledTaskRow) -> ScheduledTask: + """ORM row -> aggregate. + + Raises InvalidScheduleError for a row whose stored schedule can no + longer be parsed. That is a **new** failure mode: the legacy repository + returned bare dicts and a corrupt spec only surfaced at dispatch time. + Single-row reads let it propagate; list reads skip and log, so one bad + row cannot 500 an entire listing. + """ + return ScheduledTask( + task_id=row.id, + user_id=row.user_id, + title=row.title, + prompt=row.prompt, + schedule=spec_to_domain(row.schedule_type, row.schedule_spec, row.timezone), + context_mode=ContextMode(row.context_mode), + thread_id=row.thread_id, + assistant_id=row.assistant_id, + status=TaskStatus(row.status), + overlap_policy=row.overlap_policy, + next_run_at=_tz_aware(row.next_run_at), + last_run_at=_tz_aware(row.last_run_at), + last_run_id=row.last_run_id, + last_thread_id=row.last_thread_id, + last_error=row.last_error, + run_count=row.run_count, + created_at=_tz_aware(row.created_at), + updated_at=_tz_aware(row.updated_at), + ) + + @classmethod + def _to_domain_or_skip(cls, row: ScheduledTaskRow) -> ScheduledTask | None: + try: + return cls._to_domain(row) + except InvalidScheduleError: + logger.warning("Skipping scheduled task %s: stored schedule is unparseable", row.id, exc_info=True) + return None + + @staticmethod + def _apply(row: ScheduledTaskRow, task: ScheduledTask) -> None: + """Aggregate -> ORM row. Explicit field list: a new column stays + invisible here until it is deliberately mapped.""" + row.user_id = task.user_id + row.title = task.title + row.prompt = task.prompt + row.schedule_type = str(task.schedule.schedule_type) + row.schedule_spec = spec_to_wire(task.schedule) + row.timezone = task.schedule.timezone + row.context_mode = str(task.context_mode) + row.thread_id = task.thread_id + row.assistant_id = task.assistant_id + row.status = str(task.status) + row.overlap_policy = task.overlap_policy + row.next_run_at = task.next_run_at + row.last_run_at = task.last_run_at + row.last_run_id = task.last_run_id + row.last_thread_id = task.last_thread_id + row.last_error = task.last_error + row.run_count = task.run_count + + # ------------------------------------------------------------------ port + + async def add(self, task: ScheduledTask) -> ScheduledTask: + now = datetime.now(UTC) + row = ScheduledTaskRow(id=task.task_id, created_at=now, updated_at=now) + self._apply(row, task) + async with self._sf() as session: + session.add(row) + await session.commit() + await session.refresh(row) + return self._to_domain(row) + + async def get(self, task_id: str, *, user_id: str) -> ScheduledTask | None: + async with self._sf() as session: + row = await session.get(ScheduledTaskRow, task_id) + if row is None or row.user_id != user_id: + return None + return self._to_domain(row) + + async def list_by_user(self, user_id: str) -> list[ScheduledTask]: + stmt = select(ScheduledTaskRow).where(ScheduledTaskRow.user_id == user_id).order_by(ScheduledTaskRow.created_at.desc(), ScheduledTaskRow.id.desc()) + async with self._sf() as session: + result = await session.execute(stmt) + return [task for task in (self._to_domain_or_skip(row) for row in result.scalars()) if task is not None] + + async def list_by_user_and_thread(self, user_id: str, thread_id: str) -> list[ScheduledTask]: + stmt = ( + select(ScheduledTaskRow) + .where( + ScheduledTaskRow.user_id == user_id, + ScheduledTaskRow.thread_id == thread_id, + ) + .order_by(ScheduledTaskRow.created_at.desc(), ScheduledTaskRow.id.desc()) + ) + async with self._sf() as session: + result = await session.execute(stmt) + return [task for task in (self._to_domain_or_skip(row) for row in result.scalars()) if task is not None] + + async def save(self, task: ScheduledTask) -> ScheduledTask | None: + async with self._sf() as session: + row = await session.get(ScheduledTaskRow, task.task_id) + if row is None or row.user_id != task.user_id: + return None + self._apply(row, task) + row.updated_at = datetime.now(UTC) + await session.commit() + await session.refresh(row) + return self._to_domain(row) + + async def delete(self, task_id: str, *, user_id: str) -> bool: + async with self._sf() as session: + row = await session.get(ScheduledTaskRow, task_id) + if row is None or row.user_id != user_id: + return False + await session.delete(row) + await session.commit() + return True + + async def claim_due(self, *, now: datetime, lease_seconds: int, limit: int) -> list[ScheduledTask]: + lease_expires_at = now + timedelta(seconds=lease_seconds) + stmt = ( + select(ScheduledTaskRow) + .where( + ScheduledTaskRow.next_run_at.is_not(None), + ScheduledTaskRow.next_run_at <= now, + or_( + and_( + ScheduledTaskRow.status == "enabled", + or_( + ScheduledTaskRow.lease_expires_at.is_(None), + ScheduledTaskRow.lease_expires_at < now, + ), + ), + # A task stuck in "running" with an expired lease means the + # claiming process died between claim and dispatch; it must + # stay reclaimable or the task is dead forever. + and_( + ScheduledTaskRow.status == "running", + ScheduledTaskRow.lease_expires_at.is_not(None), + ScheduledTaskRow.lease_expires_at < now, + ), + ), + ) + .order_by(ScheduledTaskRow.next_run_at.asc(), ScheduledTaskRow.id.asc()) + .limit(limit) + .with_for_update(skip_locked=True) + ) + async with self._sf() as session: + result = await session.execute(stmt) + rows = list(result.scalars()) + for row in rows: + row.lease_owner = self._lease_owner + row.lease_expires_at = lease_expires_at + row.status = "running" + row.updated_at = datetime.now(UTC) + await session.commit() + return [task for task in (self._to_domain_or_skip(row) for row in rows) if task is not None] + + async def record_launch( + self, + task_id: str, + *, + status: TaskStatus, + next_run_at: datetime | None, + last_run_at: datetime | None, + last_run_id: str | None, + last_thread_id: str | None, + last_error: str | None, + increment_run_count: bool, + protect_terminal: bool = False, + ) -> None: + async with self._sf() as session: + row = await session.get(ScheduledTaskRow, task_id) + if row is None: + return + if protect_terminal and TaskStatus(row.status) in TERMINAL_TASK_STATUSES: + # A fast-failing run can reach the completion hook (which + # finalizes a `once` task) before this launch-path write + # commits; keep the hook's status/error and only record the + # launch bookkeeping. + pass + else: + row.status = str(status) + row.last_error = last_error + row.next_run_at = next_run_at + row.last_run_at = last_run_at + row.last_run_id = last_run_id + row.last_thread_id = last_thread_id + if increment_run_count: + row.run_count += 1 + row.lease_owner = None + row.lease_expires_at = None + row.updated_at = datetime.now(UTC) + await session.commit() + + async def cancel_stuck_once_tasks(self, *, error: str) -> int: + """Reconcile `once` tasks orphaned in `running` by a process crash. + + A launched `once` task stays `running` until the in-process completion + hook moves it to a terminal status; its lease was cleared at launch, so + the claim query's expired-lease branch never sees it. After a crash the + hook is gone and the task would be stuck forever. Tasks still holding a + lease are left alone -- they were claimed but not launched, and + expired-lease reclaim recovers them safely. + """ + stmt = select(ScheduledTaskRow).where( + ScheduledTaskRow.schedule_type == "once", + ScheduledTaskRow.status == "running", + ScheduledTaskRow.lease_expires_at.is_(None), + ) + async with self._sf() as session: + result = await session.execute(stmt) + rows = list(result.scalars()) + now = datetime.now(UTC) + for row in rows: + row.status = "cancelled" + row.last_error = error + row.updated_at = now + await session.commit() + return len(rows) diff --git a/backend/app/infra/schedule/__init__.py b/backend/app/infra/schedule/__init__.py new file mode 100644 index 000000000..71da9cbdc --- /dev/null +++ b/backend/app/infra/schedule/__init__.py @@ -0,0 +1 @@ +"""Secondary adapters for the schedule bounded context.""" diff --git a/backend/app/infra/schedule/spec_mapping.py b/backend/app/infra/schedule/spec_mapping.py new file mode 100644 index 000000000..36d83d231 --- /dev/null +++ b/backend/app/infra/schedule/spec_mapping.py @@ -0,0 +1,68 @@ +"""Wire/storage <-> ScheduleSpec mapping. + +`schedule_spec` is both an HTTP request field and a JSON column -- one shape, +two boundaries -- so the mapping lives here once and both the router and the +SQL adapter import it, rather than the domain growing a `Mapping[str, Any]` in +its signatures. + +The split is deliberate: **structural** checks (is the key present? is it a +str?) belong to this boundary, **value** rules (5-field cron, resolvable +timezone, run_at present) belong to `ScheduleSpec.__post_init__`. That is why +this module can look thin -- most of what could go wrong is caught one layer +in, and reported with the same domain error. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime +from typing import Any + +from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType + + +def spec_to_domain(schedule_type: str, spec: Mapping[str, Any] | None, timezone: str) -> ScheduleSpec: + """Parse the stored/submitted triple into the value object. + + Raises: + InvalidScheduleError: unknown schedule type, or the type's required key + is missing or not a string. Raising a *domain* error from an + adapter is intentional -- domain errors are the vocabulary the + outer ring uses to say "this violates a domain rule", and the + router maps this one family uniformly. + """ + try: + kind = ScheduleType(schedule_type) + except ValueError as exc: + raise InvalidScheduleError(f"Unsupported schedule_type: {schedule_type}") from exc + + fields = spec or {} + if kind is ScheduleType.CRON: + raw_cron = fields.get("cron") + if not isinstance(raw_cron, str): + raise InvalidScheduleError("cron schedule requires schedule_spec.cron") + return ScheduleSpec.cron_schedule(raw_cron, timezone) + + raw_run_at = fields.get("run_at") + if not isinstance(raw_run_at, str): + raise InvalidScheduleError("once schedule requires run_at") + try: + run_at = datetime.fromisoformat(raw_run_at) + except ValueError as exc: + raise InvalidScheduleError(f"once schedule has an unparseable run_at: {raw_run_at!r}") from exc + return ScheduleSpec.once_at(run_at, timezone) + + +def spec_to_wire(spec: ScheduleSpec) -> dict[str, str]: + """Rebuild the persisted/wire JSON shape. + + Note this normalizes the stored string rather than echoing the caller's + bytes: the frontend submits an already-UTC-aware ISO value + (`zonedLocalToUtcIso`), so a trailing-Z input round-trips out as "+00:00". + Both forms parse on either side, so the normalization is deliberate -- + preferable to carrying the raw dict on the value object just to preserve + the exact input spelling. + """ + if spec.schedule_type is ScheduleType.CRON: + return {"cron": spec.cron or ""} + return {"run_at": spec.run_at.isoformat() if spec.run_at else ""} diff --git a/backend/tests/test_schedule_fakes.py b/backend/tests/test_schedule_fakes.py index 21c4d1d17..8f632900f 100644 --- a/backend/tests/test_schedule_fakes.py +++ b/backend/tests/test_schedule_fakes.py @@ -1,21 +1,26 @@ -"""Semantics of the in-memory schedule doubles. +"""Contract suite for the schedule repository ports. -These are the rules ``docs`` calls the port contract: what `claim_due` selects, -when `add` refuses, what `protect_terminal` preserves. They are asserted here -against the fakes so the doubles cannot drift from the ports they stand in for -while the service tests lean on them. +Every case here runs **twice**: once against the in-memory doubles and once +against the SQL adapters on a real file-backed sqlite database. That is the +point -- a rule stated in a port docstring has to hold for both, and a +divergence becomes a failure rather than a surprise in production. -This file is the seed of the contract suite: when the SQL adapters land, the -same cases run against both implementations and any divergence between them -becomes a failure rather than a surprise in production. Concurrency stays out -of scope in both tiers -- see the note in ``schedule_fakes``. +What the contract owns is single-threaded semantics: which rows `claim_due` +selects, that `add` refuses a second active record, what `protect_terminal` +preserves. What it deliberately does **not** own is atomicity -- the doubles +provide none, and a green run here says nothing about two dispatchers racing. +That is covered against a real database in +``test_scheduled_task_dispatch_race.py``. Do not read a passing contract suite +as licence to run more than one scheduler. """ from __future__ import annotations +from collections.abc import AsyncIterator from datetime import UTC, datetime, timedelta import pytest +import pytest_asyncio from schedule_fakes import ( FakeRunLauncher, FakeThreadLookup, @@ -23,7 +28,11 @@ from schedule_fakes import ( InMemoryScheduledTaskRepository, ) +from app.infra.persistence.scheduled_task_runs import SqlScheduledRunRepository +from app.infra.persistence.scheduled_tasks import SqlScheduledTaskRepository +from deerflow.config.database_config import DatabaseConfig from deerflow.domain.schedule.model import ( + ACTIVE_RUN_STATUSES, ActiveRunConflictError, ContextMode, RunStatus, @@ -34,6 +43,7 @@ from deerflow.domain.schedule.model import ( TriggerKind, ) from deerflow.domain.schedule.ports import ScheduledRunRepository, ScheduledTaskRepository +from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config pytestmark = pytest.mark.asyncio @@ -42,6 +52,67 @@ CRON = ScheduleSpec.cron_schedule("0 9 * * *", "UTC") ONCE = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "UTC") +@pytest_asyncio.fixture(params=["memory", "sql"]) +async def repos(request, tmp_path) -> AsyncIterator[tuple[ScheduledTaskRepository, ScheduledRunRepository]]: + """One parametrized fixture, two implementations of the same ports.""" + if request.param == "memory": + yield InMemoryScheduledTaskRepository(), InMemoryScheduledRunRepository() + return + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + try: + sf = get_session_factory() + assert sf is not None + yield SqlScheduledTaskRepository(sf), SqlScheduledRunRepository(sf) + finally: + await close_engine() + + +@pytest_asyncio.fixture +async def tasks(repos) -> ScheduledTaskRepository: + return repos[0] + + +@pytest_asyncio.fixture +async def runs(repos) -> ScheduledRunRepository: + return repos[1] + + +async def seed( + repo: ScheduledTaskRepository, + task: ScheduledTask, + *, + claimed_until: datetime | None = None, +) -> ScheduledTask: + """Install a task, optionally already carrying a claim. + + A claim cannot be installed through the port -- `claim_due` only stamps + tasks that are actually due, and these cases need shapes that claiming + would never produce (running with an *expired* claim, running with a live + one). So each implementation is set up directly: the fake exposes a seed + helper, and the SQL side writes the row. That is the one place this suite + reaches past the port, and it is why the assertions that follow go back + through it. + """ + stored = await repo.add(task) + if claimed_until is None: + return stored + + if isinstance(repo, InMemoryScheduledTaskRepository): + repo.seed(task, lease_owner="prior-worker", lease_expires_at=claimed_until) + return task + + from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow + + factory = get_session_factory() + assert factory is not None + async with factory() as session: + row = await session.get(ScheduledTaskRow, task.task_id) + row.lease_owner = "prior-worker" + row.lease_expires_at = claimed_until + await session.commit() + return task + + def make_task( task_id: str = "task-1", *, @@ -67,60 +138,107 @@ def make_task( class TestProtocolConformance: """`runtime_checkable` only checks that the methods exist, not their - signatures -- enough to catch a rename that updates one side only. + signatures -- enough to catch a rename that updates one side only.""" - Declared async purely to match this module's asyncio mark; there is - nothing to await. - """ - - async def test_task_repository_satisfies_the_port(self): - assert isinstance(InMemoryScheduledTaskRepository(), ScheduledTaskRepository) - - async def test_run_repository_satisfies_the_port(self): - assert isinstance(InMemoryScheduledRunRepository(), ScheduledRunRepository) + async def test_repositories_satisfy_their_ports(self, tasks, runs): + assert isinstance(tasks, ScheduledTaskRepository) + assert isinstance(runs, ScheduledRunRepository) class TestTaskOwnershipIsolation: """Another user's task must read as absent, never as forbidden.""" - async def test_get_hides_another_users_task(self): - repo = InMemoryScheduledTaskRepository() - await repo.add(make_task(user_id="owner")) - assert await repo.get("task-1", user_id="owner") is not None - assert await repo.get("task-1", user_id="intruder") is None + async def test_get_hides_another_users_task(self, tasks): + await tasks.add(make_task(user_id="owner")) + assert await tasks.get("task-1", user_id="owner") is not None + assert await tasks.get("task-1", user_id="intruder") is None - async def test_save_refuses_another_users_task(self): - repo = InMemoryScheduledTaskRepository() - await repo.add(make_task(user_id="owner")) - assert await repo.save(make_task(user_id="intruder")) is None - assert (await repo.get("task-1", user_id="owner")).user_id == "owner" + async def test_save_refuses_another_users_task(self, tasks): + await tasks.add(make_task(user_id="owner")) + assert await tasks.save(make_task(user_id="intruder")) is None + stored = await tasks.get("task-1", user_id="owner") + assert stored.user_id == "owner" - async def test_delete_refuses_another_users_task(self): - repo = InMemoryScheduledTaskRepository() - await repo.add(make_task(user_id="owner")) - assert await repo.delete("task-1", user_id="intruder") is False - assert await repo.get("task-1", user_id="owner") is not None + async def test_delete_refuses_another_users_task(self, tasks): + await tasks.add(make_task(user_id="owner")) + assert await tasks.delete("task-1", user_id="intruder") is False + assert await tasks.get("task-1", user_id="owner") is not None - async def test_list_by_thread_only_matches_bound_tasks(self): - repo = InMemoryScheduledTaskRepository() - await repo.add(make_task("bound", context_mode=ContextMode.REUSE_THREAD, thread_id="thread-1")) - await repo.add(make_task("unbound")) - listed = await repo.list_by_user_and_thread("user-1", "thread-1") + async def test_list_by_user_excludes_other_owners(self, tasks): + await tasks.add(make_task("mine", user_id="owner")) + await tasks.add(make_task("theirs", user_id="someone-else")) + assert [t.task_id for t in await tasks.list_by_user("owner")] == ["mine"] + + async def test_list_by_thread_only_matches_bound_tasks(self, tasks): + await tasks.add(make_task("bound", context_mode=ContextMode.REUSE_THREAD, thread_id="thread-1")) + await tasks.add(make_task("unbound")) + listed = await tasks.list_by_user_and_thread("user-1", "thread-1") assert [task.task_id for task in listed] == ["bound"] -class TestClaimDue: - async def test_claims_an_enabled_due_task_and_stamps_the_lease(self): - repo = InMemoryScheduledTaskRepository() - await repo.add(make_task(next_run_at=NOW - timedelta(minutes=1))) +class TestRoundTrip: + """What goes in comes back out -- including the value object, which the + SQL side has to rebuild from three separate columns.""" - claimed = await repo.claim_due(now=NOW, lease_seconds=120, limit=10) + async def test_a_cron_task_round_trips(self, tasks): + original = await tasks.add(make_task(schedule=ScheduleSpec.cron_schedule("*/5 * * * *", "Asia/Shanghai"))) + stored = await tasks.get(original.task_id, user_id="user-1") + assert stored.schedule == original.schedule + assert stored.schedule.cron == "*/5 * * * *" + assert stored.schedule.timezone == "Asia/Shanghai" + + async def test_a_once_task_round_trips_to_the_same_instant(self, tasks): + original = await tasks.add(make_task(schedule=ONCE)) + stored = await tasks.get(original.task_id, user_id="user-1") + assert stored.schedule == original.schedule + + async def test_timestamps_come_back_timezone_aware(self, tasks): + """SQLite drops tzinfo on read; a naive datetime downstream would + compare wrong against an aware `now`.""" + await tasks.add(make_task(next_run_at=NOW)) + stored = await tasks.get("task-1", user_id="user-1") + assert stored.next_run_at.tzinfo is not None + assert stored.created_at.tzinfo is not None + + async def test_save_replaces_the_whole_aggregate(self, tasks): + from dataclasses import replace + + await tasks.add(make_task()) + stored = await tasks.get("task-1", user_id="user-1") + + await tasks.save(replace(stored, title="renamed", status=TaskStatus.PAUSED, run_count=7)) + + reloaded = await tasks.get("task-1", user_id="user-1") + assert reloaded.title == "renamed" + assert reloaded.status is TaskStatus.PAUSED + assert reloaded.run_count == 7 + + +class TestClaimDue: + async def test_claims_an_enabled_due_task(self, tasks): + await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1))) + + claimed = await tasks.claim_due(now=NOW, lease_seconds=120, limit=10) assert [task.task_id for task in claimed] == ["task-1"] assert claimed[0].status is TaskStatus.RUNNING - owner, expires = repo.lease_of("task-1") - assert owner is not None, "an implementation may record who claimed; nothing reads it back" - assert expires == NOW + timedelta(seconds=120) + + async def test_a_claimed_task_is_not_claimed_again(self, tasks): + """The lease is not readable through the port, so the rule is asserted + the way it actually matters: a second claimer comes back empty.""" + await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1))) + assert len(await tasks.claim_due(now=NOW, lease_seconds=120, limit=10)) == 1 + + assert await tasks.claim_due(now=NOW, lease_seconds=120, limit=10) == [] + + async def test_the_claim_expires(self, tasks): + await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1))) + await tasks.claim_due(now=NOW, lease_seconds=60, limit=10) + + later = NOW + timedelta(seconds=61) + reclaimed = await tasks.claim_due(now=later, lease_seconds=60, limit=10) + + assert [task.task_id for task in reclaimed] == ["task-1"] @pytest.mark.parametrize( ("label", "task_kwargs"), @@ -131,54 +249,41 @@ class TestClaimDue: ("completed", {"status": TaskStatus.COMPLETED, "next_run_at": NOW - timedelta(minutes=1)}), ], ) - async def test_does_not_claim(self, label, task_kwargs): - repo = InMemoryScheduledTaskRepository() - await repo.add(make_task(**task_kwargs)) - assert await repo.claim_due(now=NOW, lease_seconds=120, limit=10) == [], label + async def test_does_not_claim(self, tasks, label, task_kwargs): + await tasks.add(make_task(**task_kwargs)) + assert await tasks.claim_due(now=NOW, lease_seconds=120, limit=10) == [], label - async def test_does_not_steal_a_live_claim(self): - repo = InMemoryScheduledTaskRepository() - repo.seed( - make_task(next_run_at=NOW - timedelta(minutes=1)), - lease_owner="worker-1", - lease_expires_at=NOW + timedelta(seconds=60), - ) - assert await repo.claim_due(now=NOW, lease_seconds=120, limit=10) == [] - - async def test_reclaims_a_task_stuck_mid_dispatch(self): + async def test_reclaims_a_task_stuck_mid_dispatch(self, tasks): """The claimer died between claiming and launching: status is running, - the lease has expired, and the task must not stay unreachable.""" - repo = InMemoryScheduledTaskRepository() - repo.seed( + the claim has expired, and the task must not stay unreachable.""" + await seed( + tasks, make_task(status=TaskStatus.RUNNING, next_run_at=NOW - timedelta(minutes=1)), - lease_owner="dead-worker", - lease_expires_at=NOW - timedelta(seconds=1), + claimed_until=NOW - timedelta(seconds=1), ) - claimed = await repo.claim_due(now=NOW, lease_seconds=120, limit=10) + claimed = await tasks.claim_due(now=NOW, lease_seconds=120, limit=10) assert [task.task_id for task in claimed] == ["task-1"] - assert repo.lease_of("task-1")[1] == NOW + timedelta(seconds=120), "the claim was re-stamped" - async def test_claims_the_most_overdue_first_and_honours_the_limit(self): - repo = InMemoryScheduledTaskRepository() - await repo.add(make_task("late", next_run_at=NOW - timedelta(hours=2))) - await repo.add(make_task("later", next_run_at=NOW - timedelta(hours=1))) + async def test_claims_the_most_overdue_first_and_honours_the_limit(self, tasks): + await tasks.add(make_task("late", next_run_at=NOW - timedelta(hours=2))) + await tasks.add(make_task("later", next_run_at=NOW - timedelta(hours=1))) - claimed = await repo.claim_due(now=NOW, lease_seconds=120, limit=1) + claimed = await tasks.claim_due(now=NOW, lease_seconds=120, limit=1) assert [task.task_id for task in claimed] == ["late"] class TestRecordLaunch: - async def test_writes_bookkeeping_and_releases_the_claim(self): - repo = InMemoryScheduledTaskRepository() - repo.seed(make_task(next_run_at=NOW), lease_owner="w", lease_expires_at=NOW + timedelta(seconds=60)) + async def test_writes_bookkeeping_and_frees_the_task_for_the_next_round(self, tasks): + await tasks.add(make_task(next_run_at=NOW - timedelta(minutes=1))) + await tasks.claim_due(now=NOW, lease_seconds=120, limit=10) - await repo.record_launch( + await tasks.record_launch( "task-1", status=TaskStatus.ENABLED, - next_run_at=NOW + timedelta(days=1), + next_run_at=NOW - timedelta(seconds=1), last_run_at=NOW, last_run_id="run-1", last_thread_id="thread-1", @@ -186,19 +291,19 @@ class TestRecordLaunch: increment_run_count=True, ) - task = await repo.get("task-1", user_id="user-1") + task = await tasks.get("task-1", user_id="user-1") assert task.status is TaskStatus.ENABLED assert task.last_run_id == "run-1" assert task.run_count == 1 - assert repo.lease_of("task-1") == (None, None) + # The claim was released, so the next round can take it again. + assert len(await tasks.claim_due(now=NOW, lease_seconds=120, limit=10)) == 1 - async def test_protect_terminal_keeps_a_concurrently_finalized_verdict(self): + async def test_protect_terminal_keeps_a_concurrently_finalized_verdict(self, tasks): """A fast-failing run's completion hook lands before the launch path's own write; the completion is authoritative.""" - repo = InMemoryScheduledTaskRepository() - await repo.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE)) + await tasks.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE)) - await repo.record_launch( + await tasks.record_launch( "task-1", status=TaskStatus.RUNNING, next_run_at=None, @@ -210,17 +315,16 @@ class TestRecordLaunch: protect_terminal=True, ) - task = await repo.get("task-1", user_id="user-1") + task = await tasks.get("task-1", user_id="user-1") assert task.status is TaskStatus.COMPLETED, "the terminal status must survive" assert task.last_error is None, "the terminal error must survive" assert task.last_run_id == "run-1", "bookkeeping is still recorded" assert task.run_count == 1 - async def test_without_protect_terminal_the_write_wins(self): - repo = InMemoryScheduledTaskRepository() - await repo.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE)) + async def test_without_protect_terminal_the_write_wins(self, tasks): + await tasks.add(make_task(status=TaskStatus.COMPLETED, schedule=ONCE)) - await repo.record_launch( + await tasks.record_launch( "task-1", status=TaskStatus.FAILED, next_run_at=None, @@ -231,13 +335,12 @@ class TestRecordLaunch: increment_run_count=False, ) - task = await repo.get("task-1", user_id="user-1") + task = await tasks.get("task-1", user_id="user-1") assert task.status is TaskStatus.FAILED assert task.last_error == "boom" - async def test_unknown_task_is_ignored(self): - repo = InMemoryScheduledTaskRepository() - await repo.record_launch( + async def test_unknown_task_is_ignored(self, tasks): + await tasks.record_launch( "nope", status=TaskStatus.ENABLED, next_run_at=None, @@ -250,30 +353,29 @@ class TestRecordLaunch: class TestCancelStuckOnceTasks: - async def test_cancels_a_launched_once_task_with_no_claim(self): - repo = InMemoryScheduledTaskRepository() - repo.seed(make_task(status=TaskStatus.RUNNING, schedule=ONCE)) + async def test_cancels_a_launched_once_task_with_no_claim(self, tasks): + """Launched, so the claim was released; the completion hook then died + with the process. Expired-claim reclaim can never see this one.""" + await tasks.add(make_task(status=TaskStatus.RUNNING, schedule=ONCE)) - assert await repo.cancel_stuck_once_tasks(error="restarted") == 1 - task = await repo.get("task-1", user_id="user-1") + assert await tasks.cancel_stuck_once_tasks(error="restarted") == 1 + task = await tasks.get("task-1", user_id="user-1") assert task.status is TaskStatus.CANCELLED assert task.last_error == "restarted" - async def test_leaves_a_claimed_task_to_lease_expiry(self): + async def test_leaves_a_claimed_task_to_claim_expiry(self, tasks): """Claimed but not launched -- expired-claim reclaim recovers it, and cancelling here would throw away a dispatch that never happened.""" - repo = InMemoryScheduledTaskRepository() - repo.seed( - make_task(status=TaskStatus.RUNNING, schedule=ONCE), - lease_owner="w", - lease_expires_at=NOW + timedelta(seconds=60), + await seed( + tasks, + make_task(status=TaskStatus.RUNNING, schedule=ONCE, next_run_at=NOW - timedelta(minutes=1)), + claimed_until=NOW + timedelta(seconds=60), ) - assert await repo.cancel_stuck_once_tasks(error="restarted") == 0 + assert await tasks.cancel_stuck_once_tasks(error="restarted") == 0 - async def test_leaves_cron_tasks_alone(self): - repo = InMemoryScheduledTaskRepository() - repo.seed(make_task(status=TaskStatus.RUNNING, schedule=CRON)) - assert await repo.cancel_stuck_once_tasks(error="restarted") == 0 + async def test_leaves_cron_tasks_alone(self, tasks): + await tasks.add(make_task(status=TaskStatus.RUNNING, schedule=CRON)) + assert await tasks.cancel_stuck_once_tasks(error="restarted") == 0 class TestActiveSlot: @@ -283,75 +385,83 @@ class TestActiveSlot: def _tombstone(self, task_id: str = "task-1") -> ScheduledRun: return ScheduledRun.skipped_tombstone(task_id=task_id, thread_id="thread-1", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED) - async def test_second_active_record_is_refused(self): - repo = InMemoryScheduledRunRepository() - await repo.add(self._queued()) + async def test_second_active_record_is_refused(self, runs): + await runs.add(self._queued()) with pytest.raises(ActiveRunConflictError): - await repo.add(self._queued()) + await runs.add(self._queued()) - async def test_a_tombstone_never_conflicts(self): + async def test_a_tombstone_never_conflicts(self, runs): """Terminal from birth, so it sits outside the active-slot rule -- this is why the skip path cannot reuse the queued factory.""" - repo = InMemoryScheduledRunRepository() - await repo.add(self._queued()) - await repo.add(self._tombstone()) - assert await repo.count_active() == 1 + await runs.add(self._queued()) + await runs.add(self._tombstone()) + assert await runs.count_active() == 1 - async def test_another_task_is_unaffected(self): - repo = InMemoryScheduledRunRepository() - await repo.add(self._queued("task-1")) - await repo.add(self._queued("task-2")) - assert await repo.count_active() == 2 + async def test_another_task_is_unaffected(self, runs): + await runs.add(self._queued("task-1")) + await runs.add(self._queued("task-2")) + assert await runs.count_active() == 2 - async def test_slot_frees_up_once_the_record_terminalizes(self): - repo = InMemoryScheduledRunRepository() - first = await repo.add(self._queued()) - await repo.update_status(first.record_id, status=RunStatus.SUCCESS, finished_at=NOW) - await repo.add(self._queued()) - assert await repo.count_active() == 1 + async def test_slot_frees_up_once_the_record_terminalizes(self, runs): + first = await runs.add(self._queued()) + await runs.update_status(first.record_id, status=RunStatus.SUCCESS, finished_at=NOW) + await runs.add(self._queued()) + assert await runs.count_active() == 1 - async def test_has_active_is_scoped_to_one_task_while_count_is_global(self): - repo = InMemoryScheduledRunRepository() - await repo.add(self._queued("task-1")) - await repo.add(self._queued("task-2")) - assert await repo.has_active("task-1") is True - assert await repo.has_active("task-3") is False - assert await repo.count_active() == 2 + async def test_has_active_is_scoped_to_one_task_while_count_is_global(self, runs): + await runs.add(self._queued("task-1")) + await runs.add(self._queued("task-2")) + assert await runs.has_active("task-1") is True + assert await runs.has_active("task-3") is False + assert await runs.count_active() == 2 + + async def test_a_run_round_trips(self, runs): + stored = await runs.add(self._queued()) + listed = await runs.list_by_task("task-1", limit=10, offset=0) + assert [r.record_id for r in listed] == [stored.record_id] + assert listed[0].trigger is TriggerKind.SCHEDULED + assert listed[0].scheduled_for.tzinfo is not None class TestRunStatusWrites: - async def test_protect_terminal_backfills_without_overwriting(self): - repo = InMemoryScheduledRunRepository() - run = await repo.add(ScheduledRun.queued(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)) - await repo.update_status(run.record_id, status=RunStatus.FAILED, error="boom", finished_at=NOW) + async def _one_queued(self, runs) -> ScheduledRun: + return await runs.add(ScheduledRun.queued(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)) + + async def test_protect_terminal_backfills_without_overwriting(self, runs): + run = await self._one_queued(runs) + await runs.update_status(run.record_id, status=RunStatus.FAILED, error="boom", finished_at=NOW) # The launch path's write arrives late. - await repo.update_status(run.record_id, status=RunStatus.RUNNING, run_id="run-1", started_at=NOW, protect_terminal=True) + await runs.update_status(run.record_id, status=RunStatus.RUNNING, run_id="run-1", started_at=NOW, protect_terminal=True) - stored = repo.all_runs()[0] + stored = (await runs.list_by_task("t", limit=10, offset=0))[0] assert stored.status is RunStatus.FAILED assert stored.error == "boom" assert stored.run_id == "run-1", "the id the completion could not know is backfilled" assert stored.started_at == NOW - async def test_unknown_record_is_ignored(self): - repo = InMemoryScheduledRunRepository() - await repo.update_status("nope", status=RunStatus.SUCCESS) + async def test_unknown_record_is_ignored(self, runs): + await runs.update_status("nope", status=RunStatus.SUCCESS) - async def test_mark_stale_active_terminalizes_orphans(self): - repo = InMemoryScheduledRunRepository() - active = await repo.add(ScheduledRun.queued(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)) - done = await repo.add(ScheduledRun.skipped_tombstone(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)) + async def test_mark_stale_active_terminalizes_orphans(self, runs): + active = await self._one_queued(runs) + done = await runs.add(ScheduledRun.skipped_tombstone(task_id="t", thread_id="th", scheduled_for=NOW, trigger=TriggerKind.SCHEDULED)) - assert await repo.mark_stale_active(error="gateway restarted") == 1 + assert await runs.mark_stale_active(error="gateway restarted") == 1 - by_id = {run.record_id: run for run in repo.all_runs()} + by_id = {run.record_id: run for run in await runs.list_by_task("t", limit=10, offset=0)} assert by_id[active.record_id].status is RunStatus.INTERRUPTED assert by_id[active.record_id].error == "gateway restarted" assert by_id[done.record_id].status is RunStatus.SKIPPED + async def test_active_statuses_are_exactly_queued_and_running(self): + assert ACTIVE_RUN_STATUSES == (RunStatus.QUEUED, RunStatus.RUNNING) + class TestLauncherAndThreadLookup: + """Fake-only: these two ports have no SQL implementation -- one starts a + run and the other asks the thread store, so both land in later adapters.""" + async def test_launcher_records_the_call_and_echoes_the_thread(self): launcher = FakeRunLauncher() launched = await launcher.launch( diff --git a/backend/tests/test_schedule_spec_mapping.py b/backend/tests/test_schedule_spec_mapping.py new file mode 100644 index 000000000..b56514453 --- /dev/null +++ b/backend/tests/test_schedule_spec_mapping.py @@ -0,0 +1,78 @@ +"""Boundary tests for the schedule_spec mapping. + +This is where a `Mapping[str, Any]` is allowed to exist. The split it enforces +is the point: **structural** problems (key missing, wrong type, unknown +schedule type) are caught here, **value** problems (5-field cron, resolvable +timezone) are left to `ScheduleSpec.__post_init__` -- and both surface as the +same domain error, so the router maps one family. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +import pytest + +from app.infra.schedule.spec_mapping import spec_to_domain, spec_to_wire +from deerflow.domain.schedule.model import InvalidScheduleError, ScheduleSpec, ScheduleType + + +class TestStructuralChecks: + def test_unknown_schedule_type_is_rejected(self): + """The one rule the domain cannot state: ScheduleSpec only accepts the + enum, so a bad string has to be caught at the boundary.""" + with pytest.raises(InvalidScheduleError, match="Unsupported schedule_type"): + spec_to_domain("teleport", {"cron": "0 9 * * *"}, "UTC") + + @pytest.mark.parametrize("spec", [{}, None, {"cron": 5}, {"run_at": "..."}]) + def test_cron_without_a_string_cron_is_rejected(self, spec): + with pytest.raises(InvalidScheduleError, match="requires schedule_spec"): + spec_to_domain("cron", spec, "UTC") + + @pytest.mark.parametrize("spec", [{}, None, {"run_at": 5}, {"cron": "0 9 * * *"}]) + def test_once_without_a_string_run_at_is_rejected(self, spec): + with pytest.raises(InvalidScheduleError, match="requires run_at"): + spec_to_domain("once", spec, "UTC") + + def test_an_unparseable_run_at_is_rejected(self): + with pytest.raises(InvalidScheduleError, match="unparseable run_at"): + spec_to_domain("once", {"run_at": "next tuesday"}, "UTC") + + +class TestValueChecksStayInTheDomain: + """These are not re-implemented here -- they arrive from __post_init__, + which is why the boundary can stay thin.""" + + def test_a_bad_cron_expression_still_raises(self): + with pytest.raises(InvalidScheduleError, match="exactly 5 fields"): + spec_to_domain("cron", {"cron": "0 9 * *"}, "UTC") + + def test_a_bad_timezone_still_raises(self): + with pytest.raises(InvalidScheduleError, match="Unknown timezone"): + spec_to_domain("cron", {"cron": "0 9 * * *"}, "Mars/Olympus_Mons") + + +class TestRoundTrip: + def test_cron_round_trips_normalized(self): + spec = spec_to_domain("cron", {"cron": " 0 9 * * * "}, "Asia/Shanghai") + assert spec.schedule_type is ScheduleType.CRON + assert spec_to_wire(spec) == {"cron": "0 9 * * *"} + assert spec_to_domain("cron", spec_to_wire(spec), "Asia/Shanghai") == spec + + def test_once_round_trips_to_the_same_instant(self): + spec = spec_to_domain("once", {"run_at": "2026-08-01T09:00:00"}, "Asia/Shanghai") + assert spec.run_at == datetime(2026, 8, 1, 1, 0, tzinfo=UTC) + assert spec_to_domain("once", spec_to_wire(spec), "Asia/Shanghai") == spec + + def test_a_trailing_z_input_parses_and_re_emits_as_an_offset(self): + """The shape the frontend actually submits (`zonedLocalToUtcIso`). + Both spellings parse on either side, so normalizing is deliberate.""" + spec = spec_to_domain("once", {"run_at": "2026-08-01T01:00:00Z"}, "Asia/Shanghai") + emitted = spec_to_wire(spec)["run_at"] + assert emitted.endswith("+00:00") + assert spec_to_domain("once", {"run_at": emitted}, "Asia/Shanghai") == spec + + def test_wire_output_is_idempotent(self): + spec = ScheduleSpec.once_at(datetime(2026, 8, 1, 9, 0, tzinfo=UTC), "UTC") + once = spec_to_wire(spec) + assert spec_to_wire(spec_to_domain("once", once, "UTC")) == once