tests/core/test_task_executor.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 LocalTaskExecutor (core 1e Phase 2): in-process task execution.16 17 Real objects, no mocks: a REAL ``AsgiServer`` (storage on tmp_path) hosting a18 ``RoutedApplication`` with sync + async ``@route`` handlers, and the real19 ``TaskSpool`` the executor opens over ``server.storage``. Each test drives the20 full create → assign → execute lifecycle and asserts the settled folder position21 (terminated / aborted), the round-tripped result and the error stamping. The22 tests are ``async def`` so ``server.run_sync`` (the sync-handler pool seam) has a23 live event loop, matching ``test_routed_application.py``.24 """25 26 from __future__ import annotations27 28 from pathlib import Path29 from typing import Any30 31 import pytest32 from genro_routes import route33 34 from tests.storage_support import site_storage35 36 from genro_asgi import AsgiServer, RoutedApplication37 from genro_asgi.tasks import WORKER_ID, LocalTaskExecutor, TaskSpool, new_descriptor38 39 40 class DemoApp(RoutedApplication):41 """Test app: a sync handler, an async handler, and a failing one."""42 43 @route()44 def sum_sync(self, a: int = 0, b: int = 0) -> int:45 return a + b46 47 @route()48 async def sum_async(self, a: int = 0, b: int = 0) -> int:49 return a + b50 51 @route()52 def boom(self) -> None:53 raise ValueError("handler exploded")54 55 56 @pytest.fixture57 def server(tmp_path: Path) -> AsgiServer:58 """A real AsgiServer whose primary is the DemoApp, storage on tmp_path."""59 return AsgiServer(applications=[DemoApp(mount="")], storage=site_storage(tmp_path))60 61 62 @pytest.fixture63 def executor(server: AsgiServer) -> LocalTaskExecutor:64 """The executor bound to the live server (it opens the spool over server.storage)."""65 return LocalTaskExecutor(server)66 67 68 def stage(spool: TaskSpool, node_path: str, params: dict[str, Any], task_id: str = "t1") -> str:69 """Create a pending task on the primary (empty mount) and assign it to WORKER_ID."""70 descriptor = new_descriptor(task_id, owner="alice", mount="", node_path=node_path)71 spool.create(descriptor, params)72 spool.assign(task_id, WORKER_ID)73 return task_id74 75 76 class TestExecuteSuccess:77 """A handler that returns settles terminated, with the result written."""78 79 async def test_sync_handler_terminates_with_result(self, executor: LocalTaskExecutor) -> None:80 stage(executor.spool, "sum_sync", {"a": 2, "b": 3})81 outcome = await executor.execute("t1", WORKER_ID)82 assert outcome == "ok"83 assert executor.spool.read_result("t1") == 584 descriptor = executor.spool.get("t1")85 assert descriptor is not None86 assert descriptor["status"] == "terminated"87 assert descriptor["outcome"] == "ok"88 assert descriptor["error"] is None89 90 async def test_async_handler_terminates_with_result(self, executor: LocalTaskExecutor) -> None:91 stage(executor.spool, "sum_async", {"a": 10, "b": 5})92 outcome = await executor.execute("t1", WORKER_ID)93 assert outcome == "ok"94 assert executor.spool.read_result("t1") == 1595 descriptor = executor.spool.get("t1")96 assert descriptor is not None97 assert descriptor["status"] == "terminated"98 99 100 class TestExecuteFailure:101 """A handler that raises settles aborted, with the error stamped."""102 103 async def test_raising_handler_aborts_with_error(self, executor: LocalTaskExecutor) -> None:104 stage(executor.spool, "boom", {})105 outcome = await executor.execute("t1", WORKER_ID)106 assert outcome == "error"107 descriptor = executor.spool.get("t1")108 assert descriptor is not None109 assert descriptor["status"] == "aborted"110 assert descriptor["outcome"] == "error"111 assert "ValueError: handler exploded" == descriptor["error"]112 assert executor.spool.read_result("t1") is None113 114 115 class TestResolveErrors:116 """Missing task and unknown mount both raise LookupError."""117 118 async def test_unknown_task_raises(self, executor: LocalTaskExecutor) -> None:119 with pytest.raises(LookupError):120 await executor.execute("nope", WORKER_ID)121 122 async def test_unknown_mount_aborts_with_lookup_error(self, executor: LocalTaskExecutor) -> None:123 descriptor = new_descriptor("t1", owner="alice", mount="ghost", node_path="sum_sync")124 executor.spool.create(descriptor, {})125 executor.spool.assign("t1", WORKER_ID)126 outcome = await executor.execute("t1", WORKER_ID)127 assert outcome == "error"128 aborted = executor.spool.get("t1")129 assert aborted is not None130 assert aborted["status"] == "aborted"131 assert aborted["error"].startswith("LookupError:")132 133 134 class TestSpoolSeam:135 """The executor opens its own spool over the server's storage."""136 137 def test_spool_is_bound_to_server_storage(self, server: AsgiServer, executor: LocalTaskExecutor) -> None:138 assert isinstance(executor.spool, TaskSpool)139 assert executor.spool.storage is server.storage