Spaces:
Running
Running
File size: 4,206 Bytes
ad51766 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """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"
|