Leopold
101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
"""Tests for /api/chat/{agent_id} with a mocked LLM.
|
|
|
|
B2 regression: the user's message must reach the LLM (was broken when history
|
|
slice dropped the new turn-message). Test by capturing the messages list
|
|
passed to ``llm.chat`` and asserting the last item is the user's input.
|
|
"""
|
|
import pytest
|
|
|
|
from llm import LLMResponse
|
|
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
def _fake_llm_response(content: str = "FAKE_REPLY", tool_calls=None, usage=None) -> LLMResponse:
|
|
return LLMResponse(content=content, tool_calls=tool_calls or [], usage=usage or {
|
|
"prompt_tokens": 1,
|
|
"completion_tokens": 1,
|
|
"total_tokens": 2,
|
|
"finish_reason": "stop",
|
|
})
|
|
|
|
|
|
async def test_chat_empty_message_returns_422(app_client, auth_headers, seed_agent):
|
|
"""POST /api/chat with empty message -> 422 (B6 min_length=1)."""
|
|
resp = await app_client.post(
|
|
"/api/chat/test_agent",
|
|
json={"message": ""},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 422, resp.text
|
|
|
|
|
|
async def test_chat_message_over_limit_422(app_client, auth_headers, seed_agent):
|
|
"""POST /api/chat with 40000-char message -> 422 (B6 max_length=32000)."""
|
|
resp = await app_client.post(
|
|
"/api/chat/test_agent",
|
|
json={"message": "x" * 40000},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 422, resp.text
|
|
|
|
|
|
async def test_chat_with_mocked_llm_returns_assistant(app_client, auth_headers, seed_agent, monkeypatch, stub_mcp, clean_db):
|
|
"""Stub llm.chat -> 200 with assistant message; messages table has 2 rows (user + assistant)."""
|
|
async def fake_chat(messages, tools=None, model=None, temperature=0.7, max_tokens=2000, **_kw):
|
|
return _fake_llm_response(content="FAKE_REPLY")
|
|
|
|
monkeypatch.setattr("agent.chat", fake_chat)
|
|
|
|
resp = await app_client.post(
|
|
"/api/chat/test_agent",
|
|
json={"message": "hello world"},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
body = resp.json()
|
|
assert body["content"] == "FAKE_REPLY"
|
|
assert "conversation_id" in body
|
|
conv_id = body["conversation_id"]
|
|
|
|
# Verify messages table: 1 user + 1 assistant = 2 rows
|
|
async with clean_db.execute(
|
|
"SELECT role, content FROM messages WHERE conversation_id = ? ORDER BY id ASC",
|
|
(conv_id,),
|
|
) as cur:
|
|
rows = await cur.fetchall()
|
|
assert len(rows) == 2, f"expected 2 rows (user+assistant), got {len(rows)}"
|
|
assert rows[0][0] == "user"
|
|
assert rows[0][1] == "hello world"
|
|
assert rows[1][0] == "assistant"
|
|
assert rows[1][1] == "FAKE_REPLY"
|
|
|
|
|
|
async def test_chat_user_message_in_prompt(app_client, auth_headers, seed_agent, monkeypatch, stub_mcp):
|
|
"""B2 regression: capture messages list passed to llm.chat, assert last item == user input."""
|
|
captured = {}
|
|
|
|
async def capturing_chat(messages, tools=None, model=None, temperature=0.7, max_tokens=2000, **_kw):
|
|
captured["messages"] = list(messages)
|
|
return _fake_llm_response(content="FAKE_REPLY")
|
|
|
|
monkeypatch.setattr("agent.chat", capturing_chat)
|
|
|
|
user_input = "what is the meaning of life, the universe, and everything?"
|
|
resp = await app_client.post(
|
|
"/api/chat/test_agent",
|
|
json={"message": user_input},
|
|
headers=auth_headers,
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
|
|
assert "messages" in captured, "llm.chat was not invoked"
|
|
msgs = captured["messages"]
|
|
assert len(msgs) >= 2
|
|
# First message must be the system prompt
|
|
assert msgs[0]["role"] == "system"
|
|
# Last message must be the user's input (B2 regression)
|
|
assert msgs[-1]["role"] == "user"
|
|
assert msgs[-1]["content"] == user_input
|