506 lines
19 KiB
Python
506 lines
19 KiB
Python
"""Tests for Portal API routes"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from unittest.mock import AsyncMock
|
||
|
||
# Note: AGENTKIT_WS_TIMEOUT=0 is set in tests/conftest.py (before portal import)
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from agentkit.llm.gateway import LLMGateway
|
||
from agentkit.llm.protocol import LLMResponse, TokenUsage
|
||
from agentkit.server.app import create_app
|
||
from agentkit.chat.sqlite_conversation_store import SqliteConversationStore
|
||
from agentkit.server.routes.portal import CAPABILITY_CATEGORIES
|
||
from agentkit.skills.base import Skill, SkillConfig
|
||
from agentkit.skills.registry import SkillRegistry
|
||
from agentkit.tools.registry import ToolRegistry
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
def mock_llm_gateway():
|
||
gateway = LLMGateway()
|
||
mock_provider = AsyncMock()
|
||
mock_provider.chat.return_value = LLMResponse(
|
||
content='{"skill": "chat_skill", "confidence": 0.9}',
|
||
model="test-model",
|
||
usage=TokenUsage(prompt_tokens=10, completion_tokens=20),
|
||
)
|
||
|
||
# 让 chat_stream 返回 async generator — ReAct 路径需要。
|
||
# 之前 chat_stream 未配置，AsyncMock 返回 coroutine 触发 TypeError
|
||
# "Provider returned a coroutine from chat_stream()" → 执行失败 → 错误消息持久化。
|
||
async def _mock_chat_stream(request):
|
||
yield LLMResponse(
|
||
content='{"skill": "chat_skill", "confidence": 0.9}',
|
||
model="test-model",
|
||
usage=TokenUsage(prompt_tokens=10, completion_tokens=20),
|
||
)
|
||
|
||
mock_provider.chat_stream = _mock_chat_stream
|
||
gateway.register_provider("test", mock_provider)
|
||
return gateway
|
||
|
||
|
||
@pytest.fixture
|
||
def skill_registry():
|
||
return SkillRegistry()
|
||
|
||
|
||
@pytest.fixture
|
||
def tool_registry():
|
||
return ToolRegistry()
|
||
|
||
|
||
@pytest.fixture
|
||
def app(mock_llm_gateway, skill_registry, tool_registry, tmp_path, monkeypatch):
|
||
# 隔离 _conversation_store 到 tmp_path，避免 TestPortalChat 通过
|
||
# TestClient 调 /portal/chat 时污染生产数据库 data/conversations.db。
|
||
# 之前 fixture 没覆盖模块级全局 _conversation_store，导致测试产生的
|
||
# 报错对话（"执行出错: ..."）持久化到生产 SQLite，用户启动服务器后
|
||
# 在对话列表看到大量报错对话，删除后 reload 又出现（因为 pytest 重新污染）。
|
||
test_db = tmp_path / "test_conversations.db"
|
||
test_store = SqliteConversationStore(db_path=str(test_db))
|
||
from agentkit.server.routes import portal as _portal_module
|
||
|
||
monkeypatch.setattr(_portal_module, "_conversation_store", test_store)
|
||
# _sm_adapter 也持有旧 store 的引用 — 重建以指向隔离 store。
|
||
_portal_module._sm_adapter = _portal_module._ConvStoreAsSessionManager(test_store)
|
||
|
||
return create_app(
|
||
llm_gateway=mock_llm_gateway,
|
||
skill_registry=skill_registry,
|
||
tool_registry=tool_registry,
|
||
)
|
||
|
||
|
||
@pytest.fixture
|
||
def client(app):
|
||
return TestClient(app)
|
||
|
||
|
||
def _register_skill(registry: SkillRegistry, name: str = "chat_skill", **kwargs):
|
||
"""Helper to register a skill with sensible defaults."""
|
||
config = SkillConfig(
|
||
name=name,
|
||
agent_type="chat_type",
|
||
task_mode="llm_generate",
|
||
prompt={"identity": "Chat Skill", "instructions": "Handle chat"},
|
||
intent={"keywords": ["chat", "hello"], "description": "A chat skill"},
|
||
**kwargs,
|
||
)
|
||
skill = Skill(config=config)
|
||
registry.register(skill)
|
||
return skill
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ConversationStore unit tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestConversationStore:
|
||
"""Tests for SqliteConversationStore (async, in-memory DB)."""
|
||
|
||
@pytest.fixture
|
||
def store(self, tmp_path):
|
||
return SqliteConversationStore(db_path=str(tmp_path / "test.db"))
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_or_create_new(self, store):
|
||
conv = await store.get_or_create()
|
||
assert conv.id is not None
|
||
assert conv.messages == []
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_or_create_with_id(self, store):
|
||
conv = await store.get_or_create("test-id-123")
|
||
assert conv.id == "test-id-123"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_or_create_reuse(self, store):
|
||
await store.get_or_create("reuse-id")
|
||
await store.add_message("reuse-id", "user", "hello")
|
||
conv2 = await store.get_or_create("reuse-id")
|
||
assert conv2.id == "reuse-id"
|
||
assert len(conv2.messages) == 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_message(self, store):
|
||
conv = await store.get_or_create("msg-id")
|
||
msg = await store.add_message("msg-id", "user", "hello")
|
||
assert msg.role == "user"
|
||
assert msg.content == "hello"
|
||
assert len(conv.messages) == 1
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_add_message_not_found(self, store):
|
||
with pytest.raises(KeyError):
|
||
await store.add_message("nonexistent", "user", "hello")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_history(self, store):
|
||
await store.get_or_create("hist-id")
|
||
await store.add_message("hist-id", "user", "msg1")
|
||
await store.add_message("hist-id", "assistant", "msg2")
|
||
history = await store.get_history("hist-id")
|
||
assert len(history) == 2
|
||
assert history[0].role == "user"
|
||
assert history[1].role == "assistant"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_history_limit(self, store):
|
||
await store.get_or_create("limit-id")
|
||
for i in range(10):
|
||
await store.add_message("limit-id", "user", f"msg{i}")
|
||
history = await store.get_history("limit-id", limit=3)
|
||
assert len(history) == 3
|
||
assert history[0].content == "msg7"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_history_nonexistent(self, store):
|
||
history = await store.get_history("no-such-id")
|
||
assert history == []
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_conversations(self, store):
|
||
await store.get_or_create("conv-a")
|
||
await store.get_or_create("conv-b")
|
||
convs = await store.list_conversations()
|
||
assert len(convs) == 2
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_conversations_limit(self, store):
|
||
for i in range(5):
|
||
await store.get_or_create(f"conv-{i}")
|
||
convs = await store.list_conversations(limit=2)
|
||
assert len(convs) == 2
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_max_conversations_eviction(self, tmp_path):
|
||
store = SqliteConversationStore(db_path=str(tmp_path / "evict.db"), max_conversations=3)
|
||
for i in range(5):
|
||
await store.get_or_create(f"evict-{i}")
|
||
assert len(store._cache) <= 3
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# POST /portal/chat
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPortalChat:
|
||
def test_chat_with_skill_name(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill")
|
||
|
||
response = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "hello", "skill_name": "chat_skill"},
|
||
)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["conversation_id"] is not None
|
||
assert data["matched_skill"] == "chat_skill"
|
||
assert data["routing_method"] == "skill_prefix"
|
||
assert data["confidence"] == 1.0
|
||
assert data["status"] == "completed"
|
||
|
||
def test_chat_with_intent_routing(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill")
|
||
|
||
response = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "hello chat"},
|
||
)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["matched_skill"] is not None
|
||
assert data["routing_method"] is not None
|
||
assert data["conversation_id"] is not None
|
||
|
||
def test_chat_no_skills_available(self, client):
|
||
"""Greeting fast-path works even without skills (DIRECT_CHAT mode)."""
|
||
response = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "hello"},
|
||
)
|
||
# Greeting regex fast-path: no skill needed, returns 200
|
||
assert response.status_code == 200
|
||
|
||
def test_chat_skill_not_found(self, client):
|
||
response = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "hello", "skill_name": "nonexistent_skill"},
|
||
)
|
||
assert response.status_code == 404
|
||
|
||
def test_chat_with_conversation_id(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill")
|
||
|
||
response1 = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "hello", "skill_name": "chat_skill"},
|
||
)
|
||
conv_id = response1.json()["conversation_id"]
|
||
|
||
response2 = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "follow up", "skill_name": "chat_skill", "conversation_id": conv_id},
|
||
)
|
||
assert response2.status_code == 200
|
||
assert response2.json()["conversation_id"] == conv_id
|
||
|
||
def test_chat_with_sources(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill")
|
||
|
||
response = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={
|
||
"message": "search docs",
|
||
"skill_name": "chat_skill",
|
||
"sources": ["wiki", "docs"],
|
||
},
|
||
)
|
||
assert response.status_code == 200
|
||
|
||
def test_chat_error_does_not_persist_error_message(
|
||
self, client, skill_registry, mock_llm_gateway
|
||
):
|
||
"""REST 异常时不持久化错误消息为 assistant 消息。
|
||
|
||
复现：之前 portal.py 的 chat() 在 agent.execute_stream 抛异常时
|
||
把 ``f"执行出错: {e}"`` 作为 assistant 消息持久化到 SQLite，导致
|
||
用户对话列表出现大量报错对话。修复后异常时删除新创建的空对话，
|
||
已有历史的对话保留 user message 但不写入错误消息。
|
||
"""
|
||
_register_skill(skill_registry, "chat_skill")
|
||
|
||
# 让 chat_stream 抛异常，模拟 provider 实现错误
|
||
async def _failing_chat_stream(request):
|
||
raise RuntimeError("simulated provider failure")
|
||
yield # 让 Python 识别为 async generator
|
||
|
||
mock_provider = next(iter(mock_llm_gateway._providers.values()))
|
||
original_stream = mock_provider.chat_stream
|
||
mock_provider.chat_stream = _failing_chat_stream
|
||
|
||
try:
|
||
response = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "this will fail", "skill_name": "chat_skill"},
|
||
)
|
||
finally:
|
||
mock_provider.chat_stream = original_stream
|
||
|
||
# 响应仍返回 200 + 错误消息（让前端显示）
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert "执行出错" in data["message"]
|
||
|
||
# 关键断言：对话不应出现在列表中（异常时已删除空对话）
|
||
list_resp = client.get("/api/v1/portal/conversations")
|
||
convs = list_resp.json()
|
||
matching = [c for c in convs if c.get("id") == data["conversation_id"]]
|
||
assert matching == [], f"对话不应在列表中（异常时应被清理），但找到了: {matching}"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GET /portal/capabilities
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPortalCapabilities:
|
||
def test_capabilities_returns_list(self, client):
|
||
response = client.get("/api/v1/portal/capabilities")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert "capabilities" in data
|
||
caps = data["capabilities"]
|
||
assert len(caps) == len(CAPABILITY_CATEGORIES)
|
||
|
||
def test_capabilities_structure(self, client):
|
||
response = client.get("/api/v1/portal/capabilities")
|
||
caps = response.json()["capabilities"]
|
||
for cap in caps:
|
||
assert "name" in cap
|
||
assert "display_name" in cap
|
||
assert "description" in cap
|
||
assert "icon" in cap
|
||
assert "enabled" in cap
|
||
assert "skill_count" in cap
|
||
|
||
def test_capabilities_with_skills(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill", capabilities=["chat"])
|
||
|
||
response = client.get("/api/v1/portal/capabilities")
|
||
caps = response.json()["capabilities"]
|
||
chat_cap = next(c for c in caps if c["name"] == "chat")
|
||
assert chat_cap["skill_count"] >= 1
|
||
|
||
def test_capabilities_skills_category_always_increments(self, client, skill_registry):
|
||
_register_skill(skill_registry, "some_skill")
|
||
|
||
response = client.get("/api/v1/portal/capabilities")
|
||
caps = response.json()["capabilities"]
|
||
skills_cap = next(c for c in caps if c["name"] == "skills")
|
||
assert skills_cap["skill_count"] >= 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GET /portal/conversations
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPortalConversations:
|
||
def test_list_conversations_empty(self, client):
|
||
response = client.get("/api/v1/portal/conversations")
|
||
assert response.status_code == 200
|
||
assert isinstance(response.json(), list)
|
||
|
||
def test_list_conversations_after_chat(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill")
|
||
client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "hello", "skill_name": "chat_skill"},
|
||
)
|
||
|
||
response = client.get("/api/v1/portal/conversations")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert len(data) >= 1
|
||
assert "id" in data[0]
|
||
assert "message_count" in data[0]
|
||
|
||
def test_list_conversations_limit(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill")
|
||
for i in range(3):
|
||
client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": f"msg{i}", "skill_name": "chat_skill"},
|
||
)
|
||
|
||
response = client.get("/api/v1/portal/conversations?limit=2")
|
||
assert response.status_code == 200
|
||
assert len(response.json()) <= 2
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GET /portal/conversations/{id}
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPortalConversationHistory:
|
||
def test_get_conversation_history(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill")
|
||
chat_resp = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "hello", "skill_name": "chat_skill"},
|
||
)
|
||
conv_id = chat_resp.json()["conversation_id"]
|
||
|
||
response = client.get(f"/api/v1/portal/conversations/{conv_id}")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
# Response is now an IConversation object, not a bare array
|
||
assert "id" in data
|
||
assert "messages" in data
|
||
assert len(data["messages"]) >= 1
|
||
assert data["messages"][0]["role"] in ("user", "assistant")
|
||
assert "content" in data["messages"][0]
|
||
assert "timestamp" in data["messages"][0]
|
||
|
||
def test_get_conversation_not_found(self, client):
|
||
response = client.get("/api/v1/portal/conversations/nonexistent-id")
|
||
assert response.status_code == 404
|
||
|
||
def test_get_conversation_history_limit(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill")
|
||
chat_resp = client.post(
|
||
"/api/v1/portal/chat",
|
||
json={"message": "hello", "skill_name": "chat_skill"},
|
||
)
|
||
conv_id = chat_resp.json()["conversation_id"]
|
||
|
||
response = client.get(f"/api/v1/portal/conversations/{conv_id}?limit=1")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
# Response is now an IConversation object
|
||
assert len(data["messages"]) <= 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# WebSocket /portal/ws
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestPortalWebSocket:
|
||
# NOTE: Starlette TestClient's sync WS client does not properly trigger
|
||
# server-side disconnect when the `with` block exits, causing the server's
|
||
# `receive_text()` to hang indefinitely. These tests are skipped until
|
||
# we migrate to async WS testing (e.g., httpx-async or pytest-asyncio).
|
||
@pytest.mark.skip(reason="Starlette TestClient WS hangs on disconnect")
|
||
def test_ws_connect(self, client):
|
||
with client.websocket_connect("/api/v1/portal/ws") as ws:
|
||
data = ws.receive_json()
|
||
assert data["type"] == "connected"
|
||
assert "conversation_id" in data
|
||
|
||
@pytest.mark.skip(reason="Starlette TestClient WS hangs on disconnect")
|
||
def test_ws_chat_flow(self, client, skill_registry):
|
||
_register_skill(skill_registry, "chat_skill")
|
||
|
||
with client.websocket_connect("/api/v1/portal/ws") as ws:
|
||
# Receive connected message
|
||
connected = ws.receive_json()
|
||
assert connected["type"] == "connected"
|
||
|
||
# Send chat message
|
||
ws.send_json({"type": "chat", "message": "hello chat"})
|
||
|
||
# Should receive routing info
|
||
routing = ws.receive_json()
|
||
assert routing["type"] == "routing"
|
||
assert "skill" in routing
|
||
|
||
# Then receive step/result messages until we get a result
|
||
messages = []
|
||
while True:
|
||
msg = ws.receive_json()
|
||
messages.append(msg)
|
||
if msg["type"] == "result":
|
||
break
|
||
if msg["type"] == "error":
|
||
break
|
||
|
||
# At least one message should have been received
|
||
assert len(messages) >= 1
|
||
|
||
@pytest.mark.skip(reason="Starlette TestClient WS hangs on disconnect")
|
||
def test_ws_cancel(self, client):
|
||
with client.websocket_connect("/api/v1/portal/ws") as ws:
|
||
connected = ws.receive_json()
|
||
assert connected["type"] == "connected"
|
||
|
||
ws.send_json({"type": "cancel"})
|
||
result = ws.receive_json()
|
||
assert result["type"] == "result"
|
||
assert result["data"]["status"] == "cancelled"
|
||
|
||
@pytest.mark.skip(reason="Starlette TestClient WS hangs on disconnect")
|
||
def test_ws_no_skills_error(self, client):
|
||
with client.websocket_connect("/api/v1/portal/ws") as ws:
|
||
connected = ws.receive_json()
|
||
assert connected["type"] == "connected"
|
||
|
||
ws.send_json({"type": "chat", "message": "hello"})
|
||
msg = ws.receive_json()
|
||
assert msg["type"] == "error"
|
||
assert "No skills available" in msg["data"]["message"]
|