tests/spa/orchestration/test_contract_global_store_dict.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 """The global store is one dictionary on the commander, behind one FIFO lock.16 17 Contract of issue #74 as fixed by the owner on 2026-09-0718 (``temp/GLOBAL_STORE_DICT_PLAN_2026-09-07.md``, v1.1):19 20 - the commander owns ``dict[str, Any]``; values are opaque, keys are literal;21 - ``get``, ``set``, ``delete`` and a ``for_update`` turn share ONE FIFO lock,22 so every operation waits while a turn is in force, reads included;23 - the ``get`` reply carries ``exists`` and ``value``: the client answers the24 caller's default when the key is absent, the stored value — ``None``25 included — when it is there;26 - a keyed turn transfers one value and replaces only that key at the release;27 a turn with no key transfers the whole dictionary and replaces it whole;28 - the release carries the complete replacement (``apply=True``) or aborts29 (``apply=False``); a release for a turn no longer in force does nothing;30 - a holder that dies releases only its own turn, master untouched.31 32 Written BEFORE the implementation (Step 1 of the plan): red until Step 2.33 """34 35 from __future__ import annotations36 37 import asyncio38 from datetime import datetime39 40 import pytest41 from genro_bag import Bag42 from genro_tytx import to_tytx43 44 from genro_asgi_multiworker_spa.global_store import GlobalStoreCommitUnconfirmed45 from genro_asgi_multiworker_spa.orchestration import FreezeHandler46 from genro_asgi_multiworker_spa.orchestration.worker_connector import CommanderCallFailed47 48 from .conftest import WORKER_NAME, XT_WorkerCommanderLane, wait_for49 50 SECOND_WORKER_NAME = "standard_0002"51 STORE_GET = "/commander/store/get"52 STORE_LOCK = "/commander/store/lock"53 STORE_UNLOCK = "/commander/store/unlock"54 55 56 @pytest.fixture57 async def second_worker_commander_lane(worker_commander_lane, tmp_path):58 lane = XT_WorkerCommanderLane(59 worker_commander_lane.commander,60 worker_commander_lane.worker_handler.group_handler,61 FreezeHandler(tmp_path / "frozen_users_second"),62 worker_name=SECOND_WORKER_NAME,63 )64 await lane.open()65 yield lane66 await lane.close()67 68 69 # --- the dictionary and the three simple operations ---------------------------70 71 72 async def test_the_master_is_a_dictionary_with_literal_keys(worker_commander_lane):73 # wf:contract: the commander's global_register is a dict; a dotted key is74 # wf:contract: one literal key, never a path.75 store = worker_commander_lane.worker.global_store76 master = worker_commander_lane.commander.global_register77 78 await asyncio.to_thread(store.set, "a.b", 1)79 80 assert isinstance(master, dict)81 assert master == {"a.b": 1}82 83 84 async def test_get_of_an_absent_key_answers_the_callers_default(worker_commander_lane):85 # wf:contract: exists=False on the wire → the client returns the default,86 # wf:contract: which never travels to the commander.87 store = worker_commander_lane.worker.global_store88 89 reply = await worker_commander_lane.worker.call(STORE_GET, {"key": "missing"})90 assert reply["exists"] is False91 92 assert await asyncio.to_thread(store.get, "missing", default="fallback") == "fallback"93 94 95 async def test_get_of_a_key_holding_none_answers_none_not_the_default(worker_commander_lane):96 # wf:contract: exists=True on the wire → the client returns the stored97 # wf:contract: value even when it is None; the default is not consulted.98 store = worker_commander_lane.worker.global_store99 await asyncio.to_thread(store.set, "empty", None)100 101 reply = await worker_commander_lane.worker.call(STORE_GET, {"key": "empty"})102 assert reply["exists"] is True103 104 assert await asyncio.to_thread(store.get, "empty", default="fallback") is None105 106 107 async def test_delete_removes_the_key_and_is_idempotent(worker_commander_lane):108 store = worker_commander_lane.worker.global_store109 master = worker_commander_lane.commander.global_register110 await asyncio.to_thread(store.set, "a", 1)111 112 await asyncio.to_thread(store.delete, "a")113 await asyncio.to_thread(store.delete, "a")114 115 assert "a" not in master116 117 118 async def test_values_keep_their_types_across_the_wire(worker_commander_lane):119 # wf:contract: a Bag with attributes and a datetime travel as values of the120 # wf:contract: dictionary under the existing TYTX conventions; the reader121 # wf:contract: gets a copy, never the master's own object.122 store = worker_commander_lane.worker.global_store123 master = worker_commander_lane.commander.global_register124 bag = Bag()125 bag.set_item("x", datetime(2026, 1, 1, 12), _attributes={"tag": "t"})126 bag.set_item("empty", Bag())127 128 await asyncio.to_thread(store.set, "CACHE_TS", bag)129 read = await asyncio.to_thread(store.get, "CACHE_TS")130 131 assert isinstance(read, Bag)132 assert read["x"].replace(tzinfo=None) == datetime(2026, 1, 1, 12)133 assert read.get_attr("x") == {"tag": "t"}134 assert isinstance(read["empty"], Bag) and len(read["empty"]) == 0135 read.set_item("x", 0)136 assert master["CACHE_TS"]["x"] != 0137 138 139 # --- the turn ------------------------------------------------------------------140 141 142 async def test_a_keyed_turn_replaces_only_its_key(worker_commander_lane):143 # wf:contract: for_update(key) yields a lease whose value is a private copy;144 # wf:contract: the master shows nothing until the exit, then only that key145 # wf:contract: changes.146 store = worker_commander_lane.worker.global_store147 master = worker_commander_lane.commander.global_register148 master["config"] = Bag({"enabled": False})149 master["other"] = 1150 151 async with store.for_update("config") as turn:152 assert turn.exists is True153 turn.value["enabled"] = True154 assert master["config"]["enabled"] is False155 156 assert master["config"]["enabled"] is True157 assert master["other"] == 1158 159 160 async def test_an_absent_key_starts_the_turn_with_exists_false_and_value_none(161 worker_commander_lane,162 ):163 # wf:contract: no Bag is invented by the commander; the consumer decides.164 store = worker_commander_lane.worker.global_store165 master = worker_commander_lane.commander.global_register166 167 async with store.for_update("counter") as turn:168 assert turn.exists is False and turn.value is None169 turn.value = (turn.value if turn.exists else 0) + 1170 171 assert master["counter"] == 1172 173 174 async def test_a_turn_may_publish_none(worker_commander_lane):175 store = worker_commander_lane.worker.global_store176 master = worker_commander_lane.commander.global_register177 master["k"] = 1178 179 async with store.for_update("k") as turn:180 turn.value = None181 182 assert "k" in master and master["k"] is None183 184 185 async def test_a_turn_with_no_key_replaces_the_whole_dictionary(worker_commander_lane):186 # wf:contract: the legacy `with globalStore()` shape: the value is the187 # wf:contract: dictionary snapshot, edits and deletions land atomically.188 store = worker_commander_lane.worker.global_store189 master = worker_commander_lane.commander.global_register190 master.update({"a": 1, "b": 2})191 192 async with store.for_update() as turn:193 turn.value["a"] = 10194 del turn.value["b"]195 turn.value["c"] = 3196 assert master == {"a": 1, "b": 2}197 198 assert master == {"a": 10, "c": 3}199 200 201 async def test_a_body_that_raises_aborts_and_creates_nothing(worker_commander_lane):202 store = worker_commander_lane.worker.global_store203 master = worker_commander_lane.commander.global_register204 205 with pytest.raises(RuntimeError, match="fell over"):206 async with store.for_update("fresh") as turn:207 turn.value = 99208 raise RuntimeError("the site fell over")209 210 assert "fresh" not in master211 assert worker_commander_lane.commander.global_lock.holder is None212 213 214 async def test_an_aborted_turn_publishes_nothing_and_frees_the_lock_at_exit(215 worker_commander_lane,216 ):217 # wf:contract: turn.abort() marks the turn; the lock is still held until the218 # wf:contract: with exits, then the release carries apply=False — whatever219 # wf:contract: the body did to value after the call. An absent key stays absent.220 store = worker_commander_lane.worker.global_store221 commander = worker_commander_lane.commander222 223 async with store.for_update("CACHE_TS") as turn:224 assert turn.exists is False225 turn.abort()226 assert commander.global_lock.holder == turn.request_id227 turn.value = {"foo": 1}228 229 assert "CACHE_TS" not in commander.global_register230 assert commander.global_lock.holder is None231 assert not commander.global_lock.lock.locked()232 233 234 async def test_the_sync_lease_works_from_a_pool_thread(worker_commander_lane):235 store = worker_commander_lane.worker.global_store236 master = worker_commander_lane.commander.global_register237 238 def body() -> None:239 with store.for_update("n") as turn:240 turn.value = 7241 242 await asyncio.to_thread(body)243 assert master["n"] == 7244 245 246 # --- one FIFO lock for everything ---------------------------------------------247 248 249 async def test_a_read_of_any_key_waits_while_a_turn_is_in_force(250 worker_commander_lane, second_worker_commander_lane251 ):252 # wf:contract: a get during a turn is parked on the same FIFO lock, even on253 # wf:contract: an unrelated key, and answers after the release.254 master = worker_commander_lane.commander.global_register255 master.update({"held": 1, "unrelated": 1})256 granted = asyncio.Event()257 release_now = asyncio.Event()258 seen: list[dict] = []259 260 async def holder() -> None:261 async with worker_commander_lane.worker.global_store.for_update("held") as turn:262 granted.set()263 turn.value = 2264 await release_now.wait()265 266 async def reader() -> None:267 seen.append(await second_worker_commander_lane.worker.call(STORE_GET, {"key": "unrelated"}))268 269 first = asyncio.create_task(holder())270 await granted.wait()271 second = asyncio.create_task(reader())272 await wait_for(lambda: bool(second_worker_commander_lane.worker._parent_calls))273 assert seen == []274 275 release_now.set()276 await asyncio.gather(first, second)277 assert seen[0]["exists"] is True and master["held"] == 2278 279 280 async def test_waiting_turns_are_served_in_order_and_see_the_previous_release(281 worker_commander_lane, second_worker_commander_lane282 ):283 master = worker_commander_lane.commander.global_register284 master["n"] = 1285 granted = asyncio.Event()286 release_now = asyncio.Event()287 second_saw: list[int] = []288 289 async def first_holder() -> None:290 async with worker_commander_lane.worker.global_store.for_update("n") as turn:291 granted.set()292 turn.value += 10293 await release_now.wait()294 295 async def second_holder() -> None:296 async with second_worker_commander_lane.worker.global_store.for_update("n") as turn:297 second_saw.append(turn.value)298 299 first = asyncio.create_task(first_holder())300 await granted.wait()301 second = asyncio.create_task(second_holder())302 await wait_for(lambda: bool(second_worker_commander_lane.worker._parent_calls))303 release_now.set()304 await asyncio.gather(first, second)305 306 assert second_saw == [11]307 308 309 # --- releases out of turn, deaths ----------------------------------------------310 311 312 async def test_a_stale_release_neither_writes_nor_frees_a_newer_turn(worker_commander_lane):313 # wf:contract: an unlock quoting a request id that is not the holder's does314 # wf:contract: nothing: the newer turn stays in force, the master is untouched.315 worker = worker_commander_lane.worker316 commander = worker_commander_lane.commander317 commander.global_register["k"] = 1318 await worker.call(STORE_LOCK, {"worker": WORKER_NAME, "request_id": "current", "key": "k"})319 320 reply = await worker.call(321 STORE_UNLOCK, {"request_id": "stale", "apply": True, "value": to_tytx(99, "json")}322 )323 324 assert reply == {"applied": False}325 assert commander.global_lock.holds("current") is True326 assert commander.global_register["k"] == 1327 await worker.call(STORE_UNLOCK, {"request_id": "current", "apply": False})328 assert commander.global_lock.holder is None329 330 331 async def test_a_dead_holder_frees_only_its_turn_with_the_master_untouched(332 worker_commander_lane, second_worker_commander_lane333 ):334 commander = worker_commander_lane.commander335 commander.global_register["k"] = 1336 await worker_commander_lane.worker.call(337 STORE_LOCK, {"worker": WORKER_NAME, "request_id": "r1", "key": "k"}338 )339 assert commander.global_lock.held_by(WORKER_NAME) is True340 341 worker_commander_lane.worker_handler.on_child_lost()342 343 assert commander.global_lock.holder is None344 assert commander.global_register["k"] == 1345 async with second_worker_commander_lane.worker.global_store.for_update("k") as turn:346 assert turn.value == 1347 348 349 async def test_only_string_keys_are_accepted(worker_commander_lane):350 # wf:contract: key=None on the wire means the whole dictionary in a turn and351 # wf:contract: nothing else; any other non-string key is refused by the commander.352 worker = worker_commander_lane.worker353 with pytest.raises(CommanderCallFailed, match="TypeError"):354 await worker.call(STORE_GET, {"key": 3})355 with pytest.raises(CommanderCallFailed, match="TypeError"):356 await worker.call(STORE_LOCK, {"worker": WORKER_NAME, "request_id": "r", "key": 3})357 assert worker_commander_lane.commander.global_lock.holder is None358 359 360 async def test_a_commit_whose_answer_is_lost_is_reported_uncertain(worker_commander_lane):361 # wf:contract: the commander publishes the value, the wire ends before the362 # wf:contract: REPLY arrives: the lease raises GlobalStoreCommitUnconfirmed,363 # wf:contract: retries nothing, and the local turn state is cleared.364 worker = worker_commander_lane.worker365 commander = worker_commander_lane.commander366 commander.global_register["k"] = 1367 real_release = commander.global_lock.release368 369 def release_then_lose_the_wire() -> None:370 # The value is already published when the commander releases; the REPLY371 # has not left yet. Closing the worker's socket here is the lost answer.372 real_release()373 worker.stream.writer.close()374 375 commander.global_lock.release = release_then_lose_the_wire376 try:377 with pytest.raises(GlobalStoreCommitUnconfirmed) as report:378 async with worker.global_store.for_update("k") as turn:379 turn.value = 2380 finally:381 commander.global_lock.release = real_release382 383 assert commander.global_register["k"] == 2384 assert report.value.key == "k" and isinstance(report.value.cause, ConnectionError)385 assert worker.global_store.active_turn.get() is None386 assert commander.global_lock.holder is None387 388 389 async def test_the_change_batch_machinery_is_gone():390 # wf:contract: no capturing store, no applied changes, no datachange import.391 from pathlib import Path392 393 from genro_asgi_multiworker_spa import global_store as global_store_module394 from genro_asgi_multiworker_spa.orchestration import SpaCommander, spa_commander395 396 assert not hasattr(global_store_module, "CapturingGlobalStore")397 assert not hasattr(SpaCommander, "apply_global_store_changes")398 written = Path(global_store_module.__file__).read_text() + Path(spa_commander.__file__).read_text()399 assert "datachange" not in written