tests/core/test_task_spool.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 TaskSpool (core 1e Phase 1): the folder-move task model.16 17 Real filesystem (a one-mount ``StorageManager`` on tmp_path), no mocks. Covers the descriptor18 shape, the create → assign → progress → settle lifecycle, the folder positions19 behind each state, cancel/result round-trips, terminal invariants (re-settle20 raises), owner/status queries and purge.21 """22 23 from pathlib import Path24 from typing import Any25 26 import pytest27 28 from genro_storage import StorageManager29 30 from tests.storage_support import site_storage31 32 from genro_asgi.tasks import STATUSES, TaskSpool, new_descriptor33 34 35 @pytest.fixture36 def storage(tmp_path: Path) -> StorageManager:37 """Storage rooted in tmp_path (the spool lives on the 'site' mount)."""38 return site_storage(tmp_path)39 40 41 @pytest.fixture42 def spool(storage: StorageManager) -> TaskSpool:43 """A TaskSpool bound to the temporary storage."""44 return TaskSpool(storage)45 46 47 def make_task(spool: TaskSpool, task_id: str = "t1", owner: str = "alice", **kwargs: Any) -> str:48 """Create a pending task with a standard descriptor; returns the task_id."""49 descriptor = new_descriptor(task_id, owner=owner, mount="shop", node_path="cleanup", **kwargs)50 return spool.create(descriptor, {"pkeys": [1, 2, 3]})51 52 53 class TestDescriptor:54 """new_descriptor shape."""55 56 def test_fresh_descriptor_fields(self) -> None:57 desc = new_descriptor("t1", owner="alice", mount="shop", node_path="cleanup")58 assert desc["task_id"] == "t1"59 assert desc["owner"] == "alice"60 assert desc["mount"] == "shop"61 assert desc["node_path"] == "cleanup"62 assert desc["status"] == "pending"63 assert desc["worker_id"] is None64 assert desc["created_ts"] > 065 assert desc["started_ts"] is None66 assert desc["ended_ts"] is None67 assert desc["outcome"] is None68 assert desc["error"] is None69 70 def test_session_id_default_none(self) -> None:71 desc = new_descriptor("t1", owner="alice", mount="shop", node_path="cleanup")72 assert desc["session_id"] is None73 74 def test_session_id_carried(self) -> None:75 desc = new_descriptor(76 "t1", owner="alice", mount="shop", node_path="cleanup", session_id="mcp-abc"77 )78 assert desc["session_id"] == "mcp-abc"79 80 def test_statuses_constant(self) -> None:81 assert STATUSES == ("pending", "active", "terminated", "aborted")82 83 84 class TestLifecycle:85 """create → assign → progress → result → settle, and the folder positions."""86 87 def test_create_lands_in_pending(self, spool: TaskSpool, tmp_path: Path) -> None:88 make_task(spool)89 assert (tmp_path / "batches" / "pending" / "t1" / "descriptor.json").is_file()90 assert (tmp_path / "batches" / "pending" / "t1" / "params.pkl").is_file()91 assert [d["task_id"] for d in spool.list_pending()] == ["t1"]92 93 def test_read_params_round_trip(self, spool: TaskSpool) -> None:94 make_task(spool)95 assert spool.read_params("t1") == {"pkeys": [1, 2, 3]}96 97 def test_read_params_missing_raises(self, spool: TaskSpool) -> None:98 with pytest.raises(LookupError):99 spool.read_params("ghost")100 101 def test_assign_moves_to_active_worker(self, spool: TaskSpool, tmp_path: Path) -> None:102 make_task(spool)103 spool.assign("t1", "w1")104 assert not (tmp_path / "batches" / "pending" / "t1").exists()105 assert (tmp_path / "batches" / "active" / "w1" / "t1").is_dir()106 (active,) = spool.list_active("w1")107 assert active["status"] == "active"108 assert active["worker_id"] == "w1"109 assert active["started_ts"] is not None110 111 def test_assign_not_pending_raises(self, spool: TaskSpool) -> None:112 with pytest.raises(LookupError):113 spool.assign("ghost", "w1")114 115 def test_progress_round_trip(self, spool: TaskSpool) -> None:116 make_task(spool)117 spool.assign("t1", "w1")118 assert spool.read_progress("t1") is None119 spool.write_progress("t1", "w1", {"progress": 2, "maximum": 3})120 assert spool.read_progress("t1") == {"progress": 2, "maximum": 3}121 122 def test_read_progress_missing_task(self, spool: TaskSpool) -> None:123 assert spool.read_progress("ghost") is None124 125 def test_settle_ok_lands_in_terminated(self, spool: TaskSpool, tmp_path: Path) -> None:126 make_task(spool)127 spool.assign("t1", "w1")128 spool.settle("t1", "w1", "ok")129 assert (tmp_path / "batches" / "terminated" / "t1").is_dir()130 descriptor = spool.get("t1")131 assert descriptor is not None132 assert descriptor["status"] == "terminated"133 assert descriptor["outcome"] == "ok"134 assert descriptor["error"] is None135 assert descriptor["ended_ts"] is not None136 137 def test_settle_error_lands_in_aborted(self, spool: TaskSpool, tmp_path: Path) -> None:138 """The orphan/error path: any outcome != 'ok' settles aborted with the error stamped."""139 make_task(spool)140 spool.assign("t1", "w1")141 spool.settle("t1", "w1", "error", error="boom")142 assert (tmp_path / "batches" / "aborted" / "t1").is_dir()143 descriptor = spool.get("t1")144 assert descriptor is not None145 assert descriptor["status"] == "aborted"146 assert descriptor["outcome"] == "error"147 assert descriptor["error"] == "boom"148 149 def test_settle_not_active_raises(self, spool: TaskSpool) -> None:150 make_task(spool)151 with pytest.raises(LookupError):152 spool.settle("t1", "w1", "ok")153 154 def test_terminal_resettle_raises(self, spool: TaskSpool) -> None:155 """batch_id is terminal: a settled folder is no longer active, so settle raises."""156 make_task(spool)157 spool.assign("t1", "w1")158 spool.settle("t1", "w1", "ok")159 with pytest.raises(LookupError):160 spool.settle("t1", "w1", "ok")161 162 163 class TestCancelAndResult:164 """Cancel marker and pickled result inside the task folder."""165 166 def test_cancel_round_trip(self, spool: TaskSpool) -> None:167 make_task(spool)168 assert spool.is_cancelled("t1") is False169 spool.request_cancel("t1")170 assert spool.is_cancelled("t1") is True171 172 def test_cancel_missing_raises(self, spool: TaskSpool) -> None:173 with pytest.raises(LookupError):174 spool.request_cancel("ghost")175 176 def test_is_cancelled_missing_task(self, spool: TaskSpool) -> None:177 assert spool.is_cancelled("ghost") is False178 179 def test_result_round_trip(self, spool: TaskSpool) -> None:180 make_task(spool)181 spool.assign("t1", "w1")182 spool.write_result("t1", "w1", {"done": 3, "ids": [1, 2, 3]})183 spool.settle("t1", "w1", "ok")184 assert spool.read_result("t1") == {"done": 3, "ids": [1, 2, 3]}185 186 def test_result_not_written(self, spool: TaskSpool) -> None:187 make_task(spool)188 assert spool.read_result("t1") is None189 190 def test_result_missing_task(self, spool: TaskSpool) -> None:191 assert spool.read_result("ghost") is None192 193 194 class TestQueries:195 """Owner/status queries across states."""196 197 def test_get_missing(self, spool: TaskSpool) -> None:198 assert spool.get("ghost") is None199 200 def test_list_by_owner_spans_states(self, spool: TaskSpool) -> None:201 make_task(spool, "t1", owner="alice")202 make_task(spool, "t2", owner="alice")203 make_task(spool, "t3", owner="bob")204 spool.assign("t1", "w1")205 spool.settle("t1", "w1", "ok")206 spool.assign("t2", "w1")207 alice = {d["task_id"]: d["status"] for d in spool.list_by_owner("alice")}208 assert alice == {"t1": "terminated", "t2": "active"}209 assert [d["task_id"] for d in spool.list_by_owner("bob")] == ["t3"]210 211 def test_list_by_status_active_spans_workers(self, spool: TaskSpool) -> None:212 make_task(spool, "t1")213 make_task(spool, "t2")214 spool.assign("t1", "w1")215 spool.assign("t2", "w2")216 active = {d["task_id"]: d["worker_id"] for d in spool.list_by_status("active")}217 assert active == {"t1": "w1", "t2": "w2"}218 219 def test_list_by_status_empty_states(self, spool: TaskSpool) -> None:220 for status in STATUSES:221 assert spool.list_by_status(status) == []222 223 def test_list_active_unknown_worker(self, spool: TaskSpool) -> None:224 assert spool.list_active("ghost") == []225 226 227 class TestPurge:228 """Tree removal of a task folder."""229 230 def test_purge_removes_tree(self, spool: TaskSpool, tmp_path: Path) -> None:231 make_task(spool)232 spool.assign("t1", "w1")233 spool.settle("t1", "w1", "ok")234 assert spool.purge("t1") is True235 assert not (tmp_path / "batches" / "terminated" / "t1").exists()236 assert spool.get("t1") is None237 238 def test_purge_missing(self, spool: TaskSpool) -> None:239 assert spool.purge("ghost") is False