deer-flow/backend/tests/test_schedule_corrupt_rows.py

"""A corrupt stored schedule is an operator problem, not a client error.

``SqlScheduledTaskRepository._to_domain`` rebuilds the aggregate on every
read, so a row whose stored schedule no longer parses surfaces at read time.
It used to surface as ``InvalidScheduleError`` -- the same error the aggregate
raises for a *client-submitted* schedule, which the router maps to 422. That
double duty told the client "your request is wrong" about a request that was
perfectly fine, and made the row unrepairable over HTTP: PATCH reads the task
before writing, so the fix path 422'd too.

``CorruptStoredScheduleError`` splits the vocabulary: it is raised only by the
persistence adapter, and it is deliberately absent from the router's status
table so it falls through to the unclassified-500 branch -- a server-side
fault reported as one.

SQL-only: the in-memory double stores whole aggregates and cannot hold a
corrupt row, which is exactly why this file exists beside the contract suite
rather than inside it.
"""

from __future__ import annotations

from collections.abc import AsyncIterator
from datetime import UTC, datetime

import pytest
import pytest_asyncio

from app.adapters.schedule.scheduled_task_repository import SqlScheduledTaskRepository
from app.gateway.routers.schedule.router import _STATUS_BY_ERROR
from deerflow.config.database_config import DatabaseConfig
from deerflow.domain.schedule.exceptions import CorruptStoredScheduleError, InvalidScheduleError
from deerflow.domain.schedule.model import ScheduledTask, SchedulePolicy, ScheduleSpec
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow

pytestmark = pytest.mark.asyncio

NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC)


@pytest_asyncio.fixture
async def repo(tmp_path) -> AsyncIterator[SqlScheduledTaskRepository]:
    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)
    finally:
        await close_engine()


async def _add_task(repo: SqlScheduledTaskRepository, *, title: str = "healthy") -> ScheduledTask:
    return await repo.add(
        ScheduledTask.create(
            user_id="user-1",
            title=title,
            prompt="p",
            schedule=ScheduleSpec.cron_schedule("0 9 * * *", "UTC"),
            context_mode="fresh_thread_per_run",
            thread_id=None,
            now=NOW,
            policy=SchedulePolicy(),
        )
    )


async def _corrupt(task_id: str, **columns) -> None:
    """Damage a stored row directly -- the exact shape a bug or a manual edit
    would leave behind, unreachable through the port."""
    sf = get_session_factory()
    assert sf is not None
    async with sf() as session:
        row = await session.get(ScheduledTaskRow, task_id)
        assert row is not None
        for name, value in columns.items():
            setattr(row, name, value)
        await session.commit()


class TestSingleRowReads:
    async def test_a_corrupt_schedule_raises_the_dedicated_error(self, repo):
        task = await _add_task(repo)
        await _corrupt(task.task_id, schedule_spec={})

        with pytest.raises(CorruptStoredScheduleError):
            await repo.get(task.task_id, user_id="user-1")

    async def test_the_dedicated_error_is_not_the_client_facing_one(self, repo):
        """The router maps InvalidScheduleError to 422; a corrupt row must not
        ride that mapping."""
        task = await _add_task(repo)
        await _corrupt(task.task_id, schedule_spec={})

        with pytest.raises(CorruptStoredScheduleError) as exc_info:
            await repo.get(task.task_id, user_id="user-1")
        assert not isinstance(exc_info.value, InvalidScheduleError)

    async def test_a_corrupt_context_mode_is_reported_the_same_way(self, repo):
        """Enum rebuild failures are the same fault as spec parse failures: a
        raw ValueError out of the adapter would be a technical exception
        crossing the boundary untranslated."""
        task = await _add_task(repo)
        await _corrupt(task.task_id, context_mode="not-a-mode")

        with pytest.raises(CorruptStoredScheduleError):
            await repo.get(task.task_id, user_id="user-1")


class TestListReads:
    async def test_a_corrupt_row_does_not_take_down_the_listing(self, repo):
        healthy = await _add_task(repo, title="healthy")
        broken = await _add_task(repo, title="broken")
        await _corrupt(broken.task_id, schedule_spec={})

        listed = await repo.list_by_user("user-1")

        assert [t.task_id for t in listed] == [healthy.task_id]


class TestRouterMapping:
    def test_the_corrupt_row_error_is_unclassified_on_purpose(self):
        """Absent from the status table means the router's fallthrough turns
        it into a 500 -- a new protocol decision would have to add it here
        deliberately."""
        assert CorruptStoredScheduleError not in _STATUS_BY_ERROR