tests/spa/orchestration/test_contract_phase2_site_verbs.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 """Phase 2 contract: the lifecycle verbs in the forms the site calls.16 17 Derived from the pre_refactoring worker's own verb signatures18 (``src/genro_asgi/spa/worker.py``: ``new_connection`` :1917, ``new_page`` :1999,19 ``drop_page`` :2078, ``drop_connection`` :2084) and from20 ``tests/test_spa_stores.py`` (the login relabels in place). The identity is the21 first positional argument, everything else travels by name — exactly how22 ``siteregister_client.py`` calls them. The ``cascade`` the site passes on23 ``drop_page`` is added by the bridge and never reaches the worker.24 25 ``add_user`` / ``add_connection`` / ``add_page`` stay what they are: the26 mutators that announce upward. The new verbs are facades in the site's shapes27 over those same mutators, so the announcements keep rising unchanged.28 """29 30 from __future__ import annotations31 32 import pytest33 from genro_bag import Bag34 35 from genro_asgi_multiworker_spa.orchestration import FreezeHandler, SpaWorker36 from genro_asgi_multiworker_spa.orchestration.spa_worker import GUEST_PREFIX37 38 WORKER_NAME = "standard_0001"39 40 41 @pytest.fixture42 def worker(tmp_path):43     deposit = FreezeHandler(tmp_path / "frozen_users")44     worker = SpaWorker(WORKER_NAME, freeze_handler=deposit, deposit_lock_retry_interval=0.01)45     worker.open_request_slot()46     return worker47 48 49 def announced(worker):50     return [event["op"] for event in worker.worker_events]51 52 53 # ----------------------------------------------------------------------54 # new_connection / new_page: identity first, the chain built from below55 # ----------------------------------------------------------------------56 57 58 def test_new_connection_names_the_anonymous_guest_and_announces_the_cascade(worker):59     entry = worker.new_connection("sess-1")60 61     assert entry["user"] == GUEST_PREFIX + "sess-1"62     assert announced(worker) == ["new_user", "new_connection"]63     assert worker.connection_register.get("sess-1") is not None64 65 66 def test_new_page_builds_the_whole_chain_and_announces_it_in_order(worker):67     entry = worker.new_page("alice", page_id="p1", connection_id="s1")68 69     assert entry["connection_id"] == "s1"70     assert announced(worker) == ["new_user", "new_connection", "new_page"]71     page_event = worker.worker_events[-1]72     assert page_event["page_id"] == "p1"73     assert page_event["connection_id"] == "s1"74     assert page_event["user"] == "alice"75 76 77 def test_new_page_keeps_the_fields_the_site_writes(worker):78     """``start_ts`` is the site's own stamp: the row carries it verbatim."""79     worker.new_page("alice", page_id="p1", connection_id="s1", start_ts=1000.0)80 81     assert worker.page_register.get("p1")["start_ts"] == 1000.082 83 84 def test_a_second_page_on_a_known_connection_announces_only_itself(worker):85     worker.new_page("alice", page_id="p1", connection_id="s1")86     worker.worker_events.clear()87 88     worker.new_page("alice", page_id="p2", connection_id="s1")89 90     assert announced(worker) == ["new_page"]91 92 93 def test_the_new_rows_carry_the_data_plane_from_birth(worker):94     worker.new_page("alice", page_id="p1", connection_id="s1")95 96     page = worker.page_register.get("p1")97     assert isinstance(page["store"], Bag)98 99 100 # ----------------------------------------------------------------------101 # drop_page / drop_connection: the pre_refactoring forms, cascade announced102 # ----------------------------------------------------------------------103 104 105 def test_drop_page_takes_the_page_and_announces_what_its_departure_empties(worker):106     worker.new_page("alice", page_id="p1", connection_id="s1")107     worker.worker_events.clear()108 109     worker.drop_page("alice", "p1")110 111     ops = announced(worker)112     assert ops[0] == "drop_page"113     assert "drop_connection" in ops114     assert "drop_user" in ops115     assert worker.page_register.get("p1") is None116     assert worker.connection_register.get("s1") is None117     assert worker.user_register.get("alice") is None118 119 120 def test_drop_page_leaves_the_surviving_sibling_alone(worker):121     worker.new_page("alice", page_id="p1", connection_id="s1")122     worker.new_page("alice", page_id="p2", connection_id="s1")123     worker.worker_events.clear()124 125     worker.drop_page("alice", "p1")126 127     assert announced(worker) == ["drop_page"]128     assert worker.page_register.get("p2") is not None129     assert worker.connection_register.get("s1") is not None130 131 132 def test_drop_connection_demolishes_pages_first_then_connection_then_user(worker):133     worker.new_page("alice", page_id="p1", connection_id="s1")134     worker.new_page("alice", page_id="p2", connection_id="s1")135     worker.worker_events.clear()136 137     worker.drop_connection("alice", connection_id="s1")138 139     ops = announced(worker)140     assert ops[-1] == "drop_user"141     assert "drop_connection" in ops142     assert worker.page_register.get("p1") is None143     assert worker.page_register.get("p2") is None144     assert worker.connection_register.get("s1") is None145     assert worker.user_register.get("alice") is None146 147 148 def test_drop_connection_on_an_unknown_connection_is_an_explicit_error(worker):149     with pytest.raises(KeyError, match="ghost"):150         worker.drop_connection("alice", connection_id="ghost")151 152 153 # ----------------------------------------------------------------------154 # The login: change_connection_user keeps its compatible form155 # ----------------------------------------------------------------------156 157 158 def test_the_login_relabels_the_connection_in_place(worker):159     worker.new_page("guest_sess-1", page_id="p1", connection_id="sess-1")160     connection = worker.connection_register.get("sess-1")161 162     worker.change_connection_user("sess-1", user="alice")163 164     assert worker.connection_register.get("sess-1") is connection165     assert connection["user"] == "alice"166     assert worker.user_register.get("guest_sess-1") is None167     assert worker.user_register.get("alice") is not None168 169 170 def test_the_guest_store_follows_its_first_real_identity(worker):171     worker.new_page("guest_sess-1", page_id="p1", connection_id="sess-1")172     guest_store = worker.user_register.get("guest_sess-1")["store"]173     guest_store["draft"] = "half typed"174 175     worker.change_connection_user("sess-1", user="alice")176 177     entry = worker.user_register.get("alice")178     assert entry["store"] is guest_store179     assert entry["store"]["draft"] == "half typed"180 181 182 # ----------------------------------------------------------------------183 # The old mutators stay: no collision with the site's verbs184 # ----------------------------------------------------------------------185 186 187 def test_add_user_keeps_its_own_form_beside_the_site_verbs(worker):188     item = worker.add_user("alice", custom="kept")189 190     assert item["custom"] == "kept"191     assert announced(worker) == ["new_user"]192 193 194 def test_refresh_chain_answers_the_sites_single_argument_call(worker):195     worker.new_page("alice", page_id="p1", connection_id="s1")196 197     instant = worker.refresh_chain("p1")198 199     assert worker.user_register.get("alice")["last_refresh_ts"] == instant