tests/core/test_middleware.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 """Middleware core tests (SPECIFICATION.md D16, Macro 2 Phase 2).16 17 The chain is driven directly through the composed server's ``__call__`` (no18 uvicorn): a canned http scope, a recording ``send``. Asserted here: chain19 ordering by ``middleware_order``, the error middleware's exception mapping20 (404 from ``HTTPNotFound``, 302 from ``Redirect``, 500 from a plain21 ``Exception`` — and the server survives), the ``lifespan`` bypass, the22 untouched plain-``BaseServer`` composition, and the D16 leftover-kwarg23 ``TypeError`` through the full MRO.24 """25 26 from __future__ import annotations27 28 import pytest29 30 from genro_asgi import BaseApplication, BaseServer, MiddlewareMixin31 from genro_asgi.exceptions import HTTPNotFound, Redirect32 from genro_asgi.middleware import BaseMiddleware33 from genro_asgi.middleware.base import headers_dict34 from genro_asgi.types import Message, Receive, Scope, Send35 36 37 class MwServer(MiddlewareMixin, BaseServer):38 """The Phase 2 composition: middleware capability over the base server."""39 40 41 class RoutedApp(BaseApplication):42 """Test app: routes raising the control-flow exceptions under test."""43 44 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:45 path = scope["path"]46 if path == "/missing":47 raise HTTPNotFound("nothing here")48 if path == "/old":49 raise Redirect("/new")50 if path == "/boom":51 raise RuntimeError("boom")52 await send(53 {54 "type": "http.response.start",55 "status": 200,56 "headers": [(b"content-type", b"text/plain; charset=utf-8")],57 }58 )59 await send({"type": "http.response.body", "body": f"ok:{path}".encode()})60 61 62 class RecordingMiddleware(BaseMiddleware):63 """Middleware appending its label to a shared list when invoked."""64 65 def __init__(self, app, server, label="", calls=None, **options):66 super().__init__(app, server, **options)67 self.label = label68 self.calls = calls if calls is not None else []69 70 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:71 self.calls.append(self.label)72 await self.app(scope, receive, send)73 74 75 class EarlyMiddleware(RecordingMiddleware):76 middleware_order = 20077 78 79 class LateMiddleware(RecordingMiddleware):80 middleware_order = 80081 82 83 async def http_get(server: BaseServer, path: str) -> list[Message]:84 """Drive one GET through ``server`` at the ASGI level; return what it sent."""85 scope: Scope = {"type": "http", "method": "GET", "path": path, "headers": []}86 sent: list[Message] = []87 88 async def receive() -> Message:89 return {"type": "http.request"}90 91 async def send(message: Message) -> None:92 sent.append(message)93 94 await server(scope, receive, send)95 return sent96 97 98 def response_status(sent: list[Message]) -> int:99 return next(m["status"] for m in sent if m["type"] == "http.response.start")100 101 102 def response_headers(sent: list[Message]) -> dict[bytes, bytes]:103 start = next(m for m in sent if m["type"] == "http.response.start")104 return dict(start["headers"])105 106 107 def response_body(sent: list[Message]) -> bytes:108 return b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")109 110 111 class TestHeadersDict:112 """The header helper every middleware shares: latin-1 decode, lowercase, cached."""113 114 def test_names_are_lowercased_and_the_last_duplicate_wins(self) -> None:115 scope: Scope = {"headers": [(b"X-One", b"1"), (b"x-one", b"2")]}116 assert headers_dict(scope) == {"x-one": "2"}117 118 def test_the_parsed_dict_is_cached_on_the_scope(self) -> None:119 scope: Scope = {"headers": [(b"X-One", b"1")]}120 parsed = headers_dict(scope)121 assert scope["_headers"] is parsed122 assert headers_dict(scope) is parsed # second call reuses the cache123 124 125 class TestChainAssembly:126 async def test_chain_invokes_middlewares_in_order(self) -> None:127 calls: list[str] = []128 server = MwServer(129 applications=[RoutedApp(mount="")],130 middleware={131 "late": {"label": "late", "calls": calls},132 "early": {"label": "early", "calls": calls},133 },134 middleware_registry={"early": EarlyMiddleware, "late": LateMiddleware},135 )136 sent = await http_get(server, "/")137 assert calls == ["early", "late"]138 assert response_status(sent) == 200139 assert response_body(sent) == b"ok:/"140 141 async def test_unknown_middleware_name_raises(self) -> None:142 with pytest.raises(ValueError, match="bogus"):143 MwServer(applications=[RoutedApp(mount="")], middleware={"bogus": True})144 145 async def test_false_switch_disables_a_default_middleware(self) -> None:146 server = MwServer(applications=[RoutedApp(mount="")], middleware={"errors": False})147 with pytest.raises(RuntimeError, match="boom"):148 await http_get(server, "/boom")149 150 151 class TestErrorMiddleware:152 async def test_http_exception_maps_to_its_status(self) -> None:153 server = MwServer(applications=[RoutedApp(mount="")])154 sent = await http_get(server, "/missing")155 assert response_status(sent) == 404156 assert response_body(sent) == b"nothing here"157 158 async def test_redirect_maps_to_status_and_location(self) -> None:159 server = MwServer(applications=[RoutedApp(mount="")])160 sent = await http_get(server, "/old")161 assert response_status(sent) == 302162 assert response_headers(sent)[b"location"] == b"/new"163 164 async def test_plain_exception_maps_to_500_and_server_survives(self) -> None:165 server = MwServer(applications=[RoutedApp(mount="")])166 boom = await http_get(server, "/boom")167 assert response_status(boom) == 500168 assert response_body(boom) == b"Internal Server Error"169 healthy = await http_get(server, "/")170 assert response_status(healthy) == 200171 172 173 class TestScopeRouting:174 async def test_lifespan_scope_bypasses_the_chain(self) -> None:175 calls: list[str] = []176 server = MwServer(177 applications=[RoutedApp(mount="")],178 middleware={"early": {"label": "early", "calls": calls}},179 middleware_registry={"early": EarlyMiddleware},180 )181 queue = [{"type": "lifespan.startup"}, {"type": "lifespan.shutdown"}]182 sent: list[Message] = []183 184 async def receive() -> Message:185 return queue.pop(0)186 187 async def send(message: Message) -> None:188 sent.append(message)189 190 await server({"type": "lifespan"}, receive, send)191 assert calls == []192 assert {"type": "lifespan.startup.complete"} in sent193 assert {"type": "lifespan.shutdown.complete"} in sent194 195 196 class TestComposition:197 def test_plain_base_server_lacks_the_mixin_attrs(self) -> None:198 server = BaseServer(applications=[RoutedApp(mount="")])199 assert not hasattr(server, "middleware_chain")200 201 def test_leftover_kwarg_raises_naming_it_through_the_mro(self) -> None:202 with pytest.raises(TypeError, match="bogus"):203 MwServer(applications=[RoutedApp(mount="")], bogus=3)204 205 def test_middleware_options_leftover_raises_naming_it(self) -> None:206 with pytest.raises(TypeError, match="bogus"):207 MwServer(applications=[RoutedApp(mount="")], middleware={"errors": {"bogus": True}})