tests/core/test_routed_application.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 """RoutedApplication tests (Macro 4 Phase 4).16 17 Every request drives a REAL ``AsgiServer`` composition at the ASGI level18 (no uvicorn): the app is the primary, ``ErrorMiddleware`` is armed by19 default, sync handlers cross the server pool. The GET-side helpers come from20 ``tests/conftest.py``; ``json_request`` (local) adds a JSON body.21 """22 23 from __future__ import annotations24 25 import threading26 from typing import Any, Callable27 28 import pytest29 from genro_routes import RoutingClass, route30 31 from genro_asgi import AsgiServer, Avatar, BaseMiddleware, RoutedApplication32 from genro_asgi.types import Message, Scope33 34 35 class DemoApp(RoutedApplication):36 """Test app: sync/async handlers, an auth-ruled one, typed metadata."""37 38 @route()39 def hello(self) -> dict[str, str]:40 return {"hello": "world"}41 42 @route()43 async def ahello(self) -> dict[str, str]:44 return {"hello": "async"}45 46 @route()47 def echo(self, a: Any = None, b: Any = None) -> dict[str, Any]:48 return {"a": a, "b": b}49 50 @route(auth_rule="admin")51 def restricted(self) -> dict[str, bool]:52 return {"secret": True}53 54 @route()55 def sync_ident(self) -> int:56 return threading.get_ident()57 58 @route()59 async def async_ident(self) -> int:60 return threading.get_ident()61 62 @route(media_type="text/html")63 def page(self) -> str:64 return "<h1>hi</h1>"65 66 @route()67 def wrapped(self) -> Any:68 return self.result_wrapper("<p>meta</p>", media_type="text/html")69 70 71 class TypedApp(RoutedApplication):72 """Pydantic-plugged app: the body-spread reconciliation has fields to read."""73 74 def __init__(self, **kwargs: Any) -> None:75 super().__init__(**kwargs)76 self.route.plug("pydantic")77 78 @route()79 def add(self, x: int = 0, y: int = 0) -> dict[str, int]:80 return {"sum": x + y}81 82 @route()83 def raw(self, body_data: dict | None = None) -> dict[str, Any]:84 return {"body": body_data}85 86 87 class SubApi(RoutingClass):88 """External RoutingClass mounted into an app via ``add_branches`` (instance form)."""89 90 @route()91 def ping(self) -> dict[str, bool]:92 return {"sub": True}93 94 95 class StampAuthMiddleware(BaseMiddleware):96 """Test middleware: stamps a fixed identity on ``scope["auth"]``.97 98 Order 500: the AsgiServer composition arms the real ``AuthMiddleware``99 (450), which resolves the scope identity itself — the stamp must run100 after it so the fixed identity wins.101 """102 103 middleware_order = 500104 105 def __init__(self, app: Any, server: Any, *, avatar: Avatar | None = None, **options: Any):106 self._avatar = avatar107 super().__init__(app, server, **options)108 109 async def __call__(self, scope: Any, receive: Any, send: Any) -> None:110 scope["auth"] = self._avatar111 await self.app(scope, receive, send)112 113 114 def auth_server(app: RoutedApplication, avatar: Avatar | None) -> AsgiServer:115 """An AsgiServer whose chain stamps ``avatar`` as the request identity."""116 return AsgiServer(117 applications=[app],118 middleware={"stamp": {"avatar": avatar}},119 middleware_registry={"stamp": StampAuthMiddleware},120 )121 122 123 @pytest.fixture124 def json_request() -> Callable[..., object]:125 """Fixture: drive one JSON-body request through a server at the ASGI level."""126 127 async def _json_request(128 server: object, path: str, body: bytes, method: str = "POST"129 ) -> list[Message]:130 scope: Scope = {131 "type": "http",132 "method": method,133 "path": path,134 "query_string": b"",135 "headers": [(b"content-type", b"application/json")],136 }137 sent: list[Message] = []138 139 async def receive() -> Message:140 return {"type": "http.request", "body": body, "more_body": False}141 142 async def send(message: Message) -> None:143 sent.append(message)144 145 await server(scope, receive, send) # type: ignore[operator]146 return sent147 148 return _json_request149 150 151 @pytest.fixture152 def query_request() -> Callable[..., object]:153 """Fixture: drive one GET request carrying a query string through a server."""154 155 async def _query_request(server: object, path: str, query: bytes) -> list[Message]:156 scope: Scope = {157 "type": "http",158 "method": "GET",159 "path": path,160 "query_string": query,161 "headers": [],162 }163 sent: list[Message] = []164 165 async def receive() -> Message:166 return {"type": "http.request"}167 168 async def send(message: Message) -> None:169 sent.append(message)170 171 await server(scope, receive, send) # type: ignore[operator]172 return sent173 174 return _query_request175 176 177 class TestDispatch:178 async def test_sync_route_answers_json(179 self, http_request, response_status, response_headers, response_body180 ) -> None:181 server = AsgiServer(applications=[DemoApp(mount="")])182 sent = await http_request(server, "/hello")183 assert response_status(sent) == 200184 assert response_headers(sent)[b"content-type"] == b"application/json"185 assert response_body(sent) == b'{"hello":"world"}'186 187 async def test_async_route_answers_json(188 self, http_request, response_status, response_body189 ) -> None:190 server = AsgiServer(applications=[DemoApp(mount="")])191 sent = await http_request(server, "/ahello")192 assert response_status(sent) == 200193 assert response_body(sent) == b'{"hello":"async"}'194 195 async def test_query_params_reach_handler_kwargs(self, response_body) -> None:196 server = AsgiServer(applications=[DemoApp(mount="")])197 scope: Scope = {198 "type": "http",199 "method": "GET",200 "path": "/echo",201 "query_string": b"a=1&b=two",202 "headers": [],203 }204 sent: list[Message] = []205 206 async def receive() -> Message:207 return {"type": "http.request"}208 209 async def send(message: Message) -> None:210 sent.append(message)211 212 await server(scope, receive, send)213 assert response_body(sent) == b'{"a":1,"b":"two"}'214 215 async def test_unknown_path_is_404_via_error_middleware(216 self, http_request, response_status217 ) -> None:218 server = AsgiServer(applications=[DemoApp(mount="")])219 sent = await http_request(server, "/nowhere")220 assert response_status(sent) == 404221 222 async def test_metadata_media_type_reaches_the_response(223 self, http_request, response_headers, response_body224 ) -> None:225 server = AsgiServer(applications=[DemoApp(mount="")])226 sent = await http_request(server, "/page")227 assert response_headers(sent)[b"content-type"] == b"text/html; charset=utf-8"228 assert response_body(sent) == b"<h1>hi</h1>"229 230 async def test_result_wrapper_metadata_wins(231 self, http_request, response_headers, response_body232 ) -> None:233 server = AsgiServer(applications=[DemoApp(mount="")])234 sent = await http_request(server, "/wrapped")235 assert response_headers(sent)[b"content-type"] == b"text/html; charset=utf-8"236 assert response_body(sent) == b"<p>meta</p>"237 238 async def test_unmounted_app_refuses_dispatch(self) -> None:239 app = DemoApp()240 241 async def receive() -> Message:242 return {"type": "http.request"}243 244 async def send(message: Message) -> None:245 raise AssertionError("nothing must be sent")246 247 with pytest.raises(RuntimeError):248 await app({"type": "http", "method": "GET", "path": "/hello"}, receive, send)249 250 251 class TestBodyBinding:252 async def test_json_body_spread_over_params(253 self, json_request, response_status, response_body254 ) -> None:255 server = AsgiServer(applications=[TypedApp(mount="")])256 sent = await json_request(server, "/add", b'{"x": 1, "y": 2, "extra": 9}')257 assert response_status(sent) == 200258 assert response_body(sent) == b'{"sum":3}'259 260 async def test_body_data_kept_whole_when_declared(self, json_request, response_body) -> None:261 server = AsgiServer(applications=[TypedApp(mount="")])262 sent = await json_request(server, "/raw", b'{"x": 1}')263 assert response_body(sent) == b'{"body":{"x":1}}'264 265 266 class TestArgumentErrors:267 """Bad handler arguments surface as an HTTP answer, never 500.268 269 genro-routes keeps the two failures apart: an unbindable extra argument is270 a ``signature_error`` and answers 400, an uncoercible typed argument is a271 ``validation_error`` and answers 422. The dispatcher catches one marker per272 code.273 """274 275 async def test_uncoercible_typed_arg_is_422(self, query_request, response_status) -> None:276 server = AsgiServer(applications=[TypedApp(mount="")])277 sent = await query_request(server, "/add", b"x=abc")278 assert response_status(sent) == 422279 280 async def test_unbindable_extra_arg_is_400(self, query_request, response_status) -> None:281 server = AsgiServer(applications=[TypedApp(mount="")])282 sent = await query_request(server, "/add", b"x=1&y=2&z=99")283 assert response_status(sent) == 400284 285 286 class TestAuth:287 async def test_anonymous_is_401_on_ruled_entry(self, http_request, response_status) -> None:288 """No identity presented: the server asks who is calling, it does not refuse."""289 server = auth_server(DemoApp(mount=""), avatar=None)290 sent = await http_request(server, "/restricted")291 assert response_status(sent) == 401292 293 async def test_wrong_tags_are_403(self, http_request, response_status) -> None:294 server = auth_server(DemoApp(mount=""), avatar=Avatar("bob", ["viewer"]))295 sent = await http_request(server, "/restricted")296 assert response_status(sent) == 403297 298 async def test_matching_tag_is_200(self, http_request, response_status, response_body) -> None:299 server = auth_server(DemoApp(mount=""), avatar=Avatar("alice", ["admin"]))300 sent = await http_request(server, "/restricted")301 assert response_status(sent) == 200302 assert response_body(sent) == b'{"secret":true}'303 304 async def test_untagged_entry_stays_public_without_middleware(305 self, http_request, response_status306 ) -> None:307 server = AsgiServer(applications=[DemoApp(mount="")])308 sent = await http_request(server, "/hello")309 assert response_status(sent) == 200310 311 async def test_ruled_entry_denied_without_middleware(312 self, http_request, response_status313 ) -> None:314 """Default-deny with no auth chain at all: nobody was presented, so 401."""315 server = AsgiServer(applications=[DemoApp(mount="")])316 sent = await http_request(server, "/restricted")317 assert response_status(sent) == 401318 319 320 class TestSubTrees:321 async def test_attached_instance_reachable_under_its_name(322 self, http_request, response_status, response_body323 ) -> None:324 app = DemoApp(mount="")325 app.route.add_branches({"name": "sub", "instance": SubApi()})326 server = AsgiServer(applications=[app])327 sent = await http_request(server, "/sub/ping")328 assert response_status(sent) == 200329 assert response_body(sent) == b'{"sub":true}'330 331 332 class TestExecutionVehicle:333 async def test_sync_handler_runs_through_the_pool(self, http_request, response_body) -> None:334 server = AsgiServer(applications=[DemoApp(mount="")])335 sent = await http_request(server, "/sync_ident")336 assert int(response_body(sent)) != threading.get_ident()337 338 async def test_async_handler_stays_on_the_loop(self, http_request, response_body) -> None:339 server = AsgiServer(applications=[DemoApp(mount="")])340 sent = await http_request(server, "/async_ident")341 assert int(response_body(sent)) == threading.get_ident()342 343 344 class CleanupApp(RoutedApplication):345 """Test app: ``route_cleanup`` records the thread it ran on, per dispatch."""346 347 def __init__(self, **kwargs: Any) -> None:348 super().__init__(**kwargs)349 self.cleanups: list[int] = []350 351 @route()352 def sync_ident(self) -> int:353 return threading.get_ident()354 355 @route()356 async def async_hello(self) -> dict[str, str]:357 return {"hello": "async"}358 359 @route()360 def failing(self) -> None:361 raise RuntimeError("boom")362 363 def route_cleanup(self) -> None:364 self.cleanups.append(threading.get_ident())365 366 367 class TestRouteCleanup:368 async def test_the_cleanup_runs_on_the_handler_thread(369 self, http_request, response_body370 ) -> None:371 app = CleanupApp(mount="")372 server = AsgiServer(applications=[app])373 sent = await http_request(server, "/sync_ident")374 # Same dispatch, same pool thread: what the handler opened, the375 # cleanup can close.376 assert app.cleanups == [int(response_body(sent))]377 378 async def test_the_cleanup_runs_when_the_handler_raises(379 self, http_request, response_status380 ) -> None:381 app = CleanupApp(mount="")382 server = AsgiServer(applications=[app])383 sent = await http_request(server, "/failing")384 assert response_status(sent) == 500385 assert len(app.cleanups) == 1386 387 async def test_the_async_path_never_cleans(self, http_request, response_status) -> None:388 app = CleanupApp(mount="")389 server = AsgiServer(applications=[app])390 sent = await http_request(server, "/async_hello")391 assert response_status(sent) == 200392 assert app.cleanups == []393 394 395 class TestBodyTypeError:396 """The handler body is mapped to no error code: its TypeError is a 500.397 398 Binding happens before the handler runs, so a body failure is never399 mistaken for a bad call — sync path and async path alike.400 """401 402 async def test_async_handler_body_typeerror_is_500(403 self, http_request, response_status404 ) -> None:405 class Exploding(RoutedApplication):406 @route()407 async def boom(self) -> dict:408 raise TypeError("async body failure")409 410 server = AsgiServer(applications=[Exploding(mount="")])411 sent = await http_request(server, "/boom")412 assert response_status(sent) == 500413 414 async def test_sync_handler_body_typeerror_is_500(415 self, http_request, response_status416 ) -> None:417 class Exploding(RoutedApplication):418 @route()419 def boom(self) -> dict:420 raise TypeError("sync body failure")421 422 server = AsgiServer(applications=[Exploding(mount="")])423 sent = await http_request(server, "/boom")424 assert response_status(sent) == 500