"""streamGenerateContent: SSE (alt=sse) and JSON-array framings. Verified live 2026-09-18."""
from __future__ import annotations

import json

from tests.gemini._core_helpers import call, user

BODY = {"contents": user("Count from 1 to 12 separated by spaces."), "generationConfig": {"maxOutputTokens": 32}}


def _chunks_sse(lines):
    return [json.loads(ln[5:].strip()) for ln in lines if ln.startswith("data:")]


def test_sse_framing_and_sequence(gemini, models):
    st, gen, hdrs = call(gemini, "POST", f"/v1beta/models/{models['gemini']}:streamGenerateContent?alt=sse", BODY, stream=True, note="test_streaming sse")
    assert st == 200, gen
    assert hdrs.get("Content-Type", "").startswith("text/event-stream")
    lines = list(gen)
    assert all(ln.startswith("data:") or ln.strip() == "" for ln in lines), "no event:/id: lines, no [DONE]"
    chunks = _chunks_sse(lines)
    assert len(chunks) >= 2
    rid = {c["responseId"] for c in chunks}
    assert len(rid) == 1 and all("usageMetadata" in c for c in chunks), "responseId constant, usage on every chunk"
    assert all("finishReason" not in c["candidates"][0] for c in chunks[:-1])
    last = chunks[-1]["candidates"][0]
    assert last["finishReason"] in ("STOP", "MAX_TOKENS")
    assert last["content"]["parts"][-1].get("thoughtSignature"), "final chunk carries the thoughtSignature (empty text part)"
    text = "".join(p.get("text", "") for c in chunks for p in c["candidates"][0]["content"]["parts"])
    assert "1" in text and "12" in text


def test_json_array_framing(gemini, models):
    st, gen, hdrs = call(gemini, "POST", f"/v1beta/models/{models['gemini']}:streamGenerateContent", BODY, stream=True, note="test_streaming json array")
    assert st == 200, gen
    assert hdrs.get("Content-Type", "").startswith("application/json")
    lines = list(gen)
    assert lines[0].startswith("[") and lines[-1].strip() == "]"
    assert any(ln.strip() == "," for ln in lines), "chunks separated by a lone comma line"
    arr = json.loads("\n".join(lines))
    assert isinstance(arr, list) and len(arr) >= 2 and arr[-1]["candidates"][0]["finishReason"]


def test_alt_sse_on_non_streaming_method(gemini, models):
    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")
    assert st == 200 and hdrs.get("Content-Type", "").startswith("text/event-stream")
    assert body.startswith(b"data: ")