brain-of-reese/tests/unit/test_sse_events.py

"""Unit: SSE frame serialization for POST /api/chat (PLAN §4 contract)."""
from __future__ import annotations

import json

from app.api.chat import sse_event
from app.schemas import (
    ChatErrorEvent,
    ChatRetryEvent,
    ChatThinkingEvent,
    ChatToolEvent,
)


def _payload(frame: str) -> dict:
    assert frame.startswith("data: ")
    assert frame.endswith("\n\n")
    return json.loads(frame.removeprefix("data: ").strip())


def test_delta_frame_serializes_exactly() -> None:
    frame = sse_event({"type": "delta", "text": "hi"})
    assert frame == 'data: {"type": "delta", "text": "hi"}\n\n'
    assert _payload(frame) == {"type": "delta", "text": "hi"}


def test_done_frame_carries_full_contract_shape() -> None:
    payload = {
        "type": "done",
        "deflected": False,
        "sources": [{"source": "Homelab", "path": "kubernetes.md", "title": "K8s"}],
        "suggestions": [],
    }
    assert _payload(sse_event(payload)) == payload


def test_error_frame_serializes() -> None:
    frame = sse_event({"type": "error", "detail": "boom"})
    assert _payload(frame) == {"type": "error", "detail": "boom"}


def test_unicode_survives_roundtrip() -> None:
    frame = sse_event({"type": "delta", "text": "🧠 café — \"quoted\""})
    # ensure_ascii=False keeps the frame readable (no \uXXXX escapes).
    assert "🧠 café" in frame
    assert _payload(frame)["text"] == "🧠 café — \"quoted\""


def test_multi_line_text_stays_one_frame() -> None:
    """Newlines inside the JSON payload must be escaped so the frame
    delimiter ``\\n\\n`` remains unambiguous."""
    frame = sse_event({"type": "delta", "text": "line1\nline2\n\n"})
    assert frame.count("\n\n") == 1  # only the frame terminator
    assert _payload(frame)["text"] == "line1\nline2\n\n"


def test_error_event_model_serializes_exact_frame() -> None:
    """The ``ChatErrorEvent`` model is the wire shape of every server-side
    failure the UI's state machine (phase 06) must recover from.

    Phase 114: the additive ``hint`` field serializes ``null`` when
    absent (old clients ignore the field — PLAN §4; the
    ``ChatDoneEvent.related`` pattern)."""
    frame = sse_event(ChatErrorEvent(detail="boom").model_dump())
    assert frame == 'data: {"type": "error", "detail": "boom", "hint": null}\n\n'
    assert _payload(frame) == {"type": "error", "detail": "boom", "hint": None}


def test_error_event_shape_is_type_detail_and_optional_hint() -> None:
    dumped = ChatErrorEvent(detail="The chat model dropped the connection").model_dump()
    assert set(dumped.keys()) == {"type", "detail", "hint"}
    assert dumped["type"] == "error"  # default — call sites never spell it out
    assert dumped["hint"] is None  # absent hint serializes null, not dropped


def test_error_event_hint_serializes_verbatim_when_set() -> None:
    """Phase 114 (TODO L6): the too-long frame carries the reachability-
    fine hint — it survives the roundtrip byte-identically (em-dash and
    all) for the banner to show in place of the default hint."""
    hint = "The app reached the embedding model fine — only the question length is the problem."
    dumped = ChatErrorEvent(
        detail="Question too long — trim it and re-ask.", hint=hint
    ).model_dump()
    assert dumped["hint"] == hint
    payload = _payload(sse_event(dumped))
    assert payload["hint"] == hint
    assert payload["detail"] == "Question too long — trim it and re-ask."


def test_thinking_frame_serializes_exactly() -> None:
    """Phase 17 (PLAN §4 extension): the ``thinking`` frame is exactly
    ``{type: "thinking", text: str}`` — the sibling shape of ``delta``
    the client's readSSE handler will branch on."""
    frame = sse_event(ChatThinkingEvent(text="Step 1: check the docs…").model_dump())
    assert frame == 'data: {"type": "thinking", "text": "Step 1: check the docs…"}\n\n'
    assert _payload(frame) == {"type": "thinking", "text": "Step 1: check the docs…"}


def test_thinking_event_shape_is_type_and_text_only() -> None:
    dumped = ChatThinkingEvent(text="hmm").model_dump()
    assert set(dumped.keys()) == {"type", "text"}
    assert dumped["type"] == "thinking"  # default — call sites never spell it out


def test_tool_frame_serializes_exactly() -> None:
    """Phase 37 (PLAN §4 extension): the ``tool`` frame is exactly
    ``{type: "tool", name: str, argument: str | null}`` — one per
    model-requested document tool call, streamed ahead of the ``delta``
    frames of the answer."""
    frame = sse_event(ChatToolEvent(name="read", argument="S/p.md").model_dump())
    assert frame == 'data: {"type": "tool", "name": "read", "argument": "S/p.md"}\n\n'
    assert _payload(frame) == {"type": "tool", "name": "read", "argument": "S/p.md"}


def test_tool_frame_argument_is_null_for_parameterless_tools() -> None:
    """``ls`` (unscoped) carries no string argument, so its frame's
    ``argument`` serializes as JSON null (the client renders the name
    alone)."""
    dumped = ChatToolEvent(name="ls").model_dump()
    assert dumped == {"type": "tool", "name": "ls", "argument": None}
    assert _payload(sse_event(dumped))["argument"] is None


def test_tool_event_shape_is_type_name_argument_only() -> None:
    dumped = ChatToolEvent(name="read", argument="S/p.md").model_dump()
    assert set(dumped.keys()) == {"type", "name", "argument"}
    assert dumped["type"] == "tool"  # default — call sites never spell it out


def test_tool_frame_range_read_carries_offset_and_limit() -> None:
    """Phase 131 (task 04): a range ``read``'s frame carries the line
    range the model requested — the wire's assertion surface for the
    owner's grep → read(offset/limit) dance (the story E2E pins the
    REAL ints the mock parsed from the grep result)."""
    frame = sse_event(
        ChatToolEvent(
            name="read", argument="S/p.md", offset=1500, limit=30
        ).model_dump()
    )
    assert frame == (
        'data: {"type": "tool", "name": "read", "argument": "S/p.md", '
        '"offset": 1500, "limit": 30}\n\n'
    )
    assert _payload(frame) == {
        "type": "tool",
        "name": "read",
        "argument": "S/p.md",
        "offset": 1500,
        "limit": 30,
    }


def test_tool_event_range_fields_absent_not_null_when_omitted() -> None:
    """Phase 131 (task 04): the ``offset`` / ``limit`` keys are ABSENT
    (never ``null``) when omitted — a plain ``read`` (no range) and a
    ``grep`` (no range parameters at all) both serialize to the
    byte-identical pre-phase shape (the omission pattern of
    ``ChatDoneEvent.answer_image``). A bad range VALUE never reaches
    the frame at all: ``app.api.chat`` applies the whole-line-number
    guard before building the event (the E2E suite pins the frame the
    endpoint emits)."""
    for dumped in (
        ChatToolEvent(name="read", argument="S/p.md").model_dump(),
        ChatToolEvent(name="read", argument="S/p.md", offset=None, limit=None).model_dump(),
        ChatToolEvent(name="grep", argument="trap").model_dump(),
    ):
        assert set(dumped.keys()) == {"type", "name", "argument"}, dumped


def test_retry_frame_serializes_exactly() -> None:
    """Phase 67: the ``retry`` frame is exactly ``{type: "retry",
    attempt: int, max_attempts: int}`` — a transient status the client's
    readSSE handler branches on ("Communication interrupted — retrying
    (n of N)…"), never an error. ``attempt`` is the attempt the server is
    about to try next (1-based); the contract the frontend branch and the
    E2E suite key off, locked here ahead of the JS implementation."""
    frame = sse_event(ChatRetryEvent(attempt=2, max_attempts=4).model_dump())
    assert frame == 'data: {"type": "retry", "attempt": 2, "max_attempts": 4}\n\n'
    assert _payload(frame) == {"type": "retry", "attempt": 2, "max_attempts": 4}


def test_retry_event_shape_is_type_attempt_max_attempts_only() -> None:
    dumped = ChatRetryEvent(attempt=2, max_attempts=4).model_dump()
    assert set(dumped.keys()) == {"type", "attempt", "max_attempts"}
    assert dumped["type"] == "retry"  # default — call sites never spell it out
    assert isinstance(dumped["attempt"], int)
    assert isinstance(dumped["max_attempts"], int)