tests/core/test_request.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 tests (Phase 1c/2): the HTTP request parses a real ASGI scope16 (method/path/query/headers/cookies), body dispatch through ``handler_kwargs``17 follows the content-type (form merges, hydrated body → ``body_data``, opaque18 bytes → ``body_raw``, empty body → query only), the request id comes from the19 header or is generated, TYTX mode is read off the header, and auth/session ride20 the scope.21 22 Body decoding is exercised per content-type (xml/msgpack/json hydration, an23 urlencoded form with typed values, an unknown type and a type-less body kept24 raw, a body split over several ASGI messages) and multipart forms deliver text25 fields hydrated and file parts as ``UploadedFile`` kwargs.26 27 The ``db`` preparation layer is exercised end-to-end through the server: a fake28 app touches ``request.db`` and the server drains ``closeConnection`` at end of29 request; ``get_db`` never registers a cleanup; an unregistered code answers30 ``None``.31 """32 33 from __future__ import annotations34 35 import uuid36 from datetime import date37 from decimal import Decimal38 from typing import Any39 40 from genro_tytx import to_tytx41 42 from genro_asgi import (43     Avatar,44     BaseApplication,45     BaseServer,46     Request,47     Response,48     Session,49     UploadedFile,50 )51 from genro_asgi.types import Receive, Scope, Send52 53 54 async def make_request(55     *,56     headers: list[tuple[bytes, bytes]] | None = None,57     query: bytes = b"",58     body: bytes = b"",59     chunks: list[bytes] | None = None,60     method: str = "GET",61     path: str = "/",62     scope_extra: dict[str, Any] | None = None,63     **kwargs: Any,64 ) -> Request:65     """Build a ``Request`` from a synthetic ASGI scope and init it.66 67     ``chunks`` delivers the body in several ``http.request`` messages (the last68     one alone with ``more_body`` false); ``body`` delivers it in a single one.69     """70     pending = list(chunks) if chunks is not None else [body]71 72     async def receive() -> dict[str, Any]:73         chunk = pending.pop(0)74         return {"type": "http.request", "body": chunk, "more_body": bool(pending)}75 76     scope: Scope = {77         "type": "http",78         "method": method,79         "path": path,80         "query_string": query,81         "headers": headers or [],82     }83     if scope_extra:84         scope.update(scope_extra)85     request = Request(scope, receive, **kwargs)86     await request.init()87     return request88 89 90 class TestParsing:91     async def test_parses_method_path_query_headers_cookies(self) -> None:92         request = await make_request(93             method="post",94             path="/users",95             query=b"page=2&q=hello",96             headers=[97                 (b"content-type", b"application/json"),98                 (b"x-request-id", b"req-123"),99                 (b"cookie", b"sid=abc; theme=dark"),100             ],101         )102         assert request.method == "POST"  # uppercased103         assert request.path == "/users"104         assert request.query == {"page": 2, "q": "hello"}  # typed via TYTX105         assert request.content_type == "application/json"106         assert request.cookies == {"sid": "abc", "theme": "dark"}107         assert request.id == "req-123"108 109     async def test_response_is_bound_back_to_the_request(self) -> None:110         request = await make_request()111         assert isinstance(request.response, Response)112         assert request.response.request is request113 114     async def test_numeric_id_headers_are_coerced_to_str(self) -> None:115         # Header values are TYTX-hydrated: "123" arrives as int 123;116         # the id/external_id contract is str regardless.117         request = await make_request(118             headers=[(b"x-request-id", b"123"), (b"x-external-id", b"456")]119         )120         assert request.id == "123"121         assert request.external_id == "456"122 123 124 class TestHandlerKwargs:125     async def test_json_body_passed_whole_as_body_data(self) -> None:126         request = await make_request(127             method="POST",128             query=b"page=1",129             headers=[(b"content-type", b"application/json")],130             body=b'{"name":"ada","age":36}',131         )132         assert request.data == {"name": "ada", "age": 36}133         assert request.handler_kwargs() == {"page": 1, "body_data": {"name": "ada", "age": 36}}134 135     async def test_urlencoded_body_merges_typed_and_wins_on_clash(self) -> None:136         request = await make_request(137             method="POST",138             query=b"a=9&page=2",139             headers=[(b"content-type", b"application/x-www-form-urlencoded")],140             body=b"a=1&b=hello",141         )142         # The urlencoded body is hydrated via TYTX from_qs: it arrives as a143         # typed dict, not raw bytes.144         assert request.data == {"a": 1, "b": "hello"}145         kwargs = request.handler_kwargs()146         assert kwargs == {"a": 1, "b": "hello", "page": 2}  # body 'a' wins over query 'a'147 148     async def test_opaque_body_passed_as_body_raw(self) -> None:149         request = await make_request(150             method="POST",151             query=b"x=1",152             headers=[(b"content-type", b"application/octet-stream")],153             body=b"\x00\x01\x02",154         )155         assert request.data == b"\x00\x01\x02"156         assert request.handler_kwargs() == {"x": 1, "body_raw": b"\x00\x01\x02"}157 158     async def test_empty_body_yields_query_only(self) -> None:159         request = await make_request(method="GET", query=b"x=1&y=two")160         assert request.data is None161         assert request.handler_kwargs() == {"x": 1, "y": "two"}162 163 164 class TestBodyDecoding:165     async def test_xml_body_hydrated(self) -> None:166         payload = to_tytx({"root": {"attrs": {}, "value": Decimal("100.50")}}, transport="xml")167         request = await make_request(168             method="POST",169             headers=[(b"content-type", b"application/vnd.tytx+xml")],170             body=payload.encode("utf-8"),171         )172         assert request.data == {"root": {"attrs": {}, "value": Decimal("100.50")}}173 174     async def test_msgpack_body_hydrated(self) -> None:175         payload = to_tytx({"price": Decimal("100.50")}, transport="msgpack")176         request = await make_request(177             method="POST",178             headers=[(b"content-type", b"application/vnd.tytx+msgpack")],179             body=payload,180         )181         assert request.data == {"price": Decimal("100.50")}182 183     async def test_standard_json_media_type_hydrates_like_the_tytx_one(self) -> None:184         payload = to_tytx({"price": Decimal("100.50")}, transport="json")185         request = await make_request(186             method="POST",187             headers=[(b"content-type", b"application/json")],188             body=payload.encode("utf-8"),189         )190         assert request.data == {"price": Decimal("100.50")}191 192     async def test_unknown_content_type_stays_raw(self) -> None:193         blob = b"\x89PNG\r\n\x1a\n binary image bytes"194         request = await make_request(195             method="POST", headers=[(b"content-type", b"image/png")], body=blob196         )197         assert request.data == blob198         assert request.handler_kwargs() == {"body_raw": blob}199 200     async def test_body_without_content_type_is_read_and_kept_raw(self) -> None:201         blob = b"orphan payload"202         request = await make_request(method="POST", body=blob)203         assert request.data == blob204         assert request.handler_kwargs() == {"body_raw": blob}205 206     async def test_chunked_body_is_reassembled(self) -> None:207         request = await make_request(208             method="POST",209             headers=[(b"content-type", b"application/json")],210             chunks=[b'{"name":', b'"ada","age"', b":36}"],211         )212         assert request.data == {"name": "ada", "age": 36}213 214     async def test_multi_value_query_becomes_a_list_of_hydrated_values(self) -> None:215         request = await make_request(query=b"tag=100.50%3A%3AN&tag=200.75%3A%3AN")216         assert request.query == {"tag": [Decimal("100.50"), Decimal("200.75")]}217 218     async def test_urlencoded_typed_values_are_hydrated(self) -> None:219         request = await make_request(220             method="POST",221             headers=[(b"content-type", b"application/x-www-form-urlencoded")],222             body=b"n=1::L&price=100.50::N&when=2025-01-15::D",223         )224         assert request.data == {"n": 1, "price": Decimal("100.50"), "when": date(2025, 1, 15)}225 226     async def test_empty_json_body_is_none_and_adds_no_kwargs(self) -> None:227         request = await make_request(228             method="POST",229             query=b"x=1",230             headers=[(b"content-type", b"application/json")],231             body=b"",232         )233         assert request.data is None234         assert request.handler_kwargs() == {"x": 1}235 236 237 MULTIPART_BOUNDARY = "----genroasgi"238 MULTIPART_CONTENT_TYPE = f"multipart/form-data; boundary={MULTIPART_BOUNDARY}".encode()239 240 241 def form_part(242     name: str,243     payload: bytes,244     *,245     filename: str | None = None,246     content_type: str | None = None,247 ) -> bytes:248     """Render one multipart part: content-disposition, optional type, payload."""249     disposition = f'form-data; name="{name}"'250     if filename is not None:251         disposition += f'; filename="{filename}"'252     lines = [f"Content-Disposition: {disposition}".encode()]253     if content_type is not None:254         lines.append(f"Content-Type: {content_type}".encode())255     return b"\r\n".join(lines) + b"\r\n\r\n" + payload256 257 258 async def make_multipart_request(*parts: bytes) -> Request:259     """Build and init a POST request whose body is a multipart form of ``parts``."""260     marker = f"--{MULTIPART_BOUNDARY}".encode()261     body = b"".join(marker + b"\r\n" + part + b"\r\n" for part in parts) + marker + b"--\r\n"262     return await make_request(263         method="POST", headers=[(b"content-type", MULTIPART_CONTENT_TYPE)], body=body264     )265 266 267 class TestMultipart:268     async def test_text_field_and_file_arrive_as_kwargs(self) -> None:269         request = await make_multipart_request(270             form_part("title", b"my doc"),271             form_part("doc", b"file bytes", filename="a.txt", content_type="text/plain"),272         )273         kwargs = request.handler_kwargs()274         assert kwargs["title"] == "my doc"275         uploaded = kwargs["doc"]276         assert isinstance(uploaded, UploadedFile)277         assert uploaded.name == "doc"278         assert uploaded.filename == "a.txt"279         assert uploaded.content_type == "text/plain"280         assert uploaded.data == b"file bytes"281 282     async def test_text_field_is_tytx_hydrated(self) -> None:283         request = await make_multipart_request(form_part("count", b"123"))284         assert request.handler_kwargs() == {"count": 123}285 286     async def test_two_files_under_different_names(self) -> None:287         request = await make_multipart_request(288             form_part("first", b"one", filename="one.txt", content_type="text/plain"),289             form_part("second", b"two", filename="two.txt", content_type="text/plain"),290         )291         kwargs = request.handler_kwargs()292         assert kwargs["first"].filename == "one.txt"293         assert kwargs["second"].filename == "two.txt"294         assert kwargs["first"].data == b"one"295         assert kwargs["second"].data == b"two"296 297     async def test_repeated_name_collects_a_list(self) -> None:298         request = await make_multipart_request(299             form_part("doc", b"one", filename="one.txt", content_type="text/plain"),300             form_part("doc", b"two", filename="two.txt", content_type="text/plain"),301         )302         docs = request.handler_kwargs()["doc"]303         assert [type(item) for item in docs] == [UploadedFile, UploadedFile]304         assert [item.filename for item in docs] == ["one.txt", "two.txt"]305 306     async def test_binary_payload_survives_intact(self) -> None:307         blob = b"\x89PNG\x00\xff\r\n\x1a\n tail"308         request = await make_multipart_request(309             form_part("doc", blob, filename="a.png", content_type="image/png")310         )311         assert request.handler_kwargs()["doc"].data == blob312 313     async def test_data_is_the_decoded_dict(self) -> None:314         request = await make_multipart_request(315             form_part("title", b"my doc"),316             form_part("doc", b"bytes", filename="a.bin", content_type="application/octet-stream"),317         )318         assert isinstance(request.data, dict)319         assert set(request.data) == {"title", "doc"}320         assert request.data["title"] == "my doc"321 322     async def test_each_file_keeps_its_own_content_type(self) -> None:323         request = await make_multipart_request(324             form_part("image", b"png", filename="a.png", content_type="image/png"),325             form_part("blob", b"raw", filename="a.bin", content_type="application/octet-stream"),326         )327         kwargs = request.handler_kwargs()328         assert kwargs["image"].content_type == "image/png"329         assert kwargs["blob"].content_type == "application/octet-stream"330 331 332 class TestIdentityMetadata:333     async def test_request_id_generated_when_header_absent(self) -> None:334         request = await make_request()335         assert uuid.UUID(request.id)  # a valid uuid4, not empty336 337     async def test_external_id_from_header(self) -> None:338         request = await make_request(headers=[(b"x-external-id", b"corr-1")])339         assert request.external_id == "corr-1"340 341     async def test_tytx_mode_detected_from_header(self) -> None:342         request = await make_request(headers=[(b"x-tytx-transport", b"json")])343         assert request.tytx_mode is True344         assert request.tytx_transport == "json"345 346     async def test_no_tytx_header_means_plain_mode(self) -> None:347         request = await make_request()348         assert request.tytx_mode is False349         assert request.tytx_transport is None350 351 352 class TestAuthSessionAccessors:353     async def test_avatar_and_tags_from_scope(self) -> None:354         avatar = Avatar("alice", ["admin", "staff"])355         request = await make_request(scope_extra={"auth": avatar})356         assert request.avatar() is avatar357         assert request.auth_tags == ["admin", "staff"]358 359     async def test_anonymous_when_no_avatar_on_scope(self) -> None:360         request = await make_request()361         assert request.avatar() is None362         assert request.auth_tags == []363 364     async def test_keyed_avatar_delegates_to_the_session(self) -> None:365         session = Session("tok", avatar=Avatar("alice"), ttl=3600)366         session.attach_avatar(Avatar("alice@erp"), "erp")367         request = await make_request(scope_extra={"session": session})368         assert request.avatar("erp").identity == "alice@erp"369 370     async def test_keyed_avatar_without_a_session_is_none(self) -> None:371         request = await make_request()372         assert request.avatar("erp") is None373 374     async def test_session_from_scope(self) -> None:375         session = object()376         request = await make_request(scope_extra={"session": session})377         assert request.session is session378 379     async def test_server_resolved_via_application(self) -> None:380         app = BaseApplication(mount="")381         server = BaseServer(applications=[app])382         request = await make_request(application=app)383         assert request.application is app384         assert request.server is server385 386 387 class FakeDb:388     """A stand-in db handler recording how often it is closed."""389 390     def __init__(self) -> None:391         self.closed = 0392 393     def closeConnection(self) -> None:394         self.closed += 1395 396 397 class DbApp(BaseApplication):398     """Primary app that touches ``request.db`` (twice) and records the handler."""399 400     def __init__(self, **kwargs: Any) -> None:401         self.db_name: str | None = kwargs.pop("db_name", None)402         self.seen: dict[str, Any] = kwargs.pop("seen")403         super().__init__(**kwargs)404 405     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:406         request = Request(scope, receive, application=self)407         await request.init()408         self.seen["db"] = request.db409         self.seen["db_again"] = request.db  # second access is cached410         await request.response(scope, receive, send)411 412 413 class GetDbApp(BaseApplication):414     """Primary app that resolves a db via ``get_db`` (no cleanup registration)."""415 416     def __init__(self, **kwargs: Any) -> None:417         self.seen: dict[str, Any] = kwargs.pop("seen")418         super().__init__(**kwargs)419 420     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:421         request = Request(scope, receive, application=self)422         await request.init()423         self.seen["db"] = request.get_db("default")424         await request.response(scope, receive, send)425 426 427 async def drive(server: BaseServer, path: str = "/") -> None:428     """Drive one http request through the full server dispatch."""429 430     async def receive() -> dict[str, Any]:431         return {"type": "http.request", "body": b"", "more_body": False}432 433     async def send(message: dict[str, Any]) -> None:434         pass435 436     await server(437         {"type": "http", "method": "GET", "path": path, "query_string": b"", "headers": []},438         receive,439         send,440     )441 442 443 class TestDbPreparationLayer:444     async def test_db_returns_default_handler_and_closes_at_request_end(self) -> None:445         seen: dict[str, Any] = {}446         fake = FakeDb()447         server = BaseServer(applications=[DbApp(mount="", seen=seen)])448         server.add_database("default", fake)449 450         await drive(server)451 452         assert seen["db"] is fake453         assert seen["db_again"] is fake  # cached, same object454         assert fake.closed == 1  # cleanup registered once, drained by the server455 456     async def test_db_resolves_named_handler_from_app_db_name(self) -> None:457         seen: dict[str, Any] = {}458         fake = FakeDb()459         server = BaseServer(applications=[DbApp(mount="", db_name="shop", seen=seen)])460         server.add_database("shop", fake)461 462         await drive(server)463 464         assert seen["db"] is fake465         assert fake.closed == 1466 467     async def test_db_is_none_when_code_absent(self) -> None:468         seen: dict[str, Any] = {}469         server = BaseServer(applications=[DbApp(mount="", seen=seen)])  # no database registered470 471         await drive(server)472 473         assert seen["db"] is None474 475     async def test_get_db_does_not_register_cleanup(self) -> None:476         seen: dict[str, Any] = {}477         fake = FakeDb()478         server = BaseServer(applications=[GetDbApp(mount="", seen=seen)])479         server.add_database("default", fake)480 481         await drive(server)482 483         assert seen["db"] is fake484         assert fake.closed == 0  # get_db never queues closeConnection485 486     async def test_get_db_is_none_when_code_absent(self) -> None:487         seen: dict[str, Any] = {}488         server = BaseServer(applications=[GetDbApp(mount="", seen=seen)])  # nothing registered489 490         await drive(server)491 492         assert seen["db"] is None