tests/core/test_streaming.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 """Tests for StreamingResponse (core 1e Phase 5): the chunked ASGI sibling.16 17 Real objects, no mocks: drive the response through its ``__call__`` with a18 recording ``send`` and assert the ASGI message sequence — one ``start``, one19 ``body`` per chunk with ``more_body=True``, a terminal empty body with20 ``more_body=False``. A regression check confirms ``Response`` is untouched (it21 stays buffered, two messages, no ``more_body``).22 """23 24 from __future__ import annotations25 26 from collections.abc import AsyncIterable27 from typing import Any28 29 from genro_asgi.response import Response30 from genro_asgi.streaming import StreamingResponse31 from genro_asgi.types import Message, Scope32 33 34 async def drive(app: Any) -> list[Message]:35 """Run an ASGI app once and return the recorded ``send`` messages."""36 scope: Scope = {"type": "http", "method": "GET", "path": "/", "headers": []}37 sent: list[Message] = []38 39 async def receive() -> Message:40 return {"type": "http.request"}41 42 async def send(message: Message) -> None:43 sent.append(message)44 45 await app(scope, receive, send)46 return sent47 48 49 async def gen(*chunks: bytes) -> AsyncIterable[bytes]:50 """An async iterator over the given byte chunks."""51 for chunk in chunks:52 yield chunk53 54 55 class TestMessageSequence:56 """start once, one body per chunk (more_body=True), terminal more_body=False."""57 58 async def test_start_then_chunks_then_terminal(self) -> None:59 sent = await drive(StreamingResponse(gen(b"a", b"b", b"c")))60 assert sent[0]["type"] == "http.response.start"61 assert sent[0]["status"] == 20062 bodies = [m for m in sent[1:]]63 assert [m["body"] for m in bodies] == [b"a", b"b", b"c", b""]64 assert [m["more_body"] for m in bodies] == [True, True, True, False]65 66 async def test_empty_iterator_still_terminates(self) -> None:67 sent = await drive(StreamingResponse(gen()))68 assert sent[0]["type"] == "http.response.start"69 assert len(sent) == 2 # start + terminal only70 assert sent[1]["body"] == b"" and sent[1]["more_body"] is False71 72 async def test_only_one_start(self) -> None:73 sent = await drive(StreamingResponse(gen(b"x", b"y")))74 assert sum(1 for m in sent if m["type"] == "http.response.start") == 175 76 77 class TestHeaders:78 """media_type -> content-type; explicit headers preserved; text gets charset."""79 80 async def test_media_type_becomes_content_type(self) -> None:81 sent = await drive(StreamingResponse(gen(b"x"), media_type="application/octet-stream"))82 headers = dict(sent[0]["headers"])83 assert headers[b"content-type"] == b"application/octet-stream"84 85 async def test_text_media_type_gets_charset(self) -> None:86 sent = await drive(StreamingResponse(gen(b"x"), media_type="text/plain"))87 headers = dict(sent[0]["headers"])88 assert headers[b"content-type"] == b"text/plain; charset=utf-8"89 90 async def test_explicit_headers_preserved(self) -> None:91 resp = StreamingResponse(gen(b"x"), headers=[("x-custom", "v")], media_type="text/plain")92 resp.set_header("x-late", "w")93 sent = await drive(resp)94 headers = dict(sent[0]["headers"])95 assert headers[b"x-custom"] == b"v" and headers[b"x-late"] == b"w"96 97 async def test_custom_status(self) -> None:98 sent = await drive(StreamingResponse(gen(b"x"), status_code=206))99 assert sent[0]["status"] == 206100 101 102 class TestResponseUntouched:103 """Response stays buffered: two messages, no more_body (the whole point)."""104 105 async def test_response_two_messages_no_more_body(self) -> None:106 sent = await drive(Response(content="hi", media_type="text/plain"))107 assert len(sent) == 2108 assert sent[1]["type"] == "http.response.body" and sent[1]["body"] == b"hi"109 assert "more_body" not in sent[1]110 111 def test_response_has_no_streaming_attrs(self) -> None:112 assert not hasattr(Response, "body_iterator")113 assert not hasattr(StreamingResponse, "set_result")