httpx.Response.iter_text has empty string for iterator's last item
#2995
Answered
by
jamesbraza
jamesbraza
asked this question in
Potential Issue
-
|
When using Please see the below Python 3.11 code, its assertion checking for empty strings fails: """
Demo of properly unit testing a starlette StreamingResponse.
httpx==0.25.2
pytest==7.4.3
starlette==0.27.0
uvicorn==0.24.0.post1
"""
import asyncio
import statistics
import time
from collections.abc import Iterator
from threading import Thread
import httpx
import pytest
from starlette.responses import StreamingResponse
from uvicorn import Config, Server
# SEE: https://www.starlette.io/responses/#streamingresponse
async def slow_numbers(minimum, maximum):
yield "<html><body><ul>"
for number in range(minimum, maximum + 1):
yield "<li>%d</li>" % number
await asyncio.sleep(0.5)
yield "</ul></body></html>"
async def app(scope, receive, send):
assert scope["type"] == "http"
response = StreamingResponse(slow_numbers(1, 5), media_type="text/html")
await response(scope, receive, send)
# SEE: https://github.com/encode/httpx/blob/0.25.2/tests/conftest.py#L230-L293
# Workaround for https://github.com/encode/starlette/issues/1102
class TestServer(Server):
@property
def url(self) -> httpx.URL:
protocol = "https" if self.config.is_ssl else "http"
return httpx.URL(f"{protocol}://{self.config.host}:{self.config.port}/")
def install_signal_handlers(self) -> None:
# Disable the default installation of handlers for signals such as SIGTERM,
# because it can only be done in the main thread.
pass
async def serve(self, sockets=None):
self.restart_requested = asyncio.Event()
loop = asyncio.get_event_loop()
tasks = {
loop.create_task(super().serve(sockets=sockets)),
loop.create_task(self.watch_restarts()),
}
await asyncio.wait(tasks)
async def restart(self) -> None: # pragma: no cover
# This coroutine may be called from a different thread than the one the
# server is running on, and from an async environment that's not asyncio.
# For this reason, we use an event to coordinate with the server
# instead of calling shutdown()/startup() directly, and should not make
# any asyncio-specific operations.
self.started = False
self.restart_requested.set()
while not self.started:
await asyncio.sleep(0.2)
async def watch_restarts(self) -> None: # pragma: no cover
while True:
if self.should_exit:
return
try:
await asyncio.wait_for(self.restart_requested.wait(), timeout=0.1)
except asyncio.TimeoutError:
continue
self.restart_requested.clear()
await self.shutdown()
await self.startup()
def serve_in_thread(server: TestServer) -> Iterator[TestServer]:
thread = Thread(target=server.run)
thread.start()
try:
while not server.started:
time.sleep(1e-3)
yield server
finally:
server.should_exit = True
thread.join()
@pytest.fixture(name="server", scope="session")
def fixture_server() -> Iterator[TestServer]:
config = Config(app=app, lifespan="off", loop="asyncio")
server = TestServer(config=config)
yield from serve_in_thread(server)
# The actual test
def test_streaming(server: TestServer) -> None:
client = httpx.Client(base_url=server.url)
with client.stream("GET", "/") as response:
response: httpx.Response
texts, times = [], []
tic = time.perf_counter()
for text in response.iter_text():
texts.append(text)
times.append((toc := time.perf_counter()) - tic)
tic = toc
assert len(times) > 1, "Should be more than one chunk"
assert times[0] < 0.6, "Perhaps you streamed everything in first chunk"
assert statistics.mean(times) < 0.6, "Should be streaming"
assert all([bool(text) for text in texts]), "Some text was empty" |
Beta Was this translation helpful? Give feedback.
Answered by
jamesbraza
Dec 11, 2023
Replies: 1 comment 1 reply
-
|
Ah interesting. Yeah it'd be neater if it didn't ever return any empty components. The test cases for handling this are in |
Beta Was this translation helpful? Give feedback.
1 reply
-
|
Thanks! Adding this test case to @pytest.mark.parametrize(
["data", "expected"],
[((b"Hello,", b" world!"), ["Hello,", " world!"])],
)
def test_streaming_iter_text(data: typing.Iterable[bytes], expected: typing.List[str]):
response = httpx.Response(200, content=iter(data))
assert list(response.iter_text()) == expectedThe issue seems to come from
I traced the cause to some missing logic in |
Beta Was this translation helpful? Give feedback.
Answer selected by
tomchristie
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Thanks! Adding this test case to
tests/test_decoders.pywill expose the bug:The issue seems to come from
TextChunker.decode:IdentityDecoderandTextDecoder) emits a empty valueByteChunker().decode(b"")returns[], butTextChunker().decode("")returns[""]I traced the cause to some missing logic in
TextChunker.decodethatByteChunker.decodehas, so I opened #2998