test(google_genai): cover trailing SSE event flush in streaming iterator

Add sync and async regression tests for a final event that arrives without a
trailing SSE delimiter, exercising the end-of-stream flush so the last event is
never dropped.
This commit is contained in:
mateo-berri 2026-06-17 03:39:23 +00:00
parent 393d252eef
commit d2a2e15cf5
No known key found for this signature in database

View file

@ -149,3 +149,21 @@ async def test_async_iterator_preserves_multi_field_and_comment_events():
frames = [frame async for frame in iterator]
assert frames == [multi_field, comment]
@pytest.mark.asyncio
async def test_async_iterator_flushes_trailing_event_without_delimiter():
"""A final event missing its trailing delimiter must still be emitted."""
tail = b'data: {"text":"last"}'
iterator = _build_iterator(_chunk_bytes(tail, 5), is_async=True)
frames = [frame async for frame in iterator]
assert frames == [tail]
def test_sync_iterator_flushes_trailing_event_without_delimiter():
tail = b'data: {"text":"last"}'
iterator = _build_iterator(_chunk_bytes(tail, 5), is_async=False)
assert list(iterator) == [tail]