tests/core/test_response.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 """Response tests (Macro 4 Phase 1): the one flat, buffered, TYTX-aware class.16 17 Drives the Response as an ASGI app with a recording ``send`` and asserts the18 wire shape (exactly two messages), then covers ``set_header``/``set_cookie``/19 ``set_result``/``set_error`` and the TYTX branch through a stub request.20 """21 22 from __future__ import annotations23 24 import json25 from pathlib import Path26 from typing import Any27 from urllib.parse import quote28 29 from genro_tytx import json_dumps, to_tytx30 31 from genro_asgi import HTTPForbidden, HTTPNotFound, HTTPUnauthorized, Response32 from genro_asgi.media_types import TRANSPORT_MIME33 from genro_asgi.types import Message34 35 36 class StubRequest:37 """Minimal request stand-in exposing the TYTX contract Response reads."""38 39 def __init__(self, tytx_mode: bool = False, tytx_transport: str | None = None) -> None:40 self.tytx_mode = tytx_mode41 self.tytx_transport = tytx_transport42 43 44 async def drive(response: Response) -> list[Message]:45 """Run ``response`` as an ASGI app; return the messages it sent."""46 sent: list[Message] = []47 48 async def receive() -> Message:49 return {"type": "http.request", "body": b""}50 51 async def send(message: Message) -> None:52 sent.append(message)53 54 await response({}, receive, send)55 return sent56 57 58 def start_headers(sent: list[Message]) -> dict[bytes, bytes]:59 start = next(m for m in sent if m["type"] == "http.response.start")60 return dict(start["headers"])61 62 63 class TestConstruction:64 def test_content_status_media_type(self) -> None:65 response = Response(content="Hello", status_code=201, media_type="text/plain")66 assert response.status_code == 20167 assert response.body == b"Hello"68 69 def test_str_content_encoded_utf8(self) -> None:70 response = Response(content="cafè")71 assert response.body == "cafè".encode()72 73 def test_bytes_content_as_is(self) -> None:74 response = Response(content=b"\x00\x01\x02")75 assert response.body == b"\x00\x01\x02"76 77 def test_none_content_is_empty(self) -> None:78 response = Response()79 assert response.body == b""80 81 def test_headers_from_list_preserved(self) -> None:82 response = Response(content="x", headers=[("x-a", "1")])83 assert ("x-a", "1") in response._headers84 85 def test_headers_from_mapping_preserved(self) -> None:86 response = Response(content="x", headers={"x-b": "2"})87 assert ("x-b", "2") in response._headers88 89 def test_text_media_type_gets_charset(self) -> None:90 response = Response(content="hi", media_type="text/plain")91 headers = dict(response._headers)92 assert headers["content-type"] == "text/plain; charset=utf-8"93 94 def test_non_text_media_type_no_charset(self) -> None:95 response = Response(content=b"{}", media_type="application/json")96 headers = dict(response._headers)97 assert headers["content-type"] == "application/json"98 99 def test_content_length_set(self) -> None:100 response = Response(content="hello")101 headers = dict(response._headers)102 assert headers["content-length"] == "5"103 104 105 class TestAsgiCall:106 async def test_emits_exactly_start_and_body(self) -> None:107 sent = await drive(Response(content="hi", media_type="text/plain"))108 assert [m["type"] for m in sent] == ["http.response.start", "http.response.body"]109 110 async def test_wire_status_headers_body(self) -> None:111 sent = await drive(Response(content="hi", status_code=202, media_type="text/plain"))112 start = sent[0]113 assert start["status"] == 202114 headers = dict(start["headers"])115 assert headers[b"content-type"] == b"text/plain; charset=utf-8"116 assert headers[b"content-length"] == b"2"117 assert sent[1]["body"] == b"hi"118 119 async def test_header_names_lowercased_latin1(self) -> None:120 response = Response(content="x")121 response.set_header("X-Custom", "value")122 sent = await drive(response)123 headers = dict(sent[0]["headers"])124 assert headers[b"x-custom"] == b"value"125 126 127 class TestSetHeaderAndCookie:128 def test_set_header_appends(self) -> None:129 response = Response(content="x")130 response.set_header("x-one", "a")131 response.set_header("x-one", "b")132 values = [v for k, v in response._headers if k == "x-one"]133 assert values == ["a", "b"]134 135 def test_set_cookie_basic(self) -> None:136 response = Response(content="x")137 response.set_cookie("session", "abc")138 cookie = dict(response._headers)["set-cookie"]139 assert cookie == "session=abc; Path=/; SameSite=Lax"140 141 def test_set_cookie_all_attributes(self) -> None:142 response = Response(content="x")143 response.set_cookie(144 "k",145 "v",146 max_age=3600,147 path="/app",148 domain="example.com",149 secure=True,150 httponly=True,151 samesite="strict",152 )153 cookie = dict(response._headers)["set-cookie"]154 assert cookie == (155 "k=v; Max-Age=3600; Path=/app; Domain=example.com; "156 "Secure; HttpOnly; SameSite=Strict"157 )158 159 def test_set_cookie_value_url_encoded(self) -> None:160 response = Response(content="x")161 response.set_cookie("k", "a b/c")162 cookie = dict(response._headers)["set-cookie"]163 assert cookie.startswith(f"k={quote('a b/c', safe='')}")164 165 def test_set_cookie_samesite_none_omitted(self) -> None:166 response = Response(content="x")167 response.set_cookie("k", "v", samesite=None)168 cookie = dict(response._headers)["set-cookie"]169 assert "SameSite" not in cookie170 171 172 class TestSetResult:173 def test_dict_to_json_via_json_dumps(self) -> None:174 response = Response()175 response.set_result({"a": 1, "b": 2})176 assert response.body == json_dumps({"a": 1, "b": 2})177 assert dict(response._headers)["content-type"] == "application/json"178 179 def test_list_to_json(self) -> None:180 response = Response()181 response.set_result([1, 2, 3])182 assert response.body == json_dumps([1, 2, 3])183 184 def test_path_to_bytes_octet_stream(self, tmp_path: Path) -> None:185 f = tmp_path / "data.bin"186 f.write_bytes(b"\x01\x02\x03")187 response = Response()188 response.set_result(f)189 assert response.body == b"\x01\x02\x03"190 assert dict(response._headers)["content-type"] == "application/octet-stream"191 192 def test_bytes_result(self) -> None:193 response = Response()194 response.set_result(b"raw")195 assert response.body == b"raw"196 assert dict(response._headers)["content-type"] == "application/octet-stream"197 198 def test_str_result_text_plain(self) -> None:199 response = Response()200 response.set_result("hello")201 assert response.body == b"hello"202 assert dict(response._headers)["content-type"] == "text/plain; charset=utf-8"203 204 def test_none_result_empty_body(self) -> None:205 response = Response()206 response.set_result(None)207 assert response.body == b""208 assert dict(response._headers)["content-length"] == "0"209 210 def test_other_result_str_conversion(self) -> None:211 response = Response()212 response.set_result(42)213 assert response.body == b"42"214 215 def test_metadata_media_type_override(self) -> None:216 response = Response()217 response.set_result({"a": 1}, metadata={"media_type": "application/vnd.custom+json"})218 assert dict(response._headers)["content-type"] == "application/vnd.custom+json"219 220 def test_content_headers_replaced_not_duplicated(self) -> None:221 response = Response(content="old", media_type="text/plain")222 response.set_result({"a": 1})223 content_types = [v for k, v in response._headers if k.lower() == "content-type"]224 content_lengths = [v for k, v in response._headers if k.lower() == "content-length"]225 assert content_types == ["application/json"]226 assert content_lengths == [str(len(b'{"a":1}'))]227 228 229 class TestTytxBranch:230 def test_tytx_json_transport(self) -> None:231 request = StubRequest(tytx_mode=True, tytx_transport="json")232 response = Response(request=request)233 response.set_result({"a": 1})234 assert response.body.decode("utf-8") == to_tytx({"a": 1}, "json")235 # media type is sourced from genro-tytx, never a local literal236 assert dict(response._headers)["content-type"] == TRANSPORT_MIME["json"]237 238 def test_tytx_msgpack_transport_bytes(self) -> None:239 request = StubRequest(tytx_mode=True, tytx_transport="msgpack")240 response = Response(request=request)241 response.set_result({"a": 1})242 assert response.body == to_tytx({"a": 1}, "msgpack")243 assert dict(response._headers)["content-type"] == TRANSPORT_MIME["msgpack"]244 245 def test_tytx_default_transport_json_when_missing(self) -> None:246 request = StubRequest(tytx_mode=True, tytx_transport=None)247 response = Response(request=request)248 response.set_result([1, 2])249 assert dict(response._headers)["content-type"] == TRANSPORT_MIME["json"]250 251 def test_no_request_falls_back_to_json(self) -> None:252 response = Response(request=None)253 response.set_result({"a": 1})254 assert response.body == b'{"a":1}'255 assert dict(response._headers)["content-type"] == "application/json"256 257 def test_tytx_mode_off_uses_json(self) -> None:258 request = StubRequest(tytx_mode=False)259 response = Response(request=request)260 response.set_result({"a": 1})261 assert response.body == b'{"a":1}'262 263 264 class TestSetError:265 def test_http_exception_carries_own_status(self) -> None:266 response = Response()267 response.set_error(HTTPNotFound("missing"))268 assert response.status_code == 404269 assert json.loads(response.body) == {"error": "missing"}270 271 def test_http_unauthorized_status(self) -> None:272 response = Response()273 response.set_error(HTTPUnauthorized())274 assert response.status_code == 401275 276 def test_http_forbidden_status(self) -> None:277 response = Response()278 response.set_error(HTTPForbidden())279 assert response.status_code == 403280 281 def test_value_error_maps_400(self) -> None:282 response = Response()283 response.set_error(ValueError("bad"))284 assert response.status_code == 400285 assert json.loads(response.body) == {"error": "bad"}286 287 def test_type_error_maps_400(self) -> None:288 response = Response()289 response.set_error(TypeError("wrong type"))290 assert response.status_code == 400291 292 def test_file_not_found_maps_404(self) -> None:293 response = Response()294 response.set_error(FileNotFoundError("nope"))295 assert response.status_code == 404296 297 def test_permission_error_maps_403(self) -> None:298 response = Response()299 response.set_error(PermissionError("denied"))300 assert response.status_code == 403301 302 def test_unknown_exception_maps_500(self) -> None:303 response = Response()304 305 class Boom(Exception):306 pass307 308 response.set_error(Boom("kaboom"))309 assert response.status_code == 500310 assert json.loads(response.body) == {"error": "kaboom"}311 312 async def test_error_body_wire_shape(self) -> None:313 response = Response()314 response.set_error(HTTPNotFound("gone"))315 sent = await drive(response)316 assert sent[0]["status"] == 404317 headers: dict[bytes, Any] = dict(sent[0]["headers"])318 assert headers[b"content-type"] == b"application/json"