tests/spa/orchestration/test_contract_consumer_seams.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 """Contract: the seams a consumer overrides to pair its own data with the core (#59, block 3).16 17 The core knows four opaque data: the vertex's, the user's, the connection's,18 the page's. Every row is a ``dict`` of the registry's row class — ``UserRow``,19 ``ConnectionRow``, ``PageRow`` — so ``row["field"]`` reads everywhere as20 before, and the class carries what the core used to hard-code about the row:21 the fields it is born with (``default_fields``), the ones the parcel leaves22 behind (``fields_left_behind``), the ones that travel but are put back after23 the birth (``fields_replayed``, ``replay_fields``), and what the birth24 announces (``announcement_fields``). A consumer subclasses the row and names it25 on its registry (``page_row_class``); the worker asks the row and knows nothing26 of the fields. Beside the rows: the request slot comes from27 ``SpaWorker.build_request_slot``, the vertex's data from28 ``SpaCommander.new_global_store`` (a dict, whose values are the consumer's own29 types), every served request ends in30 ``SpaWorker.on_request_served``, and a process that has just presented itself31 is told to the vertex through ``SpaCommander.on_worker_presented`` — the seam32 the source filter of a hosted site is pushed from, once per process, on the33 envelope that carries the presentation and no other. The last layer of the34 envelope chain is the commander's ``envelope_handler`` property: a consumer35 returns its subclass of ``CommanderEnvelopeHandler`` and reads, in an36 ``on_<op>`` that calls the core's, what a worker event carries for it.37 """38 39 from __future__ import annotations40 41 import asyncio42 from concurrent.futures import ThreadPoolExecutor43 from typing import Any44 45 import pytest46 from genro_bag import Bag47 from genro_tytx import from_tytx, to_tytx48 49 from genro_asgi_multiworker_spa import RegisterRegistry50 from genro_asgi_multiworker_spa.orchestration import FreezeHandler, GroupHandler, SpaCommander, SpaWorker51 from genro_asgi_multiworker_spa.orchestration.envelope_handler import CommanderEnvelopeHandler52 from genro_asgi_multiworker_spa.orchestration.spa_worker import RequestSlot53 from genro_asgi_multiworker_spa.register_row import ConnectionRow, PageRow, UserRow54 55 from .conftest import XT_WorkerCommanderLane, attach_wire56 57 STORE_GET = "/commander/store/get"58 STORE_LOCK = "/commander/store/lock"59 STORE_UNLOCK = "/commander/store/unlock"60 61 62 class XT_PageRow(PageRow):63 """A consumer's page row: one field of its own in every category."""64 65 fields_left_behind = PageRow.fields_left_behind | {"xt_live"}66 fields_replayed = (*PageRow.fields_replayed, "xt_replayed", "xt_tables")67 68 def default_fields(self) -> dict[str, Any]:69 return {70 **super().default_fields(),71 "xt_marker": "born",72 "xt_live": object(),73 "xt_tables": set(),74 }75 76 def replay_fields(self, registry: Any, fields: dict[str, Any]) -> None:77 super().replay_fields(registry, fields)78 self["xt_replayed_seen"] = fields.get("xt_replayed")79 self["xt_tables"].update(fields.get("xt_tables", ()))80 81 def announcement_fields(self) -> dict[str, Any]:82 return {83 **super().announcement_fields(),84 "xt_marker": self["xt_marker"],85 "xt_tables": sorted(self["xt_tables"]),86 }87 88 89 class XT_Registry(RegisterRegistry):90 page_row_class = XT_PageRow91 92 93 class XT_Slot(RequestSlot):94 def __init__(self) -> None:95 super().__init__()96 self.xt_field = "mine"97 98 99 class XT_Worker(SpaWorker):100 """A consumer's worker: its registry, its slot, its end-of-request hook."""101 102 def __init__(self, *args: Any, **kwargs: Any) -> None:103 super().__init__(*args, **kwargs)104 self.served: list[str] = []105 106 def build_registry(self) -> RegisterRegistry:107 return XT_Registry()108 109 def build_request_slot(self) -> RequestSlot:110 return XT_Slot()111 112 def on_request_served(self) -> None:113 super().on_request_served()114 self.served.append(type(self.request_slot).__name__)115 116 117 class XT_EnvelopeHandler(CommanderEnvelopeHandler):118 """A consumer's last layer: it reads what a page's birth carries for it."""119 120 def on_new_page(self, worker_event: dict[str, Any]) -> None:121 super().on_new_page(worker_event)122 self.spa_commander.announced_pages.append(123 (worker_event["page_id"], worker_event["xt_tables"])124 )125 126 127 class XT_Commander(SpaCommander):128 """A consumer's commander: it builds the vertex data and applies the writes itself.129 130 The data stays a Bag: the grant carries the whole store TYTX-encoded down131 the lane, so whatever type a consumer chooses must be one the codec knows.132 """133 134 def __init__(self, *args: Any, **kwargs: Any) -> None:135 self.built: list[str] = []136 super().__init__(*args, **kwargs)137 self.applied: list[list[dict[str, Any]]] = []138 self.presented: list[str] = []139 self.announced_pages: list[tuple[str, list[str]]] = []140 141 @property142 def envelope_handler(self) -> CommanderEnvelopeHandler:143 return XT_EnvelopeHandler(self)144 145 def new_global_store(self) -> dict[str, Any]:146 self.built.append("xt")147 return {"seed": Bag({"a": 0})}148 149 def on_worker_presented(self, worker_handler: Any) -> None:150 self.presented.append(worker_handler.name)151 152 153 @pytest.fixture154 def worker(tmp_path):155 worker = XT_Worker("standard_0001", freeze_handler=FreezeHandler(tmp_path / "frozen_users"))156 attach_wire(worker)157 yield worker158 worker.exit_process()159 160 161 # ----------------------------------------------------------------------162 # The rows163 # ----------------------------------------------------------------------164 165 166 def test_rows_are_dicts_of_the_registrys_row_classes():167 registry = RegisterRegistry()168 page = registry.new_page("p1", user="mario", connection_id="c1")169 assert isinstance(page, PageRow) and isinstance(page, dict)170 assert isinstance(registry.connection_items.get("c1"), ConnectionRow)171 assert isinstance(registry.user_items.get("mario"), UserRow)172 assert page["register_item_id"] == "p1" and "item_lock" in page173 174 175 def test_a_row_born_with_fields_keeps_them_over_the_defaults():176 registry = RegisterRegistry()177 page = registry.new_page("p1", user="mario", connection_id="c1", last_refresh_ts=1.0)178 assert page["last_refresh_ts"] == 1.0179 180 181 def test_the_consumers_row_class_is_what_the_registry_builds(worker):182 worker.add_connection("c1", "mario")183 page = worker.add_page("p1", "c1")184 assert isinstance(page, XT_PageRow)185 assert page["xt_marker"] == "born"186 assert page["xt_tables"] == set()187 188 189 def test_the_parcel_leaves_behind_what_the_row_says_and_carries_the_rest(worker):190 worker.add_connection("c1", "mario")191 worker.add_page("p1", "c1")192 parcel = worker._connection_parcel("c1")193 page = parcel["pages"]["p1"]194 assert "xt_live" not in page and "item_lock" not in page195 assert "connection_id" not in page196 assert page["xt_marker"] == "born"197 198 199 def test_the_birth_announces_what_the_row_says(worker):200 worker.add_connection("c1", "mario")201 worker.add_page("p1", "c1")202 announced = [event for event in worker.request_slot.worker_events if event["op"] == "new_page"]203 assert announced[-1]["xt_marker"] == "born"204 assert announced[-1]["xt_tables"] == []205 206 207 def test_the_row_class_replays_what_travelled(worker):208 worker.add_connection("c1", "mario")209 page = worker.add_page("p1", "c1")210 page.replay_fields(worker.registry, {"xt_replayed": 7, "xt_tables": ["t"]})211 assert page["xt_replayed_seen"] == 7212 assert page["xt_tables"] == {"t"}213 214 215 # ----------------------------------------------------------------------216 # The slot, the vertex data, the end of the request217 # ----------------------------------------------------------------------218 219 220 async def test_every_request_slot_is_the_consumers(worker):221 assert isinstance(worker.open_request_slot(), XT_Slot)222 pool = ThreadPoolExecutor(max_workers=1)223 try:224 on_thread = await worker._run_in_pool(pool, lambda: type(worker.request_slot).__name__)225 finally:226 pool.shutdown(wait=True)227 assert on_thread == "XT_Slot"228 229 230 async def test_on_request_served_runs_after_every_request_failed_ones_included(worker):231 # Rewritten in phase 4a of #68: the stitching goes through232 # `hosted_app_seam` now, and `_serve_on_thread` is gone with the dict233 # entrance it served. The assertion is the one it always was — the hook234 # runs on a pool thread, in the request's own slot, whatever the site did.235 def wsgi_app(environ, start_response):236 start_response("200 OK", [("Content-Type", "text/plain")])237 return [b"ok"]238 239 def failing_app(environ, start_response):240 raise RuntimeError("the site failed")241 242 payload = {"http": {"method": "GET", "path": "/", "cid": "c1"}, "identity": "mario"}243 worker.wsgi_app = wsgi_app244 await worker._serve_request(payload)245 worker.wsgi_app = failing_app246 with pytest.raises(RuntimeError):247 await worker._serve_request(payload)248 assert worker.served == ["XT_Slot", "XT_Slot"]249 250 251 async def test_the_vertex_data_is_the_commanders_seam(short_root, tmp_path):252 # The consumer fills the dictionary at birth with values of its own type; a253 # turn's release publishes the complete value the worker sent, whatever it is.254 commander = XT_Commander(short_root / "frozen_users")255 assert commander.built == ["xt"]256 group = GroupHandler(257 commander,258 "standard",259 memory_concession_bytes=8 * 1024 * 1024 * 1024,260 instance_dir=short_root / "i",261 frozen_users_path=short_root / "frozen_users",262 entry_module="never.launched",263 )264 lane = XT_WorkerCommanderLane(commander, group, FreezeHandler(tmp_path / "frozen_users"))265 await lane.open()266 try:267 grant = await lane.worker.call(268 STORE_LOCK, {"worker": lane.worker_name, "request_id": "r1", "key": "seed"}269 )270 seed = from_tytx(grant["value"], "json")271 seed["a"] = 1272 reply = await lane.worker.call(273 STORE_UNLOCK, {"request_id": "r1", "apply": True, "value": to_tytx(seed, "json")}274 )275 finally:276 await lane.close()277 assert reply == {"applied": True}278 assert isinstance(commander.global_register["seed"], Bag)279 assert commander.global_register["seed"]["a"] == 1280 await asyncio.sleep(0)281 282 283 async def test_a_newborn_process_is_told_to_the_vertex_once(short_root, tmp_path):284 commander = XT_Commander(short_root / "frozen_users")285 group = GroupHandler(286 commander,287 "standard",288 memory_concession_bytes=8 * 1024 * 1024 * 1024,289 instance_dir=short_root / "i",290 frozen_users_path=short_root / "frozen_users",291 entry_module="never.launched",292 )293 lane = XT_WorkerCommanderLane(commander, group, FreezeHandler(tmp_path / "frozen_users"))294 await lane.open()295 try:296 assert commander.presented == [lane.worker_name]297 await lane.worker.call(STORE_GET, {"key": "a"})298 await lane.announce()299 finally:300 await lane.close()301 assert commander.presented == [lane.worker_name]302 303 304 async def test_the_consumers_envelope_layer_reads_what_a_pages_birth_carries(short_root, tmp_path):305 commander = XT_Commander(short_root / "frozen_users")306 group = GroupHandler(307 commander,308 "standard",309 memory_concession_bytes=8 * 1024 * 1024 * 1024,310 instance_dir=short_root / "i",311 frozen_users_path=short_root / "frozen_users",312 entry_module="never.launched",313 )314 lane = XT_WorkerCommanderLane(commander, group, FreezeHandler(tmp_path / "frozen_users"))315 await lane.open()316 try:317 # The lane's worker is the core's, so the event a consumer's row would318 # announce is sent by hand, on the worker's own channel.319 await lane.worker.announce_worker_events(320 [{"op": "new_page", "worker": lane.worker_name, "page_id": "p1", "connection_id": "c1", "xt_tables": ["t"]}]321 )322 finally:323 await lane.close()324 assert commander.page_connection_map == {"p1": "c1"}325 assert commander.announced_pages == [("p1", ["t"])]