tests/spa/orchestration/test_contract_phase7_worker_call_lane.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 7 contract: the worker-to-commander CALL lane.16 17 The redesign's foundation (registro 2026-08-20 §1): ``WorkerConnector`` learns18 the second dispatch branch — a CALL arriving FROM the child is served as a task19 and answered with a REPLY — and the worker learns to place a call and await its20 answer. The transport is already full duplex and frames carry ids: the21 conversations interleave without confusion. The channel doctrine always had the22 two sides («REPLY si risolve inline, CALL/EVENT si servono come task»); this23 phase finishes the half the connector implemented.24 25 Bindings (method names, the handler hook the commander exposes) are settled by26 the phase: skeletons state the behaviour, the executable shape is the phase's27 work. The `ENVELOPE_SLOT_*` renames ride this phase too — they touch the same28 file.29 """30 31 from __future__ import annotations32 33 import asyncio34 from pathlib import Path35 from typing import Any36 37 import pytest38 39 from genro_asgi.channel.frame import Frame40 from genro_asgi_multiworker_spa.orchestration import FreezeHandler, SpaWorker, WorkerConnector41 from genro_asgi_multiworker_spa.orchestration import spa_worker as spa_worker_module42 from genro_asgi_multiworker_spa.orchestration import worker_connector as worker_connector_module43 from genro_asgi_multiworker_spa.orchestration import envelope_handler as envelope_handler_module44 from genro_asgi_multiworker_spa.orchestration.worker_connector import (45 ENVELOPE_SLOT_PRESENTATION,46 ENVELOPE_SLOT_WORKER_EVENTS,47 ENVELOPE_SLOT_WORKER_SNAPSHOT,48 CommanderCallFailed,49 )50 51 WORKER_NAME = "standard_0001"52 UPWARD_OP_PATH = "/op/upward"53 SERVED_OP_PATH = "/op/ask"54 55 56 class XT_ServingHandler:57 """The WorkerHandler seen by its wire, with the parent half the lane needs.58 59 Args:60 answers: what the parent hands back, by routing key; a key it does not61 hold is a path the parent does not serve.62 """63 64 def __init__(self, answers: dict[str, Any] | None = None) -> None:65 self.name = WORKER_NAME66 self.answers = answers or {}67 self.served: list[tuple[str, Any]] = []68 self.gate: asyncio.Event | None = None69 70 def read_envelope(self, envelope: dict[str, Any]) -> dict[str, Any]:71 """Whatever arrives is taken; nothing of the store goes down these tests."""72 return {}73 74 def on_child_lost(self) -> None:75 pass76 77 async def serve_child_call(self, path: str, data: Any) -> Any:78 """Answer one call the child placed, parked on the gate while there is one."""79 self.served.append((path, data))80 if self.gate is not None:81 await self.gate.wait()82 return self.answers[path]83 84 85 class X_LaneWorker(SpaWorker):86 """A worker with one op that asks the parent before answering its own caller."""87 88 async def answer_call(self, frame: Frame) -> None:89 if frame.path != UPWARD_OP_PATH:90 await super().answer_call(frame)91 return92 answer = await self.call(SERVED_OP_PATH, {"who": self.name})93 await self.send_reply(frame, result=answer)94 95 96 class XT_LanePair:97 """A connector and a worker on one real UDS, both on this test's loop.98 99 Args:100 handler: the parent side of the wire, which serves the child's calls.101 socket_path: where to bind, under the short root the UDS cap needs.102 freeze_handler: the deposit the worker is built with.103 """104 105 def __init__(106 self, handler: XT_ServingHandler, socket_path: Path, freeze_handler: FreezeHandler107 ) -> None:108 self.handler = handler109 self.connector = WorkerConnector(handler, socket_path)110 self.worker = X_LaneWorker(WORKER_NAME, freeze_handler=freeze_handler)111 self._reader_task: asyncio.Task[None] | None = None112 113 async def open(self) -> None:114 """Bind, connect, present, and put the worker's read loop on the air."""115 await self.connector.start()116 reader, writer = await asyncio.open_unix_connection(str(self.connector.socket_path))117 self.worker.attach_stream(spa_worker_module.FrameStream(reader, writer))118 await self.worker.send_presentation({})119 self._reader_task = asyncio.create_task(self.worker.receive_frames())120 await self.connector.wait_connected()121 122 async def close(self) -> None:123 if self._reader_task is not None:124 self._reader_task.cancel()125 try:126 await self._reader_task127 except asyncio.CancelledError:128 pass129 self.worker.exit_process()130 await self.connector.stop()131 132 133 @pytest.fixture134 async def pair(short_root, tmp_path):135 handler = XT_ServingHandler({SERVED_OP_PATH: {"served": "answered"}})136 built = XT_LanePair(137 handler,138 short_root / "i" / f"{WORKER_NAME}.sock",139 FreezeHandler(tmp_path / "frozen_users"),140 )141 await built.open()142 yield built143 await built.close()144 145 146 async def test_a_worker_call_is_served_and_answered_while_the_parents_call_is_still_open(pair):147 # wf:contract: while the worker is serving a CALL the commander made (the148 # wf:contract: request is mid-flight), the worker places its own CALL on149 # wf:contract: the same wire; the connector serves it as a task, a handler150 # wf:contract: on the parent side answers, and the worker's awaited future151 # wf:contract: resolves with that REPLY — the parent's original CALL is152 # wf:contract: still pending throughout and completes normally afterwards.153 pair.handler.gate = asyncio.Event()154 downward = asyncio.create_task(pair.connector.call(UPWARD_OP_PATH))155 156 await wait_until(lambda: pair.handler.served)157 assert pair.handler.served == [(SERVED_OP_PATH, {"who": WORKER_NAME})]158 assert not downward.done()159 160 pair.handler.gate.set()161 reply = await asyncio.wait_for(downward, 5.0)162 163 assert reply["result"] == {"served": "answered"}164 assert ENVELOPE_SLOT_WORKER_EVENTS in reply165 166 167 async def test_two_worker_calls_interleave_by_frame_id(pair):168 # wf:contract: two CALLs placed by the worker without awaiting the first169 # wf:contract: resolve each with its own REPLY, matched by frame id, in170 # wf:contract: whatever order the parent answers.171 pair.handler.answers = {"/op/one": "first", "/op/two": "second"}172 gate = asyncio.Event()173 pair.handler.gate = gate174 175 one = asyncio.create_task(pair.worker.call("/op/one"))176 await wait_until(lambda: pair.handler.served)177 two = asyncio.create_task(pair.worker.call("/op/two"))178 await wait_until(lambda: len(pair.handler.served) == 2)179 180 # Both are on the wire and neither is answered: the parent is holding the181 # gate, so the order the answers come in is the gate's, not the calls'.182 assert not one.done() and not two.done()183 gate.set()184 185 assert await asyncio.wait_for(two, 5.0) == "second"186 assert await asyncio.wait_for(one, 5.0) == "first"187 188 189 async def test_a_worker_call_from_a_pool_thread_reaches_the_loop_and_returns(pair):190 # wf:contract: the request runs on a traffic-pool thread; the worker's191 # wf:contract: call() is reachable from that thread (hop onto the loop,192 # wf:contract: the pre_refactoring pattern of the global lock) and hands193 # wf:contract: the REPLY payload back to the calling thread.194 worker = pair.worker195 196 def on_the_pool_thread() -> Any:197 return worker.run_on_loop(worker.call(SERVED_OP_PATH, {"from": "the pool"}))198 199 answer = await asyncio.get_running_loop().run_in_executor(200 worker.traffic_pool, on_the_pool_thread201 )202 203 assert answer == {"served": "answered"}204 assert pair.handler.served == [(SERVED_OP_PATH, {"from": "the pool"})]205 206 207 async def test_a_call_the_parent_has_no_handler_for_answers_an_error_not_silence(pair):208 # wf:contract: a CALL path the commander does not serve comes back as an209 # wf:contract: error REPLY the worker can raise on — never a dropped frame210 # wf:contract: and never a warning-and-discard.211 with pytest.raises(CommanderCallFailed) as refusal:212 await asyncio.wait_for(pair.worker.call("/op/nothing_here"), 5.0)213 214 assert refusal.value.path == "/op/nothing_here"215 assert "KeyError" in refusal.value.cause216 217 # The wire is unharmed: the next call on a path the parent serves answers.218 assert await asyncio.wait_for(pair.worker.call(SERVED_OP_PATH), 5.0) == {"served": "answered"}219 220 221 def test_the_envelope_slot_constants_wear_the_family_prefix():222 # wf:contract: the surviving envelope slot constants are named223 # wf:contract: ENVELOPE_SLOT_WORKER_EVENTS, ENVELOPE_SLOT_WORKER_SNAPSHOT,224 # wf:contract: ENVELOPE_SLOT_PRESENTATION, live in worker_connector.py,225 # wf:contract: keep their wire values ("worker_events", "worker_snapshot",226 # wf:contract: "pid"), and no bare string literal writes those slots any227 # wf:contract: more (the two M2/M3 stray literals are gone).228 assert ENVELOPE_SLOT_WORKER_EVENTS == "worker_events"229 assert ENVELOPE_SLOT_WORKER_SNAPSHOT == "worker_snapshot"230 assert ENVELOPE_SLOT_PRESENTATION == "pid"231 for dead in ("WORKER_EVENTS_KEY", "WORKER_SNAPSHOT_KEY", "PRESENTATION_KEY"):232 assert not hasattr(worker_connector_module, dead)233 assert not hasattr(envelope_handler_module, dead)234 235 source = Path(spa_worker_module.__file__).read_text()236 assert '"worker_events"' not in source237 assert '"worker_snapshot"' not in source238 assert '{"pid": os.getpid()' not in source239 240 241 async def wait_until(condition, timeout: float = 5.0) -> None:242 """Spin until the condition holds — the frames of a lane land on the loop."""243 deadline = asyncio.get_running_loop().time() + timeout244 while not condition():245 if asyncio.get_running_loop().time() >= deadline:246 raise TimeoutError("the lane never got there")247 await asyncio.sleep(0.01)