tests/core/test_task_scheduler.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """Tests for tasks.scheduler (core 1e Phase 4): the recurring loop.16 17 Real objects, no mocks: a real ``AsgiServer`` (storage on tmp_path) whose18 primary + mount declare tasks via ``@route(task=..., task_every=...)``. The19 scan/sync_defaults/tick/run_now steps are driven DIRECTLY (no wall-clock20 sleeping); a final lifespan-driven test asserts the manager starts and stops the21 scheduler loop alongside the worker loop.22 """23 24 from __future__ import annotations25 26 import asyncio27 from pathlib import Path28 29 import pytest30 from genro_routes import route31 32 from tests.storage_support import site_storage33 34 from genro_asgi import AsgiServer, RoutedApplication35 from genro_asgi.tasks.scheduler import TaskScheduler36 37 RUN_MARKS: list[str] = []38 39 40 async def settle(predicate, timeout: float = 5.0, interval: float = 0.02) -> None:41 """Wait until ``predicate()`` is truthy, sampling every ``interval`` seconds.42 43 The spawned ``_execute`` task crosses two thread-pool hops plus a storage44 write, so a fixed sleep is a race by construction. While polling, ANY45 exception counts as "not yet" — a store record read mid-write raises46 ``JSONDecodeError`` — but the caller's own asserts run after this returns,47 so a condition that never converges still fails loudly on the real check.48 """49 loop = asyncio.get_running_loop()50 deadline = loop.time() + timeout51 while loop.time() < deadline:52 try:53 if predicate():54 return55 except Exception:56 pass57 await asyncio.sleep(interval)58 59 60 class DemoApp(RoutedApplication):61 """Primary app: a default-scheduled task, a store-only task, a failing one."""62 63 @route(task="cleanup", task_every="1s")64 def cleanup(self) -> str:65 RUN_MARKS.append("cleanup")66 return "cleaned"67 68 @route(task="report") # schedulable only through a store record69 def report(self) -> str:70 RUN_MARKS.append("report")71 return "reported"72 73 @route(task="boom", task_every="1s")74 def boom(self) -> None:75 raise ValueError("scheduled boom")76 77 @route(task="slow", task_every="1s")78 async def slow(self) -> str:79 RUN_MARKS.append("slow")80 return "slow-done"81 82 83 class MountApp(RoutedApplication):84 """A secondary mount contributing its own task (tests multi-app scan)."""85 86 @route(task="mounted", task_every="1s")87 def mounted(self) -> str:88 return "mounted"89 90 91 @pytest.fixture(autouse=True)92 def _clear_marks() -> None:93 RUN_MARKS.clear()94 95 96 @pytest.fixture97 def server(tmp_path: Path) -> AsgiServer:98 """A real AsgiServer: DemoApp primary + MountApp mounted, storage on tmp_path."""99 srv = AsgiServer(100 applications=[DemoApp(mount=""), MountApp(code="extra")],101 storage=site_storage(tmp_path),102 )103 return srv104 105 106 def scheduler(server: AsgiServer) -> TaskScheduler:107 """The server's scheduler (built with the manager, lazily)."""108 return server.tasks.scheduler109 110 111 class TestScan:112 """The routing tree is the live registry."""113 114 def test_scan_collects_tasks_from_every_application(self, server: AsgiServer) -> None:115 registry = scheduler(server).scan()116 assert {"cleanup", "report", "boom", "slow", "mounted"} <= set(registry)117 assert callable(registry["cleanup"]["callable"])118 assert registry["cleanup"]["metadata"]["task_every"] == "1s"119 120 def test_duplicate_task_name_excluded(self, tmp_path: Path) -> None:121 class Dup(RoutedApplication):122 @route(task="twin")123 def one(self) -> None: ...124 125 @route(task="twin")126 def two(self) -> None: ...127 128 srv = AsgiServer(applications=[Dup(mount="")], storage=site_storage(tmp_path))129 assert "twin" not in srv.tasks.scheduler.scan() # both excluded, no silent pick130 131 132 class TestSyncDefaults:133 """A declared task_every/task_cron auto-creates the code-default record."""134 135 def test_default_record_created(self, server: AsgiServer) -> None:136 sch = scheduler(server)137 sch.sync_defaults(sch.scan(), now=1000.0)138 rec = sch.store.get("cleanup")139 assert rec is not None140 assert rec["kind"] == "every" and rec["spec"] == "1s"141 assert rec["next_run_ts"] == 1001.0 # now + 1s142 143 def test_store_only_task_gets_no_default(self, server: AsgiServer) -> None:144 sch = scheduler(server)145 sch.sync_defaults(sch.scan(), now=1000.0)146 assert sch.store.get("report") is None # no task_every/task_cron147 148 def test_existing_record_wins(self, server: AsgiServer) -> None:149 sch = scheduler(server)150 sch.store.save({151 "code": "cleanup", "task_name": "cleanup", "target_kind": "task",152 "kwargs": {}, "kind": "every", "spec": "1s", "enabled": True,153 "next_run_ts": 42.0, "last_run_ts": None, "last_outcome": None,154 "last_error": None, "last_duration": None,155 })156 sch.sync_defaults(sch.scan(), now=1000.0)157 rec = sch.store.get("cleanup")158 assert rec is not None and rec["next_run_ts"] == 42.0 # user record preserved159 160 161 class TestTick:162 """tick: scan -> sync_defaults -> run every due schedule."""163 164 async def test_due_task_runs_and_records_outcome(self, server: AsgiServer) -> None:165 sch = scheduler(server)166 sch.sync_defaults(sch.scan(), now=0.0) # creates cleanup at 1.0167 await sch.tick() # now >> 1.0 -> due168 # The log append is _execute's LAST write (record first, log after),169 # so a non-empty log implies every earlier assertion target is settled.170 await settle(lambda: sch.store.read_log("cleanup"))171 rec = sch.store.get("cleanup")172 assert rec is not None and rec["last_outcome"] == "ok"173 assert "cleanup" in RUN_MARKS174 assert sch.store.read_log("cleanup")[-1]["outcome"] == "ok"175 176 async def test_failing_task_recorded_error(self, server: AsgiServer) -> None:177 sch = scheduler(server)178 sch.sync_defaults(sch.scan(), now=0.0)179 await sch.tick()180 await settle(lambda: (sch.store.get("boom") or {}).get("last_outcome"))181 rec = sch.store.get("boom")182 assert rec is not None and rec["last_outcome"] == "error"183 assert "scheduled boom" in rec["last_error"]184 185 async def test_async_task_runs(self, server: AsgiServer) -> None:186 sch = scheduler(server)187 sch.sync_defaults(sch.scan(), now=0.0)188 await sch.tick()189 await settle(lambda: "slow" in RUN_MARKS)190 assert "slow" in RUN_MARKS191 192 async def test_no_overlap(self, server: AsgiServer) -> None:193 sch = scheduler(server)194 sch.sync_defaults(sch.scan(), now=0.0)195 sch._running.add("cleanup") # simulate an in-flight run196 await sch.tick()197 await asyncio.sleep(0.05)198 assert "cleanup" not in RUN_MARKS # skipped, not re-run199 200 async def test_orphan_record_never_runs(self, server: AsgiServer) -> None:201 sch = scheduler(server)202 sch.store.save({203 "code": "ghost", "task_name": "ghost", "target_kind": "task",204 "kwargs": {}, "kind": "every", "spec": "1s", "enabled": True,205 "next_run_ts": 0.0, "last_run_ts": None, "last_outcome": None,206 "last_error": None, "last_duration": None,207 })208 await sch.tick()209 await asyncio.sleep(0.05)210 rec = sch.store.get("ghost")211 assert rec is not None and rec["last_run_ts"] is None # never executed212 213 214 class TestRunNow:215 """run_now fires immediately with the no-overlap guard and outcome path."""216 217 def test_run_now_inline_when_no_loop(self, server: AsgiServer) -> None:218 # sync context: no running loop -> the schedule executes inline ("done")219 sch = scheduler(server)220 sch.sync_defaults(sch.scan(), now=0.0)221 assert sch.run_now("cleanup") == "done"222 assert "cleanup" in RUN_MARKS223 224 async def test_run_now_started_on_live_loop(self, server: AsgiServer) -> None:225 sch = scheduler(server)226 sch.start()227 try:228 sch.sync_defaults(sch.scan(), now=0.0)229 assert sch.run_now("cleanup") == "started"230 await settle(lambda: "cleanup" in RUN_MARKS)231 assert "cleanup" in RUN_MARKS232 finally:233 await sch.stop()234 235 def test_run_now_unknown_raises(self, server: AsgiServer) -> None:236 with pytest.raises(LookupError, match="schedule not found"):237 scheduler(server).run_now("nope")238 239 def test_run_now_orphan_raises(self, server: AsgiServer) -> None:240 sch = scheduler(server)241 sch.store.save({242 "code": "ghost", "task_name": "ghost", "target_kind": "task",243 "kwargs": {}, "kind": "every", "spec": "1s", "enabled": True,244 "next_run_ts": 0.0, "last_run_ts": None, "last_outcome": None,245 "last_error": None, "last_duration": None,246 })247 with pytest.raises(LookupError, match="orphan task"):248 sch.run_now("ghost")249 250 def test_run_now_running_skips(self, server: AsgiServer) -> None:251 sch = scheduler(server)252 sch.sync_defaults(sch.scan(), now=0.0)253 sch._running.add("cleanup")254 assert sch.run_now("cleanup") == "running"255 256 257 class TestLifespanLifecycle:258 """The manager starts and stops the scheduler loop with the worker loop."""259 260 async def test_scheduler_started_and_stopped(self, server: AsgiServer) -> None:261 sent: list[dict[str, object]] = []262 queue = [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]263 sch_seen_running = asyncio.Event()264 265 async def receive() -> dict[str, object]:266 if queue[0]["type"] == "lifespan.shutdown":267 # by now startup has run: the scheduler loop task must be live268 assert server.tasks.scheduler._loop_task is not None269 assert not server.tasks.scheduler._loop_task.done()270 sch_seen_running.set()271 return queue.pop(0)272 273 async def send(message: dict[str, object]) -> None:274 sent.append(message)275 276 await server({"type": "lifespan"}, receive, send)277 assert sch_seen_running.is_set()278 assert server.tasks.scheduler._loop_task is None # stopped at shutdown