From fb4e17b1ea50d92f4f5dbba99347b1b0b18ef544 Mon Sep 17 00:00:00 2001
From: kshitijk4poor <82637225+kshitijk4poor@users.noreply.github.com>
Date: Tue, 4 Aug 2026 13:00:45 +0530
Subject: [PATCH] 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.
---
 tests/gateway/test_api_server.py | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/tests/gateway/test_api_server.py b/tests/gateway/test_api_server.py
index 459bd908e1973..5a17047914db0 100644
--- a/tests/gateway/test_api_server.py
+++ b/tests/gateway/test_api_server.py
@@ -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)