Skip to content

tests/spa/test_spa_global_store.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's own classes, tested bare.16 17 The store lives ONLY on the commander — one dictionary behind one FIFO lock —18 and the whole surface (get/set/delete on the lane, the turn whose grant carries19 the selected value and whose release publishes it whole) is pinned end to end by20 the orchestration contract tests (``tests/spa/orchestration/test_contract_global_store_dict.py``,21 ``test_contract_phase10_global_store.py``, ``test_orchestration_store_get.py``).22 What belongs here is the lock itself: who holds the turn, and what a release23 for a turn no longer in force must not do.24 """25 26 from __future__ import annotations27 28 from genro_asgi_multiworker_spa.global_store import GlobalStoreLock29 30 31 async def test_the_turn_records_its_request_worker_and_key() -> None:32     lock = GlobalStoreLock()33     await lock.acquire("standard_0001", "r1", "config")34 35     assert lock.holds("r1") and lock.held_by("standard_0001") and lock.holder_key == "config"36     assert not lock.holds("r2") and not lock.held_by("standard_0002")37 38     lock.release()39     assert lock.holder is None and lock.holder_worker is None and lock.holder_key is None40     assert not lock.lock.locked()41 42 43 async def test_a_whole_store_turn_has_no_key() -> None:44     lock = GlobalStoreLock()45     await lock.acquire("standard_0001", "r1")46 47     assert lock.holder_key is None48     lock.release()49 50 51 async def test_a_release_for_a_turn_no_longer_in_force_is_told_apart() -> None:52     """The commander asks ``holds`` before it touches anything: a stale id says no."""53     lock = GlobalStoreLock()54     await lock.acquire("standard_0001", "current")55 56     assert lock.holds("stale") is False57     assert lock.holds("current") is True58     lock.release()