tests/spa/orchestration/test_orchestration_asgi_seam.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 worker hosts an ASGI application, and the legacy reaches it (#68 phase 4a).16 17 Contract tests. There is ONE seam on the worker, ``asgi_app``, and one road out18 of ``_serve_request``: ``hosted_app_seam``. A worker that hosts only a WSGI site19 takes the shortcut — it assigns ``wsgi_app`` and the core wraps it in a20 ``WsgiSeam`` — and a consumer whose own ASGI application must delegate some21 paths to a legacy site builds that same ``WsgiSeam`` and calls it from its own22 router. The core knows no path prefixes (owner, 2026-09-06, form B).23 24 The shortcut is covered by the whole existing rig, which passes unchanged:25 ``test_orchestration_m2_e2e.py`` drives a real process whose site is assigned to26 ``wsgi_app``, and every story it tells now travels through the adapter.27 """28 29 from __future__ import annotations30 31 import asyncio32 from typing import Any33 34 import pytest35 36 from genro_asgi.exceptions import HTTPException37 from genro_asgi_multiworker_spa.environ import WsgiSeam38 from genro_asgi_multiworker_spa.orchestration import FreezeHandler, SpaWorker39 40 from .conftest import attach_wire41 42 CID = "cid-a"43 USER = "mario"44 45 46 def http_call(path: str = "/main", body: bytes = b"", **http: Any) -> dict[str, Any]:47 """The http CALL form as the front packs it."""48 call: dict[str, Any] = {49 "method": "GET",50 "path": path,51 "query_string": "who=mario",52 "headers": [["host", "site.example:8080"], ["cookie", f"spa_connection_id={CID}"]],53 "body": body,54 "client": ["10.0.0.9", 51234],55 "scheme": "https",56 "cid": CID,57 }58 call.update(http)59 return {"http": call, "identity": USER}60 61 62 def body_of(served: dict[str, Any]) -> bytes:63 """The answer's body, out of the wire form."""64 return served["body"]65 66 67 def headers_of(served: dict[str, Any]) -> dict[str, str]:68 return {name.lower(): value for name, value in served["headers"]}69 70 71 class XT_AsgiWorker(SpaWorker):72 """A worker hosting an ASGI application, the way a consumer builds one.73 74 The application is built HERE and given this worker (N22: the core writes no75 live object into a scope), so it can call the worker's own verbs while it76 serves — which is what a real consumer does with `new_connection`.77 """78 79 def __init__(self, name: str, **kwargs: Any) -> None:80 super().__init__(name, **kwargs)81 self.seen: list[dict[str, Any]] = []82 self.asgi_app = self.application83 84 async def application(self, scope: dict[str, Any], receive: Any, send: Any) -> None:85 """Say back what arrived, and register the connection while serving."""86 message = await receive()87 self.seen.append(dict(scope))88 if self.connection_register.get(CID) is None:89 self.new_connection(CID, user=scope["genro.identity"])90 answer = (91 f"{scope['method']} {scope['path']}?{scope['query_string'].decode()} "92 f"for {scope['genro.identity']} body={message['body'].decode()}"93 ).encode()94 await send(95 {96 "type": "http.response.start",97 "status": 201,98 "headers": [(b"content-type", b"text/plain"), (b"x-worker", self.name.encode())],99 }100 )101 await send({"type": "http.response.body", "body": answer})102 103 104 class XT_MixedWorker(SpaWorker):105 """A worker whose ASGI application delegates some paths to a legacy site.106 107 This is form B seen from the consumer's side: the router is the108 application's own, and `/legacy/...` goes to the WSGI callable through the109 core's adapter, built once with this worker in hand.110 """111 112 def __init__(self, name: str, **kwargs: Any) -> None:113 super().__init__(name, **kwargs)114 self.legacy = WsgiSeam(self.tiny_site, self)115 self.environs: list[dict[str, Any]] = []116 self.asgi_app = self.application117 118 def tiny_site(self, environ: dict[str, Any], start_response: Any) -> list[bytes]:119 """The legacy callable: it sets a cookie and redirects, like a real one."""120 self.environs.append(dict(environ))121 start_response(122 "302 Found",123 [124 ("Content-Type", "text/plain"),125 ("Set-Cookie", "legacy_session=abc; Path=/"),126 ("Location", "/legacy/next"),127 ],128 )129 return [f"legacy saw {environ['SCRIPT_NAME']}|{environ['PATH_INFO']}".encode()]130 131 async def application(self, scope: dict[str, Any], receive: Any, send: Any) -> None:132 if scope["path"].startswith("/legacy"):133 # The consumer's own choice: move the prefix into `root_path`, so134 # the legacy site keeps the view of its URLs it always had.135 delegated = {**scope, "root_path": "/legacy"}136 await self.legacy(delegated, receive, send)137 return138 await send({"type": "http.response.start", "status": 200, "headers": []})139 await send({"type": "http.response.body", "body": b"served by the new side"})140 141 142 class XT_WholePathWorker(XT_MixedWorker):143 """The same delegation, with the path left whole and no ``root_path``."""144 145 async def application(self, scope: dict[str, Any], receive: Any, send: Any) -> None:146 await self.legacy(scope, receive, send)147 148 149 def with_open_page(worker: SpaWorker, page_id: str = "p1") -> None:150 """Give the worker a page that already opened its channel.151 152 A message that names a page is refused unless that page said153 ``openchannel`` first (#68 phase 4), so a test that passes the page keys154 has to have a page.155 """156 worker.open_request_slot()157 worker.new_page(USER, page_id, connection_id=CID)158 worker.page_register.get(page_id)["wsx"] = True159 160 161 def worker_of(worker_class: type, tmp_path: Any) -> SpaWorker:162 worker = worker_class(163 "standard_0001", freeze_handler=FreezeHandler(tmp_path / "frozen_users")164 )165 attach_wire(worker)166 return worker167 168 169 @pytest.fixture170 def asgi_worker(tmp_path):171 worker = worker_of(XT_AsgiWorker, tmp_path)172 yield worker173 worker.exit_process()174 175 176 @pytest.fixture177 def mixed_worker(tmp_path):178 worker = worker_of(XT_MixedWorker, tmp_path)179 yield worker180 worker.exit_process()181 182 183 class TestAnAsgiApplicationAlone:184 async def test_the_request_reaches_it_as_a_scope(self, asgi_worker) -> None:185 served = await asgi_worker._serve_request(http_call(body=b"payload"))186 assert body_of(served) == b"GET /main?who=mario for mario body=payload"187 188 async def test_the_answer_carries_status_and_headers(self, asgi_worker) -> None:189 served = await asgi_worker._serve_request(http_call())190 assert served["status"] == 201191 assert headers_of(served)["x-worker"] == "standard_0001"192 193 async def test_the_scope_is_a_plausible_asgi_one(self, asgi_worker) -> None:194 await asgi_worker._serve_request(http_call())195 scope = asgi_worker.seen[0]196 assert scope["type"] == "http" and scope["asgi"]["version"] == "3.0"197 assert scope["scheme"] == "https" and scope["root_path"] == ""198 assert scope["server"] == ("site.example", 8080)199 assert scope["client"] == ("10.0.0.9", 51234)200 assert (b"cookie", f"spa_connection_id={CID}".encode()) in scope["headers"]201 202 async def test_the_identity_of_the_call_is_the_one_the_pool_routed_on(203 self, asgi_worker204 ) -> None:205 await asgi_worker._serve_request(http_call())206 assert asgi_worker.seen[0]["genro.identity"] == USER207 208 async def test_the_page_keys_are_absent_when_the_call_did_not_carry_them(209 self, asgi_worker210 ) -> None:211 await asgi_worker._serve_request(http_call())212 scope = asgi_worker.seen[0]213 assert "genro.page_id" not in scope and "genro.reply_path" not in scope214 215 async def test_the_page_keys_travel_when_the_call_carries_them(self, asgi_worker) -> None:216 with_open_page(asgi_worker)217 await asgi_worker._serve_request(218 http_call(page_id="p1", reply_path="/main/done")219 )220 scope = asgi_worker.seen[0]221 assert (scope["genro.page_id"], scope["genro.reply_path"]) == ("p1", "/main/done")222 223 async def test_the_application_registers_its_connection_while_serving(224 self, asgi_worker225 ) -> None:226 # The rows are the site's: the application reaches the worker because227 # whoever built it gave it the worker, not because a scope key did.228 await asgi_worker._serve_request(http_call())229 assert asgi_worker.connection_register.get(CID) is not None230 assert asgi_worker.request_slot.connection_id == CID231 232 233 class TestAnAsgiApplicationDelegatingToTheLegacy:234 async def test_the_legacy_answer_comes_back_whole(self, mixed_worker) -> None:235 served = await mixed_worker._serve_request(http_call(path="/legacy/invoices"))236 assert served["status"] == 302237 assert body_of(served) == b"legacy saw /legacy|/invoices"238 239 async def test_set_cookie_and_location_travel_untouched(self, mixed_worker) -> None:240 served = await mixed_worker._serve_request(http_call(path="/legacy/invoices"))241 headers = headers_of(served)242 assert headers["set-cookie"] == "legacy_session=abc; Path=/"243 assert headers["location"] == "/legacy/next"244 245 async def test_the_legacy_sees_the_urls_it_always_saw(self, mixed_worker) -> None:246 await mixed_worker._serve_request(http_call(path="/legacy/invoices"))247 environ = mixed_worker.environs[0]248 assert (environ["SCRIPT_NAME"], environ["PATH_INFO"]) == ("/legacy", "/invoices")249 250 async def test_the_legacy_gets_the_facts_of_the_request(self, mixed_worker) -> None:251 await mixed_worker._serve_request(http_call(path="/legacy/x", body=b"sent"))252 environ = mixed_worker.environs[0]253 assert environ["REQUEST_METHOD"] == "GET"254 assert environ["QUERY_STRING"] == "who=mario"255 assert environ["wsgi.input"].read() == b"sent"256 assert environ["HTTP_COOKIE"] == f"spa_connection_id={CID}"257 assert environ["genro.identity"] == USER258 assert (environ["SERVER_NAME"], environ["SERVER_PORT"]) == ("site.example", "8080")259 260 async def test_the_new_side_is_served_by_the_application_itself(self, mixed_worker) -> None:261 served = await mixed_worker._serve_request(http_call(path="/v2/orders"))262 assert body_of(served) == b"served by the new side"263 assert mixed_worker.environs == []264 265 async def test_a_router_that_leaves_the_path_whole_is_served_too(self, tmp_path) -> None:266 worker = worker_of(XT_WholePathWorker, tmp_path)267 try:268 await worker._serve_request(http_call(path="/legacy/invoices"))269 environ = worker.environs[0]270 assert (environ["SCRIPT_NAME"], environ["PATH_INFO"]) == ("", "/legacy/invoices")271 finally:272 worker.exit_process()273 274 275 class TestTheSeamAWorkerDeclares:276 def test_both_seams_assigned_is_refused_by_name(self, asgi_worker) -> None:277 asgi_worker.wsgi_app = lambda environ, start_response: [b""]278 with pytest.raises(RuntimeError, match="both assigned"):279 asgi_worker.hosted_app_seam280 281 def test_a_worker_that_hosts_nothing_says_so(self, tmp_path) -> None:282 worker = worker_of(SpaWorker, tmp_path)283 try:284 with pytest.raises(RuntimeError, match="hosts no application"):285 worker.hosted_app_seam286 finally:287 worker.exit_process()288 289 def test_the_shortcut_is_served_through_the_adapter(self, tmp_path) -> None:290 worker = worker_of(SpaWorker, tmp_path)291 try:292 worker.wsgi_app = lambda environ, start_response: [b""]293 seam = worker.hosted_app_seam294 assert isinstance(seam.asgi_app, WsgiSeam)295 assert seam.asgi_app.worker is worker296 finally:297 worker.exit_process()298 299 300 class TestSynchronousWorkInsideARequest:301 async def test_run_sync_runs_in_the_slot_of_its_call(self, tmp_path) -> None:302 # What the application announces from a pool thread rides the reply of303 # the CALL it is serving: the slot follows the work onto the thread.304 class XT_SyncWorker(XT_AsgiWorker):305 async def application(self, scope, receive, send) -> None:306 await receive()307 slot_name = await self.run_sync(self.on_the_thread)308 await send({"type": "http.response.start", "status": 200, "headers": []})309 await send({"type": "http.response.body", "body": slot_name.encode()})310 311 def on_the_thread(self) -> str:312 self.new_connection(CID, user=USER)313 return type(self.request_slot).__name__314 315 worker = worker_of(XT_SyncWorker, tmp_path)316 try:317 worker.open_request_slot()318 served = await worker._serve_request(http_call())319 assert body_of(served) == b"RequestSlot"320 # The birth of a connection under an unseen user announces the321 # user first: both rode the slot the pool thread found.322 assert [event["op"] for event in worker.worker_events] == [323 "new_user",324 "new_connection",325 ]326 finally:327 worker.exit_process()328 329 330 class TestTheFrozenUserComesBack:331 async def test_a_frozen_user_is_adopted_and_the_application_serves_him(332 self, worker_commander_lane, tmp_path333 ) -> None:334 # The whole road: the user is parked in the deposit, and the next335 # request wakes him — the ASGI application serves it like any other.336 lane = worker_commander_lane337 await lane.verb("new_connection", CID, user=USER)338 await lane.worker.freeze_designated_user(USER)339 assert lane.worker.user_register.get(USER) is None340 341 served_by = XT_AsgiWorker(342 "standard_0001", freeze_handler=lane.worker.freeze_handler343 )344 attach_wire(served_by)345 try:346 served = await served_by._serve_request(347 {**http_call(), "user_frozen": True}348 )349 assert body_of(served).startswith(b"GET /main")350 assert served_by.user_register.get(USER) is not None351 finally:352 served_by.exit_process()353 354 355 class TestTheWorkerThatHostsOnlyWsgi:356 async def test_the_shortcut_serves_the_request_end_to_end(self, tmp_path) -> None:357 # The bridge's own shape, unchanged: assign `wsgi_app`, and the core358 # takes it through the adapter without the consumer knowing.359 seen: list[dict[str, Any]] = []360 361 def site(environ: dict[str, Any], start_response: Any) -> list[bytes]:362 seen.append(dict(environ))363 start_response("200 OK", [("Content-Type", "text/plain")])364 return [b"the legacy answered"]365 366 worker = worker_of(SpaWorker, tmp_path)367 try:368 worker.wsgi_app = site369 served = await worker._serve_request(http_call())370 assert body_of(served) == b"the legacy answered"371 assert seen[0]["PATH_INFO"] == "/main"372 assert seen[0]["genro.identity"] == USER373 finally:374 worker.exit_process()375 376 async def test_the_body_of_a_post_reaches_the_legacy(self, tmp_path) -> None:377 def site(environ: dict[str, Any], start_response: Any) -> list[bytes]:378 sent = environ["wsgi.input"].read()379 start_response("200 OK", [("Content-Type", "text/plain")])380 return [sent]381 382 worker = worker_of(SpaWorker, tmp_path)383 try:384 worker.wsgi_app = site385 served = await worker._serve_request(http_call(method="POST", body=b"a=1&b=2"))386 assert body_of(served) == b"a=1&b=2"387 finally:388 worker.exit_process()389 390 391 class TestTheContractOfTheTwoSeams:392 async def test_a_second_read_of_the_body_says_the_client_is_gone(self, tmp_path) -> None:393 # The ASGI contract after the body: an application that reads twice394 # must not hang waiting for a chunk that will never come.395 read: list[str] = []396 397 class XT_TwiceWorker(SpaWorker):398 def __init__(self, name: str, **kwargs: Any) -> None:399 super().__init__(name, **kwargs)400 self.asgi_app = self.application401 402 async def application(self, scope, receive, send) -> None:403 read.append((await receive())["type"])404 read.append((await receive())["type"])405 await send({"type": "http.response.start", "status": 200, "headers": []})406 await send({"type": "http.response.body", "body": b""})407 408 worker = worker_of(XT_TwiceWorker, tmp_path)409 try:410 await worker._serve_request(http_call())411 assert read == ["http.request", "http.disconnect"]412 finally:413 worker.exit_process()414 415 async def test_the_legacy_iterable_is_closed_as_the_spec_requires(self, tmp_path) -> None:416 closed: list[bool] = []417 418 class XT_ClosingBody:419 def __iter__(self):420 return iter([b"done"])421 422 def close(self) -> None:423 closed.append(True)424 425 def site(environ: dict[str, Any], start_response: Any) -> Any:426 start_response("200 OK", [])427 return XT_ClosingBody()428 429 worker = worker_of(SpaWorker, tmp_path)430 try:431 worker.wsgi_app = site432 served = await worker._serve_request(http_call())433 assert body_of(served) == b"done" and closed == [True]434 finally:435 worker.exit_process()436 437 async def test_the_deprecated_write_leads_the_body(self, tmp_path) -> None:438 # PEP 3333: what `write` put down comes before what the iterable yields.439 def site(environ: dict[str, Any], start_response: Any) -> list[bytes]:440 write = start_response("200 OK", [])441 write(b"first ")442 return [b"then"]443 444 worker = worker_of(SpaWorker, tmp_path)445 try:446 worker.wsgi_app = site447 served = await worker._serve_request(http_call())448 assert body_of(served) == b"first then"449 finally:450 worker.exit_process()451 452 async def test_the_client_address_reaches_the_legacy(self, tmp_path) -> None:453 def site(environ: dict[str, Any], start_response: Any) -> list[bytes]:454 start_response("200 OK", [])455 return [f"{environ['REMOTE_ADDR']}:{environ['REMOTE_PORT']}".encode()]456 457 worker = worker_of(SpaWorker, tmp_path)458 try:459 worker.wsgi_app = site460 served = await worker._serve_request(http_call())461 assert body_of(served) == b"10.0.0.9:51234"462 finally:463 worker.exit_process()464 465 async def test_repeated_headers_reach_the_legacy_the_way_pep_3333_wants(466 self, tmp_path467 ) -> None:468 # Duplicates are joined by a comma, cookies by "; " — a comma there469 # would fuse two cookies into one mangled value.470 def site(environ: dict[str, Any], start_response: Any) -> list[bytes]:471 start_response("200 OK", [])472 return [f"{environ['HTTP_X_TWICE']}|{environ['HTTP_COOKIE']}".encode()]473 474 worker = worker_of(SpaWorker, tmp_path)475 try:476 worker.wsgi_app = site477 served = await worker._serve_request(478 http_call(479 headers=[480 ["x-twice", "one"],481 ["x-twice", "two"],482 ["cookie", "a=1"],483 ["cookie", "b=2"],484 ]485 )486 )487 assert body_of(served) == b"one,two|a=1; b=2"488 finally:489 worker.exit_process()490 491 async def test_a_delegation_after_the_body_was_read_sends_an_empty_one(492 self, tmp_path493 ) -> None:494 # The application read the body itself and only then delegated: the495 # adapter finds the disconnect, and the legacy gets no body rather than496 # waiting for one.497 class XT_LateWorker(SpaWorker):498 def __init__(self, name: str, **kwargs: Any) -> None:499 super().__init__(name, **kwargs)500 self.legacy = WsgiSeam(self.site, self)501 self.asgi_app = self.application502 self.seen_body: list[bytes] = []503 504 def site(self, environ: dict[str, Any], start_response: Any) -> list[bytes]:505 self.seen_body.append(environ["wsgi.input"].read())506 start_response("200 OK", [])507 return [b"ok"]508 509 async def application(self, scope, receive, send) -> None:510 await receive()511 await self.legacy(scope, receive, send)512 513 worker = worker_of(XT_LateWorker, tmp_path)514 try:515 await worker._serve_request(http_call(body=b"read by the app"))516 assert worker.seen_body == [b""]517 finally:518 worker.exit_process()519 520 async def test_the_page_keys_reach_the_legacy_too(self, mixed_worker) -> None:521 with_open_page(mixed_worker)522 await mixed_worker._serve_request(523 http_call(path="/legacy/x", page_id="p1", reply_path="/main/done")524 )525 environ = mixed_worker.environs[0]526 assert (environ["genro.page_id"], environ["genro.reply_path"]) == ("p1", "/main/done")527 528 529 class TestAMessageOfAPage:530 async def test_a_page_that_never_opened_its_channel_is_refused(self, asgi_worker) -> None:531 # A refusal of the CLIENT, with a status of its own: the browser reads532 # 409 and these words, not the 502 of a site that broke (#70 C).533 asgi_worker.open_request_slot()534 asgi_worker.new_page(USER, "p1", connection_id=CID)535 with pytest.raises(HTTPException, match="no open channel") as refused:536 await asgi_worker._serve_request(http_call(page_id="p1"))537 assert refused.value.status == 409538 539 async def test_a_page_this_worker_never_saw_is_refused(self, asgi_worker) -> None:540 with pytest.raises(HTTPException, match="no open channel"):541 await asgi_worker._serve_request(http_call(page_id="never-born"))542 543 async def test_a_request_that_names_no_page_passes(self, asgi_worker) -> None:544 # An ordinary http request of the site names no page and is untouched.545 served = await asgi_worker._serve_request(http_call())546 assert served["status"] == 201547 548 549 class TestWhatTheSeamRefuses:550 async def test_an_application_that_answers_nothing_is_an_explicit_error(551 self, tmp_path552 ) -> None:553 class XT_SilentWorker(SpaWorker):554 def __init__(self, name: str, **kwargs: Any) -> None:555 super().__init__(name, **kwargs)556 self.asgi_app = self.application557 558 async def application(self, scope, receive, send) -> None:559 await asyncio.sleep(0)560 561 worker = worker_of(XT_SilentWorker, tmp_path)562 try:563 with pytest.raises(RuntimeError, match="did not start a response"):564 await worker._serve_request(http_call())565 finally:566 worker.exit_process()