tests/test_api_chat.py

"""Tests for the headless ``api_chat`` endpoint exposed via ``gr.api``."""

from __future__ import annotations

import pytest

import chat as chat_mod


def _fake_stream(yields):
    """Build a fake ``stream_response`` that yields the given frames.

    Each frame is ``(ops, assistant_content, reasoning_content,
    tool_calls, request_id)`` — the same tuple shape ``core.chat
    .stream_response`` produces.
    """
    def _gen(_kwargs):
        for frame in yields:
            yield frame
    return _gen


def test_api_chat_yields_cumulative_then_terminal_history(monkeypatch):
    frames = [
        ([{"type": "content_delta", "delta": "Hel"}], "Hel", "", [], "rid"),
        ([{"type": "content_delta", "delta": "lo"}], "Hello", "", [], "rid"),
    ]
    monkeypatch.setattr(chat_mod, "stream_response", _fake_stream(frames))

    out = list(chat_mod.api_chat(message="hi", history=[]))

    # 2 streaming yields + 1 terminal frame (reflecting persisted history).
    assert len(out) == 3
    assert out[0][0] == "Hel"
    assert out[1][0] == "Hello"
    final_content, final_reasoning, final_tool_calls, final_history = out[-1]
    assert final_content == "Hello"
    assert final_reasoning == ""
    assert final_tool_calls == []
    assert final_history[-2] == {"role": "user", "content": "hi"}
    assert final_history[-1]["role"] == "assistant"
    assert final_history[-1]["content"] == "Hello"


def test_api_chat_round_trips_history(monkeypatch):
    frames = [(
        [{"type": "content_delta", "delta": "you said hi"}],
        "you said hi", "", [], "rid",
    )]
    monkeypatch.setattr(chat_mod, "stream_response", _fake_stream(frames))

    history = [
        {"role": "user", "content": "hi"},
        {"role": "assistant", "content": "hello"},
    ]
    out = list(chat_mod.api_chat(message="recap", history=history))

    final_history = out[-1][3]
    assert [m["role"] for m in final_history] == [
        "user", "assistant", "user", "assistant",
    ]
    assert final_history[2] == {"role": "user", "content": "recap"}
    # Caller's list is never mutated.
    assert len(history) == 2


def test_api_chat_surfaces_tool_calls(monkeypatch):
    tc = {
        "id": "call_1",
        "type": "function",
        "function": {"name": "ping", "arguments": "{}"},
    }
    frames = [
        ([{"type": "tool_calls", "tool_calls": [tc]}], "", "", [tc], "rid"),
    ]
    monkeypatch.setattr(chat_mod, "stream_response", _fake_stream(frames))

    out = list(chat_mod.api_chat(message="call it", history=[]))
    final_content, _, final_tool_calls, final_history = out[-1]
    assert final_content == ""
    assert final_tool_calls == [tc]
    # The assistant turn was persisted with tool_calls attached.
    assert final_history[-1]["tool_calls"] == [tc]


def test_api_chat_propagates_reasoning(monkeypatch):
    frames = [(
        [{"type": "reasoning_delta", "delta": "let me think"}],
        "answer", "let me think", [], "rid",
    )]
    monkeypatch.setattr(chat_mod, "stream_response", _fake_stream(frames))

    out = list(chat_mod.api_chat(message="q", history=[]))
    _, reasoning, _, history = out[-1]
    assert reasoning == "let me think"
    assert history[-1].get("reasoning_content") == "let me think"


def test_api_chat_rejects_empty_message():
    with pytest.raises(ValueError, match="non-empty"):
        list(chat_mod.api_chat(message="   "))


def test_api_chat_rejects_invalid_functions_json():
    with pytest.raises(ValueError, match="not valid JSON"):
        list(chat_mod.api_chat(message="hi", functions_json_str="{not json"))


def test_api_chat_emits_terminal_frame_when_stream_yields_nothing(monkeypatch):
    monkeypatch.setattr(chat_mod, "stream_response", _fake_stream([]))

    out = list(chat_mod.api_chat(message="hi", history=[]))
    assert len(out) == 1
    content, reasoning, tool_calls, history = out[0]
    assert content == ""
    assert reasoning == ""
    assert tool_calls == []
    # The user turn is preserved; the assistant turn is the empty
    # placeholder finalize_response persists when nothing streamed.
    assert history[0] == {"role": "user", "content": "hi"}
    assert history[-1]["role"] == "assistant"