spb/doc-api

1"""streamGenerateContent: SSE (alt=sse) and JSON-array framings. Verified live 2026-09-18."""2from __future__ import annotations34import json56from tests.gemini._core_helpers import call, user78BODY = {"contents": user("Count from 1 to 12 separated by spaces."), "generationConfig": {"maxOutputTokens": 32}}91011def _chunks_sse(lines):12    return [json.loads(ln[5:].strip()) for ln in lines if ln.startswith("data:")]131415def test_sse_framing_and_sequence(gemini, models):16    st, gen, hdrs = call(gemini, "POST", f"/v1beta/models/{models['gemini']}:streamGenerateContent?alt=sse", BODY, stream=True, note="test_streaming sse")17    assert st == 200, gen18    assert hdrs.get("Content-Type", "").startswith("text/event-stream")19    lines = list(gen)20    assert all(ln.startswith("data:") or ln.strip() == "" for ln in lines), "no event:/id: lines, no [DONE]"21    chunks = _chunks_sse(lines)22    assert len(chunks) >= 223    rid = {c["responseId"] for c in chunks}24    assert len(rid) == 1 and all("usageMetadata" in c for c in chunks), "responseId constant, usage on every chunk"25    assert all("finishReason" not in c["candidates"][0] for c in chunks[:-1])26    last = chunks[-1]["candidates"][0]27    assert last["finishReason"] in ("STOP", "MAX_TOKENS")28    assert last["content"]["parts"][-1].get("thoughtSignature"), "final chunk carries the thoughtSignature (empty text part)"29    text = "".join(p.get("text", "") for c in chunks for p in c["candidates"][0]["content"]["parts"])30    assert "1" in text and "12" in text313233def test_json_array_framing(gemini, models):34    st, gen, hdrs = call(gemini, "POST", f"/v1beta/models/{models['gemini']}:streamGenerateContent", BODY, stream=True, note="test_streaming json array")35    assert st == 200, gen36    assert hdrs.get("Content-Type", "").startswith("application/json")37    lines = list(gen)38    assert lines[0].startswith("[") and lines[-1].strip() == "]"39    assert any(ln.strip() == "," for ln in lines), "chunks separated by a lone comma line"40    arr = json.loads("\n".join(lines))41    assert isinstance(arr, list) and len(arr) >= 2 and arr[-1]["candidates"][0]["finishReason"]424344def test_alt_sse_on_non_streaming_method(gemini, models):45    st, body, hdrs = call(gemini, "POST", f"/v1beta/models/{models['gemini']}:generateContent?alt=sse", {"contents": user("Reply with OK."), "generationConfig": {"maxOutputTokens": 8}}, note="test_streaming alt=sse on generateContent")46    assert st == 200 and hdrs.get("Content-Type", "").startswith("text/event-stream")47    assert body.startswith(b"data: ")48