deerflow-code/offline-backend-20260512/backend/tests/test_curator_scheduler_startup.py

"""Tests that the curator scheduler does not block application startup.

Verifies:
1. _curator_scheduler_loop is launched as a non-blocking asyncio task.
2. The scheduler sleeps for the startup delay before doing any work.
3. maybe_run_curator is NOT called during the startup delay window.
4. When the lock is unavailable the loop exits immediately without calling the curator.

NOTE: ``import app.gateway.app as m`` resolves to the FastAPI *instance* because
``app/gateway/__init__.py`` re-exports ``app = FastAPI(...)``.  The actual module
object lives in ``sys.modules["app.gateway.app"]`` and is used here via a helper.
"""

from __future__ import annotations

import asyncio
import sys
from unittest.mock import MagicMock, patch


def _gateway_module():
    """Return the real app.gateway.app *module* (not the FastAPI instance)."""
    import importlib
    importlib.import_module("app.gateway.app")  # side-effect: populates sys.modules
    return sys.modules["app.gateway.app"]


_STARTUP_DELAY = 300   # matches _CURATOR_STARTUP_DELAY_SECONDS in app.gateway.app
_CHECK_INTERVAL = 3600  # matches _CURATOR_CHECK_INTERVAL_SECONDS


def test_scheduler_does_not_block_startup():
    """create_task must return immediately — startup must not await the loop body."""
    mod = _gateway_module()
    curator_called = False

    async def fake_sleep(_delay):
        pass

    async def fake_curator():
        nonlocal curator_called
        curator_called = True

    async def run():
        with (
            patch.object(mod, "_acquire_singleton_lock", return_value=MagicMock()),
            patch.object(mod, "_release_singleton_lock"),
            patch.object(mod, "_CURATOR_STARTUP_DELAY_SECONDS", _STARTUP_DELAY),
            patch.object(mod, "_CURATOR_CHECK_INTERVAL_SECONDS", _CHECK_INTERVAL),
            patch("deerflow.skills.curator.maybe_run_curator", fake_curator),
            patch("asyncio.sleep", fake_sleep),
        ):
            task = asyncio.create_task(mod._curator_scheduler_loop())
            # create_task must return immediately; the coroutine is still pending
            assert not task.done(), "task should be running in the background, not finished"
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
                pass

    asyncio.run(run())
    assert not curator_called, "curator must not fire before the startup delay completes"


def test_scheduler_sleeps_startup_delay_before_first_curator_call():
    """The very first asyncio.sleep must use the startup-delay value (300 s)."""
    mod = _gateway_module()
    sleep_calls: list[float] = []
    curator_called = False

    async def fake_sleep(delay):
        sleep_calls.append(delay)
        raise asyncio.CancelledError  # stop after recording first sleep

    async def fake_curator():
        nonlocal curator_called
        curator_called = True

    async def run():
        with (
            patch.object(mod, "_acquire_singleton_lock", return_value=MagicMock()),
            patch.object(mod, "_release_singleton_lock"),
            patch.object(mod, "_CURATOR_STARTUP_DELAY_SECONDS", _STARTUP_DELAY),
            patch.object(mod, "_CURATOR_CHECK_INTERVAL_SECONDS", _CHECK_INTERVAL),
            patch("deerflow.skills.curator.maybe_run_curator", fake_curator),
            patch("asyncio.sleep", fake_sleep),
        ):
            task = asyncio.create_task(mod._curator_scheduler_loop())
            try:
                await task
            except asyncio.CancelledError:
                pass

    asyncio.run(run())

    assert sleep_calls, "asyncio.sleep was never called"
    assert sleep_calls[0] == _STARTUP_DELAY, (
        f"first sleep must be the startup delay ({_STARTUP_DELAY}s), got {sleep_calls[0]}"
    )
    assert not curator_called, "curator must not be called before startup delay elapses"


def test_scheduler_actually_calls_curator_after_startup_delay():
    """After the startup delay elapses, maybe_run_curator must be called."""
    mod = _gateway_module()
    curator_call_count = 0

    async def fake_sleep(*_):
        pass  # instant — simulate time passing

    async def fake_curator():
        nonlocal curator_call_count
        curator_call_count += 1
        # After first call cancel the loop so the test terminates
        raise asyncio.CancelledError

    async def run():
        with (
            patch.object(mod, "_acquire_singleton_lock", return_value=MagicMock()),
            patch.object(mod, "_release_singleton_lock"),
            patch.object(mod, "_CURATOR_STARTUP_DELAY_SECONDS", _STARTUP_DELAY),
            patch.object(mod, "_CURATOR_CHECK_INTERVAL_SECONDS", _CHECK_INTERVAL),
            patch("deerflow.skills.curator.maybe_run_curator", fake_curator),
            patch("asyncio.sleep", fake_sleep),
        ):
            task = asyncio.create_task(mod._curator_scheduler_loop())
            try:
                await task
            except asyncio.CancelledError:
                pass

    asyncio.run(run())
    assert curator_call_count >= 1, (
        f"maybe_run_curator should have been called at least once, got {curator_call_count}"
    )


def test_scheduler_skips_when_lock_unavailable():
    """When another worker holds the lock the loop must exit without doing any work."""
    mod = _gateway_module()
    curator_called = False

    async def fake_curator():
        nonlocal curator_called
        curator_called = True

    async def run():
        with (
            patch.object(mod, "_acquire_singleton_lock", return_value=None),
            patch("deerflow.skills.curator.maybe_run_curator", fake_curator),
        ):
            await mod._curator_scheduler_loop()

    asyncio.run(run())
    assert not curator_called, "curator must not run when lock is not acquired"