fix(test): feed the SSE writers an asyncio queue, not queue.Queue

CI caught a missed caller-shape update. Both PRODUCTION callers of
_write_sse_chat_completion / _write_sse_responses were converted to
ThreadSafeAsyncQueue, but two pre-existing tests in
tests/gateway/test_api_server.py construct the writer's queue
themselves and still passed a stdlib queue.Queue.

The consumer now does 'await asyncio.wait_for(stream_q.get(), ...)',
which on a queue.Queue blocks the thread forever:
test_stream_cancelled_persists_incomplete_snapshot hung until
pytest-timeout killed it (CI reported the whole file as 'no tests
ran (timeout before collection)'). The sibling disconnect test only
survived because it pre-fills before the first await.

tests/gateway/test_api_server.py: 99 passed (was 1 failed + a 60s
hang); with the SSE/api_server suites: 147 passed.
This commit is contained in:
kshitijk4poor 2026-08-04 13:00:45 +05:30 committed by kshitij
parent 7e344dc0dc
commit fb4e17b1ea
1 changed files with 10 additions and 7 deletions

View File

@ -1674,13 +1674,15 @@ class TestResponsesStreaming:
# Patch web.StreamResponse for the duration of the writer call.
import gateway.platforms.api_server as api_mod
import queue as _q
stream_q: _q.Queue = _q.Queue()
# The SSE writers consume an asyncio queue (ThreadSafeAsyncQueue),
# not a plain queue.Queue — a stdlib queue would block the drain
# loop's ``await stream_q.get()`` forever.
stream_q = api_mod.ThreadSafeAsyncQueue()
async def _agent_coro():
# Feed one partial delta into the stream queue...
stream_q.put("partial output")
stream_q.put_nowait("partial output")
# ...then give the drain loop a moment to pick it up before
# raising CancelledError to simulate a server-side cancel.
await asyncio.sleep(0.01)
@ -1745,11 +1747,12 @@ class TestResponsesStreaming:
raise ConnectionResetError("simulated client disconnect")
import gateway.platforms.api_server as api_mod
import queue as _q
stream_q: _q.Queue = _q.Queue()
stream_q.put("some streamed text")
stream_q.put(None) # EOS sentinel
# asyncio queue to match the writers' consumer (see the note in
# test_stream_cancelled_persists_incomplete_snapshot).
stream_q = api_mod.ThreadSafeAsyncQueue()
stream_q.put_nowait("some streamed text")
stream_q.put_nowait(None) # EOS sentinel
async def _agent_coro():
await asyncio.sleep(0.01)