tests/core/test_task_manager.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 TaskManager + TaskMixin lifespan hook (core 1e Phase 3): the backbone.16 17 Real objects, no mocks: a REAL ``AsgiServer`` (storage on tmp_path) hosting a18 ``RoutedApplication`` with a sync + an async ``@route`` handler, its real spool,19 and the real ASGI ``lifespan`` protocol driven through ``server.__call__``.20 21 Driving the loop end-to-end is the crux: ``server.__call__`` on a ``lifespan``22 scope blocks until it receives ``lifespan.shutdown``, while the worker loop runs23 in the background. The ``run_until_settled`` driver returns ``startup`` first,24 then polls the spool and returns ``shutdown`` only once every staged task has25 settled (with a timeout guard) — so the assertions run against a fully drained26 spool and a cleanly stopped manager.27 """28 29 from __future__ import annotations30 31 import asyncio32 from pathlib import Path33 34 import pytest35 from genro_routes import route36 37 from tests.storage_support import site_storage38 39 from genro_asgi import AsgiServer, BaseServer, RoutedApplication40 from genro_asgi.application import BaseApplication41 from genro_asgi.tasks import TaskManager, new_descriptor42 from genro_asgi.tasks.manager import POLL_SECONDS43 from genro_asgi.tasks.scheduler import TaskScheduler44 from genro_asgi.tasks.store import FileTaskStore45 46 47 class DemoApp(RoutedApplication):48 """Test app: a sync handler, an async handler, and a failing one."""49 50 @route()51 def sum_sync(self, a: int = 0, b: int = 0) -> int:52 return a + b53 54 @route()55 async def sum_async(self, a: int = 0, b: int = 0) -> int:56 return a + b57 58 @route()59 def boom(self) -> None:60 raise ValueError("handler exploded")61 62 63 @pytest.fixture64 def server(tmp_path: Path) -> AsgiServer:65 """A real AsgiServer whose primary is the DemoApp, storage on tmp_path."""66 return AsgiServer(applications=[DemoApp(mount="")], storage=site_storage(tmp_path))67 68 69 def stage(server: AsgiServer, node_path: str, params: dict[str, int], task_id: str) -> None:70 """Drop a pending task on the primary (empty mount) for the loop to pick up."""71 descriptor = new_descriptor(task_id, owner="alice", mount="", node_path=node_path)72 server.tasks.spool.create(descriptor, params)73 74 75 async def run_until_settled(server: AsgiServer, task_ids: list[str], timeout: float = 5.0) -> None:76 """Drive one startup/shutdown round-trip, releasing shutdown once tasks settle.77 78 Returns ``lifespan.startup`` first (arming the worker loop), then polls the79 spool and returns ``lifespan.shutdown`` only when every ``task_id`` has left80 the pending/active states (settled terminated or aborted) — or the timeout81 fires, so a stuck loop fails the test instead of hanging.82 """83 spool = server.tasks.spool84 started = False85 deadline = asyncio.get_running_loop().time() + timeout86 87 def settled() -> bool:88 for task_id in task_ids:89 descriptor = spool.get(task_id)90 if descriptor is None or descriptor["status"] not in ("terminated", "aborted"):91 return False92 return True93 94 async def receive() -> dict[str, object]:95 nonlocal started96 if not started:97 started = True98 return {"type": "lifespan.startup"}99 while not settled() and asyncio.get_running_loop().time() < deadline:100 await asyncio.sleep(POLL_SECONDS / 2)101 return {"type": "lifespan.shutdown"}102 103 async def send(message: dict[str, object]) -> None:104 pass105 106 await server({"type": "lifespan"}, receive, send)107 108 109 class TestManagerWiring:110 """The manager owns the spool/executor/hub and reuses the storage seam."""111 112 def test_manager_owns_the_seam(self, server: AsgiServer) -> None:113 manager = server.tasks114 assert isinstance(manager, TaskManager)115 assert manager.server is server116 assert manager.spool is manager.executor.spool117 assert manager.spool.storage is server.storage118 assert isinstance(manager.scheduler, TaskScheduler) # wired in Phase 4119 assert manager.scheduler.manager is manager120 assert isinstance(manager.task_store, FileTaskStore)121 assert manager.task_store.storage is server.storage122 assert manager.worker_id == "local"123 124 def test_manager_built_lazily_and_cached(self, server: AsgiServer) -> None:125 assert server.tasks is server.tasks # same instance on re-access126 127 def test_no_mixin_no_tasks(self) -> None:128 plain = BaseServer(applications=[BaseApplication(mount="")])129 assert not hasattr(plain, "tasks_enabled")130 131 def test_disabled_server_raises_on_access(self, tmp_path: Path) -> None:132 disabled = AsgiServer(applications=[DemoApp(mount="")], tasks=False,133 storage=site_storage(tmp_path))134 assert disabled.tasks_enabled is False135 with pytest.raises(RuntimeError, match="disabled"):136 disabled.tasks137 138 139 class TestLifespanDrivenExecution:140 """The lifespan hook runs the worker loop: pending tasks get executed."""141 142 async def test_pending_task_runs_and_terminates(self, server: AsgiServer) -> None:143 stage(server, "sum_sync", {"a": 2, "b": 3}, "t1")144 await run_until_settled(server, ["t1"])145 descriptor = server.tasks.spool.get("t1")146 assert descriptor is not None147 assert descriptor["status"] == "terminated"148 assert server.tasks.spool.read_result("t1") == 5149 assert server.tasks.running is False # loop stopped at shutdown150 151 async def test_async_and_failing_tasks_both_settle(self, server: AsgiServer) -> None:152 stage(server, "sum_async", {"a": 10, "b": 5}, "ok")153 stage(server, "boom", {}, "bad")154 await run_until_settled(server, ["ok", "bad"])155 ok = server.tasks.spool.get("ok")156 bad = server.tasks.spool.get("bad")157 assert ok is not None and ok["status"] == "terminated"158 assert server.tasks.spool.read_result("ok") == 15159 assert bad is not None and bad["status"] == "aborted"160 assert bad["error"] == "ValueError: handler exploded"161 162 async def test_loop_not_running_before_startup(self, server: AsgiServer) -> None:163 assert server.tasks.running is False # armed but idle until lifespan164 165 166 class TestNonLifespanPassThrough:167 """A disabled server passes the lifespan straight through (no loop)."""168 169 async def test_disabled_lifespan_still_acks(self, tmp_path: Path) -> None:170 disabled = AsgiServer(applications=[DemoApp(mount="")], tasks=False,171 storage=site_storage(tmp_path))172 sent: list[dict[str, object]] = []173 queue = [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]174 175 async def receive() -> dict[str, object]:176 return queue.pop(0)177 178 async def send(message: dict[str, object]) -> None:179 sent.append(message)180 181 await disabled({"type": "lifespan"}, receive, send)182 assert {"type": "lifespan.startup.complete"} in sent183 assert {"type": "lifespan.shutdown.complete"} in sent