tests/core/test_request_registry.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 """Request-registry tests (SPECIFICATION.md §4): two concurrent requests each16 see their OWN request via the instance-owned ContextVar (the ContextVar test);17 ``in_flight`` counts both at a rendezvous and returns to zero afterwards;18 ``snapshot()`` exposes the slotted record fields; registration is cleaned up19 even when the handler raises.20 21 Driven at the ASGI level (no uvicorn): two concurrent tasks call the server and22 an ``asyncio.Barrier`` holds both in-flight so the picture is observed23 simultaneously — deterministic, and it mirrors the other Phase 0 tests.24 """25 26 from __future__ import annotations27 28 import asyncio29 from typing import Any30 31 import pytest32 33 from genro_asgi import BaseApplication, BaseServer, MiddlewareMixin34 from genro_asgi.request_registry import RegisteredRequest35 from genro_asgi.server import QUITTING, REFUSED_RETRY_AFTER_SECONDS, RUNNING36 from genro_asgi.types import Receive, Scope, Send37 38 39 class RendezvousApp(BaseApplication):40 """Primary app recording the registry picture while blocked at a barrier.41 42 Constructor kwargs peeled here (cooperative chain): ``barrier`` (shared by43 both requests) and ``observed`` (a dict the app writes its picture into).44 """45 46 def __init__(self, **kwargs: Any) -> None:47 self.barrier: asyncio.Barrier = kwargs.pop("barrier")48 self.observed: dict[int, dict[str, Any]] = kwargs.pop("observed")49 super().__init__(**kwargs)50 51 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:52 registry = self.server.requests53 own = registry.current54 assert own is not None # inside a request the ContextVar is always set55 await self.barrier.wait() # both requests are registered past this point56 self.observed[own.request_id] = {57 "current_is_own": registry.current is own,58 "in_flight": registry.in_flight,59 "snapshot": registry.snapshot(),60 "path": own.path,61 "scope_type": own.scope_type,62 }63 await self.barrier.wait() # hold both here so neither unregisters early64 await send({"type": "http.response.start", "status": 200, "headers": []})65 await send({"type": "http.response.body", "body": b"ok"})66 67 68 class RaisingApp(BaseApplication):69 """Primary app that always raises — to test cleanup on the error path."""70 71 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:72 raise RuntimeError("boom")73 74 75 class CleanupThenRaiseApp(BaseApplication):76 """Registers a request cleanup on ``current`` and then raises.77 78 Constructor kwarg peeled here: ``ran`` — a list the cleanup appends to, so79 the test can assert the cleanup ran despite the handler raising.80 """81 82 def __init__(self, **kwargs: Any) -> None:83 self.ran: list[str] = kwargs.pop("ran")84 super().__init__(**kwargs)85 86 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:87 current = self.server.requests.current88 assert current is not None89 current.add_cleanup(lambda: self.ran.append("cleanup"))90 raise RuntimeError("boom")91 92 93 async def drive(server: BaseServer, path: str) -> None:94 """Drive one http request through the server at the ASGI level."""95 96 async def receive() -> dict[str, object]:97 return {"type": "http.request"}98 99 async def send(message: dict[str, object]) -> None:100 pass101 102 await server({"type": "http", "path": path}, receive, send)103 104 105 class TestRegisteredRequest:106 def test_record_is_slotted_and_exposes_its_fields(self) -> None:107 item = RegisteredRequest(1, "http", "/x")108 assert item.request_id == 1109 assert item.scope_type == "http"110 assert item.path == "/x"111 assert isinstance(item.started_at, float)112 assert not hasattr(item, "__dict__") # D18: slotted, no per-instance dict113 114 115 class TestEmptyRegistry:116 def test_fresh_registry_is_empty(self) -> None:117 server = BaseServer(applications=[BaseApplication(mount="")])118 assert server.requests.current is None119 assert server.requests.in_flight == 0120 assert server.requests.snapshot() == []121 122 123 class TestConcurrentRequests:124 async def test_each_request_sees_its_own_current_and_in_flight_counts_both(125 self,126 ) -> None:127 barrier = asyncio.Barrier(2)128 observed: dict[int, dict[str, Any]] = {}129 server = BaseServer(applications=[RendezvousApp(mount="", barrier=barrier, observed=observed)])130 131 await asyncio.gather(drive(server, "/a"), drive(server, "/b"))132 133 assert len(observed) == 2134 for picture in observed.values():135 assert picture["current_is_own"] is True # the ContextVar test136 assert picture["in_flight"] == 2137 # monotonic ids: two distinct requests numbered 1 and 2138 assert set(observed) == {1, 2}139 # the two requests carry the two distinct paths140 assert {p["path"] for p in observed.values()} == {"/a", "/b"}141 142 async def test_snapshot_lists_the_in_flight_records(self) -> None:143 barrier = asyncio.Barrier(2)144 observed: dict[int, dict[str, Any]] = {}145 server = BaseServer(applications=[RendezvousApp(mount="", barrier=barrier, observed=observed)])146 147 await asyncio.gather(drive(server, "/a"), drive(server, "/b"))148 149 snapshot = next(iter(observed.values()))["snapshot"]150 assert len(snapshot) == 2151 assert all(isinstance(r, RegisteredRequest) for r in snapshot)152 assert all(r.scope_type == "http" for r in snapshot)153 assert {r.path for r in snapshot} == {"/a", "/b"}154 155 async def test_registry_is_empty_after_both_complete(self) -> None:156 barrier = asyncio.Barrier(2)157 observed: dict[int, dict[str, Any]] = {}158 server = BaseServer(applications=[RendezvousApp(mount="", barrier=barrier, observed=observed)])159 160 await asyncio.gather(drive(server, "/a"), drive(server, "/b"))161 162 assert server.requests.in_flight == 0163 assert server.requests.snapshot() == []164 assert server.requests.current is None165 166 167 class TestErrorPath:168 async def test_request_is_unregistered_even_when_handler_raises(self) -> None:169 server = BaseServer(applications=[RaisingApp(mount="")])170 with pytest.raises(RuntimeError, match="boom"):171 await drive(server, "/boom")172 assert server.requests.in_flight == 0173 assert server.requests.current is None174 175 async def test_cleanups_drained_even_when_handler_raises(self) -> None:176 ran: list[str] = []177 server = BaseServer(applications=[CleanupThenRaiseApp(mount="", ran=ran)])178 with pytest.raises(RuntimeError, match="boom"):179 await drive(server, "/boom")180 assert ran == ["cleanup"] # the finally drained the cleanup despite the raise181 assert server.requests.in_flight == 0182 183 184 class TestCleanups:185 def test_add_cleanup_runs_lifo(self) -> None:186 item = RegisteredRequest(1, "http", "/")187 order: list[str] = []188 item.add_cleanup(lambda: order.append("a"))189 item.add_cleanup(lambda: order.append("b"))190 item.run_cleanups()191 assert order == ["b", "a"] # last registered runs first192 193 def test_exception_in_one_cleanup_does_not_stop_the_rest(self) -> None:194 item = RegisteredRequest(1, "http", "/")195 order: list[str] = []196 197 def boom() -> None:198 raise RuntimeError("cleanup failure")199 200 item.add_cleanup(lambda: order.append("first"))201 item.add_cleanup(boom)202 item.add_cleanup(lambda: order.append("last"))203 item.run_cleanups()204 assert order == ["last", "first"] # LIFO; the raising one is isolated205 206 def test_exception_in_a_cleanup_is_logged(207 self, caplog: pytest.LogCaptureFixture208 ) -> None:209 item = RegisteredRequest(1, "http", "/")210 211 def boom() -> None:212 raise RuntimeError("cleanup failure")213 214 item.add_cleanup(boom)215 with caplog.at_level("ERROR", logger="genro_asgi.request_registry"):216 item.run_cleanups()217 assert any(218 "Request cleanup" in record.message and record.exc_info219 for record in caplog.records220 )221 222 def test_run_cleanups_is_noop_without_any(self) -> None:223 item = RegisteredRequest(1, "http", "/")224 item.run_cleanups() # no cleanups queued → nothing happens, no error225 226 227 class OkApp(BaseApplication):228 """Test app: a plain 200 for every path."""229 230 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:231 await send({"type": "http.response.start", "status": 200, "headers": []})232 await send({"type": "http.response.body", "body": b"ok"})233 234 235 class HeldApp(BaseApplication):236 """Test app blocked on an Event until the test lets it answer.237 238 Constructor kwargs peeled here (cooperative chain): ``gate`` — the Event the239 test sets to let the request finish.240 """241 242 def __init__(self, **kwargs: Any) -> None:243 self.gate: asyncio.Event = kwargs.pop("gate")244 super().__init__(**kwargs)245 246 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:247 await self.gate.wait()248 await send({"type": "http.response.start", "status": 200, "headers": []})249 await send({"type": "http.response.body", "body": b"ok"})250 251 252 class MwServer(MiddlewareMixin, BaseServer):253 """The middleware capability over the base server, for the chain test."""254 255 256 class TestServerState:257 async def test_a_server_that_is_not_running_refuses_a_new_request(258 self, http_request, response_status, response_headers259 ) -> None:260 server = BaseServer(applications=[OkApp(mount="")])261 server.state = QUITTING262 sent = await http_request(server, "/")263 assert response_status(sent) == 503264 retry_after = response_headers(sent)[b"retry-after"]265 assert retry_after == str(REFUSED_RETRY_AFTER_SECONDS).encode()266 267 async def test_a_server_back_to_running_serves_again(268 self, http_request, response_status269 ) -> None:270 server = BaseServer(applications=[OkApp(mount="")])271 server.state = QUITTING272 server.state = RUNNING273 assert response_status(await http_request(server, "/")) == 200274 275 async def test_a_refused_request_is_never_registered(self, http_request) -> None:276 server = BaseServer(applications=[OkApp(mount="")])277 server.state = QUITTING278 await http_request(server, "/")279 assert server.requests.in_flight == 0280 assert server.requests.snapshot() == []281 282 async def test_an_empty_registry_drains_at_once(self) -> None:283 server = BaseServer(applications=[OkApp(mount="")])284 assert await server.requests.await_drain(timeout=0.1) == 0285 286 async def test_the_drain_returns_as_the_last_request_ends(287 self, http_request, response_status288 ) -> None:289 gate = asyncio.Event()290 server = BaseServer(applications=[HeldApp(mount="", gate=gate)])291 in_flight = asyncio.ensure_future(http_request(server, "/"))292 await asyncio.sleep(0)293 server.state = QUITTING294 assert server.requests.in_flight == 1295 draining = asyncio.ensure_future(server.requests.await_drain())296 gate.set()297 assert await draining == 0298 assert response_status(await in_flight) == 200299 300 async def test_the_drain_reports_what_is_still_in_flight_past_its_timeout(301 self, http_request302 ) -> None:303 gate = asyncio.Event()304 server = BaseServer(applications=[HeldApp(mount="", gate=gate)])305 in_flight = asyncio.ensure_future(http_request(server, "/"))306 await asyncio.sleep(0)307 server.state = QUITTING308 assert await server.requests.await_drain(timeout=0.05) == 1309 gate.set()310 await in_flight311 312 async def test_what_the_chain_answers_itself_passes_a_server_not_running(313 self, http_request, response_status314 ) -> None:315 server = MwServer(applications=[OkApp(mount="")], middleware={"wellknown": True})316 server.state = QUITTING317 # The wellknown middleware answers /.well-known/* itself, before the318 # dispatch the state guards: 404 from the chain, never the 503.319 assert response_status(await http_request(server, "/.well-known/probe")) == 404320 assert response_status(await http_request(server, "/")) == 503