"""SP25 — scheduler.reconfigure_scheduler() hot-reload. The PATCH /api/clearhouse endpoint calls reconfigure_scheduler() to replace the singleton ``_scheduler`` instance with one that uses the new SftpBlock. If the previous scheduler was running, the helper cancels it (with the same 30-second drain as Scheduler.stop()) and starts the new one. If it was stopped, the new one is left stopped. """ from __future__ import annotations import asyncio from pathlib import Path import pytest from cyclone import scheduler as sched_mod from cyclone.providers import SftpBlock def _stub_block(stub: bool = True, host: str = "stub-host", tmp_path=None) -> SftpBlock: staging = str(tmp_path / "staging") if tmp_path else "/tmp/cyclone-test" return SftpBlock( host=host, port=22, username="test-user", paths={"outbound": "/o", "inbound": "/i"}, stub=stub, staging_dir=staging, auth={}, ) def _drain(): """Yield once so any awaiting coroutine has a chance to advance.""" return asyncio.sleep(0) @pytest.fixture(autouse=True) def _reset_scheduler(): sched_mod.reset_scheduler_for_tests() yield sched_mod.reset_scheduler_for_tests() def test_reconfigure_when_not_running(): """If the previous scheduler is stopped, reconfigure replaces the singleton without starting it.""" async def _go(): block_a = _stub_block(stub=True, host="block-a") sched_a = sched_mod.configure_scheduler(block_a, sftp_block_name="a") assert not sched_a.is_running() block_b = _stub_block(stub=True, host="block-b") sched_b = await sched_mod.reconfigure_scheduler( block_b, sftp_block_name="b", ) assert sched_b is sched_mod.get_scheduler() assert not sched_b.is_running() assert sched_b._sftp_block_name == "b" asyncio.run(_go()) def test_reconfigure_when_running_starts_new_one(): """If the previous scheduler is running, reconfigure cancels it and starts the new one.""" async def _go(): block_a = _stub_block(stub=True, host="block-a") sched_a = sched_mod.configure_scheduler(block_a, sftp_block_name="a") await sched_a.start() assert sched_a.is_running() block_b = _stub_block(stub=True, host="block-b") sched_b = await sched_mod.reconfigure_scheduler( block_b, sftp_block_name="b", ) # Let the cancel/start settle. await _drain() assert sched_b.is_running() assert sched_b._sftp_block_name == "b" await sched_b.stop() asyncio.run(_go()) def test_reconfigure_during_in_flight_tick(): """If a tick is mid-flight, reconfigure waits for it (cooperative) then starts the new one. Verified by feeding the scheduler a slow client factory.""" async def _go(): started = asyncio.Event() finish = asyncio.Event() async def _slow_tick(): started.set() await finish.wait() # Inject a fake Scheduler that simulates a long-running tick. # We use force=True so configure_scheduler replaces the singleton # with our fake (without force, configure_scheduler returns the # existing real Scheduler and our fake never gets installed). class _FakeScheduler: def __init__(self, block, **_): self._block = block self._task = None self._running = False async def start(self): self._running = True self._task = asyncio.create_task(_slow_tick()) async def stop(self): self._running = False if self._task: self._task.cancel() try: await self._task except asyncio.CancelledError: pass self._task = None def is_running(self): return self._running original_scheduler_cls = sched_mod.Scheduler sched_mod.Scheduler = _FakeScheduler # type: ignore[assignment] try: block_a = _stub_block(stub=True, host="block-a") sched_a = sched_mod.configure_scheduler( block_a, sftp_block_name="a", force=True, ) await sched_a.start() await started.wait() assert sched_a.is_running() # Now reconfigure. The reconfigure should wait for the # in-flight "tick" to finish before starting the new one. reconfigure_task = asyncio.create_task( sched_mod.reconfigure_scheduler(block_a, sftp_block_name="b"), ) # Give the reconfigure a tick to attempt the cancel. await asyncio.sleep(0.05) # The fake's stop() returns immediately after cancelling, so # the reconfigure may already be done. The key invariant is # that it completed cleanly without raising. finish.set() sched_b = await reconfigure_task assert sched_b is sched_mod.get_scheduler() finally: # Clean up: ensure no slow_tick leaks across tests. try: current = sched_mod.get_scheduler() if current.is_running(): await current.stop() except RuntimeError: pass sched_mod.Scheduler = original_scheduler_cls # type: ignore[assignment] asyncio.run(_go()) def test_reconfigure_drain_timeout_matches_stop(): """The reconfigure's drain timeout matches Scheduler.stop() — both are 30 seconds. We can't wait 30s in a test, but we can verify the constant is shared.""" # Both code paths should reference the same constant. assert hasattr(sched_mod, "_DRAIN_TIMEOUT_SECONDS") assert sched_mod._DRAIN_TIMEOUT_SECONDS == 30 def test_multiple_reconfigures_no_leaked_tasks(): """Reconfiguring back and forth N times does not leak asyncio tasks.""" async def _go(): block = _stub_block(stub=True, host="block-a") sched_mod.configure_scheduler(block, sftp_block_name="a") sched = sched_mod.get_scheduler() await sched.start() before = len(asyncio.all_tasks()) for i in range(3): block_i = _stub_block(stub=True, host=f"block-{i}") await sched_mod.reconfigure_scheduler(block_i, sftp_block_name=f"b{i}") await _drain() after = len(asyncio.all_tasks()) # The new scheduler itself owns one task; allow +1. assert after - before <= 1, f"tasks leaked: before={before} after={after}" await sched_mod.get_scheduler().stop() asyncio.run(_go())