tests/spa/orchestration/test_orchestration_worker_events_channels.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 two channels of the worker events (owner, 2026-09-04; genro-asgi #60).16 17 An event born while a CALL is being served has an effect on the commander, and it18 returns WITH the REPLY of that CALL — never with the REPLY of another CALL that19 happened to finish first, never through a queue shared by the whole worker. An20 event that no CALL produced — the transfer cycle of a quit — goes up with a CALL21 of the worker's own. The login is the first case this contract protects: a ping22 ending while ``doLogin`` still runs must evict nobody.23 24 The worker here speaks on a stub wire that keeps every frame it writes and answers25 the worker's own CALLs, so a test reads exactly what each REPLY and each CALL26 carried. The deposit is a real ``FreezeHandler`` on a real directory.27 """28 29 from __future__ import annotations30 31 import asyncio32 import threading33 from typing import Any34 35 import pytest36 37 from tests.spa.orchestration.frame_helpers import control_frame, read_control38 39 from genro_asgi.channel.frame import Frame40 from genro_asgi_multiworker_spa.orchestration import FreezeHandler, SpaWorker41 from genro_asgi_multiworker_spa.orchestration.worker_connector import (42     CALL_METHOD,43     ENVELOPE_SLOT_WORKER_EVENTS,44     ENVELOPE_SLOT_WORKER_SNAPSHOT,45 )46 from genro_asgi_multiworker_spa.orchestration.worker_handler import ANNOUNCE_OP_PATH, PING_OP_PATH47 48 from .conftest import XT_Wire, wait_for49 50 WORKER_NAME = "standard_0001"51 CID = "a1b2"52 53 54 class XT_SiteWorker(SpaWorker):55     """A worker hosting a two-path site: ``/visit`` registers, ``/login`` logs in.56 57     Both paths block on ``gate`` after acting on the registers, so a test can let58     another CALL finish while the request is still being served.59     """60 61     def __init__(self, name: str, **kwargs: Any) -> None:62         super().__init__(name, **kwargs)63         self.wsgi_app = self.site64         self.gate = threading.Event()65         self.acted = threading.Event()66 67     def site(self, environ: dict[str, Any], start_response: Any) -> list[bytes]:68         if self.connection_register.get(CID) is None:69             self.add_connection(CID)70         if environ["PATH_INFO"] == "/login":71             self.change_connection_user(CID, "mario")72         self.acted.set()73         self.gate.wait(timeout=10)74         start_response("200 OK", [("Content-Type", "text/plain")])75         return [b"served"]76 77 78 def http_call(path: str) -> dict[str, Any]:79     """The http CALL form as the front packs an anonymous visit."""80     return {81         "http": {82             "method": "GET",83             "path": path,84             "query_string": "",85             "headers": [["host", "site.example:8080"]],86             "body": "",87             "cid": CID,88         },89         "identity": None,90     }91 92 93 def events_of(frame: Frame) -> list[str]:94     return [event["op"] for event in (read_control(frame) or {}).get(ENVELOPE_SLOT_WORKER_EVENTS) or ()]95 96 97 @pytest.fixture98 def deposit(tmp_path):99     return FreezeHandler(tmp_path / "frozen_users")100 101 102 @pytest.fixture103 async def worker(deposit):104     worker = XT_SiteWorker(WORKER_NAME, freeze_handler=deposit, deposit_lock_retry_interval=0.01)105     wire = XT_Wire(worker)106     worker.attach_stream(wire)107     yield worker108     worker.gate.set()109 110 111 async def serve(worker: SpaWorker, path: str, data: Any) -> Frame:112     """Hand the worker one CALL the way the wire does; return the frame, not the answer."""113     frame = control_frame(method=CALL_METHOD, path=path, data=data)114     worker.handle_frame(frame)115     return frame116 117 118 # -- channel one: the REPLY of the CALL that caused the event --119 120 121 async def test_a_reply_carries_only_the_events_its_own_call_caused(worker):122     wire = worker.stream123     visit = await serve(worker, "/http", http_call("/visit"))124     await asyncio.get_running_loop().run_in_executor(None, worker.acted.wait)125 126     ping = await serve(worker, PING_OP_PATH, {})127     await wait_for(lambda: wire.reply_to(ping.id) is not None)128     assert events_of(wire.reply_to(ping.id)) == []129 130     worker.gate.set()131     await wait_for(lambda: wire.reply_to(visit.id) is not None)132     assert events_of(wire.reply_to(visit.id)) == ["new_user", "new_connection"]133 134 135 def test_an_event_outside_any_call_is_refused(deposit):136     worker = SpaWorker(WORKER_NAME, freeze_handler=deposit)137 138     with pytest.raises(RuntimeError):139         worker.add_connection(CID)140 141 142 # -- the login: its tail belongs to its own request --143 144 145 async def test_a_ping_ending_during_a_login_evicts_nobody(worker, deposit):146     wire = worker.stream147     login = await serve(worker, "/http", http_call("/login"))148     await asyncio.get_running_loop().run_in_executor(None, worker.acted.wait)149 150     ping = await serve(worker, PING_OP_PATH, {})151     await wait_for(lambda: wire.reply_to(ping.id) is not None)152     assert events_of(wire.reply_to(ping.id)) == []153     assert worker.connection_register.get(CID)["user"] == "mario"154     assert deposit.read_connection_register_item("mario", CID) is None155 156     worker.gate.set()157     await wait_for(lambda: wire.reply_to(login.id) is not None)158     assert events_of(wire.reply_to(login.id)) == [159         "new_user",160         "new_connection",161         "connection_user_changed",162         "user_rows_released",163     ]164     assert worker.connection_register.get(CID) is None165     assert deposit.read_connection_register_item("mario", CID) is not None166 167 168 # -- channel two: a CALL of the worker's own for what no CALL produced --169 170 171 async def test_the_transfer_cycle_announces_each_freeze_with_a_call_of_its_own(worker, deposit):172     wire = worker.stream173     worker.open_request_slot()174     worker.add_connection(CID)175     worker.change_connection_user(CID, "mario")176     worker.worker_events.clear()177 178     worker.plan_transfers(transfer_users=["mario"])179     worker._transfers_start_ts = 0.0180     await worker.execute_transfers()181 182     announced = wire.calls(ANNOUNCE_OP_PATH)183     assert [events_of(frame) for frame in announced] == [["user_frozen"]]184     assert ENVELOPE_SLOT_WORKER_SNAPSHOT in read_control(announced[0])185     assert wire.replies() == []186     assert deposit.read_user_register_item("mario") is not None187 188 189 async def test_the_vertex_folds_an_announcement_like_a_reply(worker_commander_lane):190     lane = worker_commander_lane191     vertex = lane.commander192     await lane.open_request()193     await lane.verb("add_connection", CID)194     await lane.verb("change_connection_user", CID, "mario")195     await lane.announce()196     await wait_for(lambda: vertex.resolve_user(CID) == "mario")197     assert "mario" in lane.worker_handler.hosted_users198 199     await lane.verb("plan_transfers", transfer_users=["mario"])200     lane.worker._transfers_start_ts = 0.0201     await lane.worker.execute_transfers()202 203     await wait_for(lambda: vertex.user_map["mario"]["frozen"] is True)204     assert lane.worker_handler.group_handler.user_worker_map["mario"] is None205     assert "mario" not in lane.worker_handler.hosted_users206 207 208 async def test_reply_encoding_failure_answers_and_keeps_its_events(worker):209     """An invalid endpoint result must not consume the slot and strand the caller."""210     async def answer(frame):211         worker.add_connection(CID)212         await worker.send_reply(frame, result=object())213 214     worker.answer_call = answer215     frame = control_frame(method=CALL_METHOD, path="/invalid-result", data={})216     await worker._guarded_call(frame)217     reply = worker.stream.reply_to(frame.id)218     assert "invalid control payload" in reply.info["error"]219     assert events_of(reply) == ["new_user", "new_connection"]220     assert worker._request_slot_var.get() is None