mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-17 01:49:15 +08:00
466 lines
17 KiB
Python
466 lines
17 KiB
Python
"""Mock-based unit tests for ConversationRoute in conversation.py.
|
|
|
|
All tests mock ``request`` and ``g`` at the module level so no Quart app
|
|
fixture is required. The Route base class requires a mock ``app`` on the
|
|
context object; register_routes() calls add_url_rule through the mock.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from astrbot.dashboard.routes.conversation import ConversationRoute
|
|
from astrbot.dashboard.routes.route import RouteContext
|
|
|
|
# Sentinel to distinguish "no conv_mgr argument" from "conv_mgr=None".
|
|
_UNSET = object()
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_route_context():
|
|
"""Return a RouteContext with a mock app and config."""
|
|
return RouteContext(
|
|
app=MagicMock(),
|
|
config=MagicMock(),
|
|
)
|
|
|
|
|
|
def _make_mock_args_getter(values: dict):
|
|
"""Build a callable that mimics request.args.get with optional type
|
|
coercion (the pattern used by Quart's MultiDict)."""
|
|
|
|
def get(key, default=None, type=None): # noqa: A002
|
|
val = values.get(key)
|
|
if val is not None and type is not None:
|
|
return type(val)
|
|
return val if val is not None else default
|
|
|
|
return get
|
|
|
|
|
|
def _build_route(conv_mgr=_UNSET):
|
|
"""Return a (ConversationRoute, mock_db, mock_core_lifecycle) tuple.
|
|
|
|
All dependencies are MagicMock based. Pass ``conv_mgr=None`` to simulate
|
|
an unavailable conversation manager; the default creates a fresh mock.
|
|
"""
|
|
ctx = _make_route_context()
|
|
mock_db = MagicMock()
|
|
mock_core = MagicMock()
|
|
if conv_mgr is _UNSET:
|
|
mock_core.conversation_manager = MagicMock()
|
|
else:
|
|
mock_core.conversation_manager = conv_mgr
|
|
route = ConversationRoute(
|
|
context=ctx,
|
|
db_helper=mock_db,
|
|
core_lifecycle=mock_core,
|
|
)
|
|
return route, mock_db, mock_core
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list_conversations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListConversations:
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@patch("astrbot.dashboard.routes.conversation.g")
|
|
@pytest.mark.asyncio
|
|
async def test_default_pagination(self, mock_g, mock_request):
|
|
"""Default page/page_size when no query params are supplied."""
|
|
mock_request.args.get.side_effect = _make_mock_args_getter({})
|
|
mock_g.get.return_value = "testuser"
|
|
|
|
route, _db, _core = _build_route()
|
|
result = await route.list_conversations()
|
|
|
|
assert result["status"] == "ok"
|
|
pag = result["data"]["pagination"]
|
|
assert pag["page"] == 1
|
|
assert pag["page_size"] == 20
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@patch("astrbot.dashboard.routes.conversation.g")
|
|
@pytest.mark.asyncio
|
|
async def test_custom_pagination(self, mock_g, mock_request):
|
|
"""Custom page/page_size passed as query strings."""
|
|
mock_request.args.get.side_effect = _make_mock_args_getter(
|
|
{"page": "3", "page_size": "50"}
|
|
)
|
|
mock_g.get.return_value = "testuser"
|
|
|
|
route, _db, _core = _build_route()
|
|
result = await route.list_conversations()
|
|
|
|
assert result["status"] == "ok"
|
|
pag = result["data"]["pagination"]
|
|
assert pag["page"] == 3
|
|
assert pag["page_size"] == 50
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@patch("astrbot.dashboard.routes.conversation.g")
|
|
@pytest.mark.asyncio
|
|
async def test_page_size_clamped_to_max_100(self, mock_g, mock_request):
|
|
"""page_size beyond 100 is clamped."""
|
|
mock_request.args.get.side_effect = _make_mock_args_getter(
|
|
{"page_size": "999"}
|
|
)
|
|
mock_g.get.return_value = "testuser"
|
|
|
|
route, _db, _core = _build_route()
|
|
result = await route.list_conversations()
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["data"]["pagination"]["page_size"] == 100
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@patch("astrbot.dashboard.routes.conversation.g")
|
|
@pytest.mark.asyncio
|
|
async def test_page_clamped_to_minimum_1(self, mock_g, mock_request):
|
|
"""page below 1 is raised to 1."""
|
|
mock_request.args.get.side_effect = _make_mock_args_getter(
|
|
{"page": "0"}
|
|
)
|
|
mock_g.get.return_value = "testuser"
|
|
|
|
route, _db, _core = _build_route()
|
|
result = await route.list_conversations()
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["data"]["pagination"]["page"] == 1
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@patch("astrbot.dashboard.routes.conversation.g")
|
|
@pytest.mark.asyncio
|
|
async def test_filter_params_passed_to_manager(self, mock_g, mock_request):
|
|
"""Filter query params are split and forwarded."""
|
|
mock_request.args.get.side_effect = _make_mock_args_getter(
|
|
{
|
|
"platforms": "webchat,discord",
|
|
"message_types": "FriendMessage",
|
|
"search": "hello",
|
|
}
|
|
)
|
|
mock_g.get.return_value = "testuser"
|
|
|
|
route, _db, core = _build_route()
|
|
await route.list_conversations()
|
|
|
|
core.conversation_manager.get_filtered_conversations.assert_awaited_once_with(
|
|
page=1,
|
|
page_size=20,
|
|
platforms=["webchat", "discord"],
|
|
message_types=["FriendMessage"],
|
|
search_query="hello",
|
|
exclude_ids=[],
|
|
exclude_platforms=[],
|
|
)
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@patch("astrbot.dashboard.routes.conversation.g")
|
|
@pytest.mark.asyncio
|
|
async def test_error_when_conv_mgr_unavailable(self, mock_g, mock_request):
|
|
"""Returns error when conversation_manager is falsy."""
|
|
mock_g.get.return_value = "testuser"
|
|
|
|
route, _db, _core = _build_route(conv_mgr=None)
|
|
result = await route.list_conversations()
|
|
|
|
assert result["status"] == "error"
|
|
assert "not available" in (result.get("message") or "")
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@patch("astrbot.dashboard.routes.conversation.g")
|
|
@pytest.mark.asyncio
|
|
async def test_db_query_exception_wrapped(self, mock_g, mock_request):
|
|
"""Exception from get_filtered_conversations is caught and returned as error."""
|
|
mock_g.get.return_value = "testuser"
|
|
|
|
route, _db, core = _build_route()
|
|
core.conversation_manager.get_filtered_conversations.side_effect = (
|
|
RuntimeError("db connection lost")
|
|
)
|
|
result = await route.list_conversations()
|
|
|
|
assert result["status"] == "error"
|
|
assert "数据库查询出错" in str(result.get("message") or "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_conv_detail
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetConvDetail:
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_missing_params_returns_error(self, mock_request):
|
|
mock_request.get_json = AsyncMock(return_value={})
|
|
route, _db, _core = _build_route()
|
|
result = await route.get_conv_detail()
|
|
|
|
assert result["status"] == "error"
|
|
assert "缺少必要参数" in str(result.get("message") or "")
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_conversation_not_found(self, mock_request):
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={"user_id": "user1", "cid": "cid1"}
|
|
)
|
|
route, _db, core = _build_route()
|
|
core.conversation_manager.get_conversation = AsyncMock(return_value=None)
|
|
result = await route.get_conv_detail()
|
|
|
|
assert result["status"] == "error"
|
|
assert "不存在" in str(result.get("message") or "")
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_success_with_valid_params(self, mock_request):
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={"user_id": "user1", "cid": "cid1"}
|
|
)
|
|
route, _db, core = _build_route()
|
|
mock_conv = MagicMock()
|
|
mock_conv.title = "test-title"
|
|
mock_conv.persona_id = "p1"
|
|
mock_conv.history = "[]"
|
|
mock_conv.created_at = "2024-01-01"
|
|
mock_conv.updated_at = "2024-01-02"
|
|
core.conversation_manager.get_conversation = AsyncMock(return_value=mock_conv)
|
|
result = await route.get_conv_detail()
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["data"]["title"] == "test-title"
|
|
assert result["data"]["persona_id"] == "p1"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# upd_conv
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUpdConv:
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_missing_params_returns_error(self, mock_request):
|
|
mock_request.get_json = AsyncMock(return_value={})
|
|
route, _db, _core = _build_route()
|
|
result = await route.upd_conv()
|
|
|
|
assert result["status"] == "error"
|
|
assert "缺少必要参数" in str(result.get("message") or "")
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_not_found(self, mock_request):
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={"user_id": "user1", "cid": "cid1"}
|
|
)
|
|
route, _db, core = _build_route()
|
|
core.conversation_manager.get_conversation = AsyncMock(return_value=None)
|
|
result = await route.upd_conv()
|
|
|
|
assert result["status"] == "error"
|
|
assert "不存在" in str(result.get("message") or "")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# del_conv (single + batch)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDelConv:
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_single_missing_params(self, mock_request):
|
|
mock_request.get_json = AsyncMock(return_value={})
|
|
route, _db, _core = _build_route()
|
|
result = await route.del_conv()
|
|
|
|
assert result["status"] == "error"
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_batch_delete(self, mock_request):
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={
|
|
"conversations": [
|
|
{"user_id": "u1", "cid": "c1"},
|
|
{"user_id": "u2", "cid": "c2"},
|
|
]
|
|
}
|
|
)
|
|
route, _db, core = _build_route()
|
|
core.conversation_manager.delete_conversation = AsyncMock()
|
|
result = await route.del_conv()
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["data"]["deleted_count"] == 2
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_batch_with_failures(self, mock_request):
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={
|
|
"conversations": [
|
|
{"user_id": "u1", "cid": "c1"},
|
|
{"user_id": "", "cid": ""}, # missing params
|
|
]
|
|
}
|
|
)
|
|
route, _db, core = _build_route()
|
|
core.conversation_manager.delete_conversation = AsyncMock()
|
|
result = await route.del_conv()
|
|
|
|
assert result["status"] == "ok"
|
|
assert result["data"]["deleted_count"] == 1
|
|
assert result["data"]["failed_count"] == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# update_history
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUpdateHistory:
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_missing_user_id_or_cid(self, mock_request):
|
|
mock_request.get_json = AsyncMock(return_value={"history": []})
|
|
route, _db, _core = _build_route()
|
|
result = await route.update_history()
|
|
|
|
assert result["status"] == "error"
|
|
assert "缺少必要参数" in str(result.get("message") or "")
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_missing_history(self, mock_request):
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={"user_id": "u1", "cid": "c1"}
|
|
)
|
|
route, _db, _core = _build_route()
|
|
result = await route.update_history()
|
|
|
|
assert result["status"] == "error"
|
|
assert "缺少必要参数" in str(result.get("message") or "")
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_json_string_returns_error(self, mock_request):
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={
|
|
"user_id": "u1",
|
|
"cid": "c1",
|
|
"history": "not-json",
|
|
}
|
|
)
|
|
route, _db, _core = _build_route()
|
|
result = await route.update_history()
|
|
|
|
assert result["status"] == "error"
|
|
assert "有效的 JSON" in str(result.get("message") or "")
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_valid_list_history(self, mock_request):
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={
|
|
"user_id": "u1",
|
|
"cid": "c1",
|
|
"history": [{"role": "user", "content": "hello"}],
|
|
}
|
|
)
|
|
route, _db, core = _build_route()
|
|
core.conversation_manager.get_conversation = AsyncMock(
|
|
return_value=MagicMock()
|
|
)
|
|
core.conversation_manager.update_conversation = AsyncMock()
|
|
result = await route.update_history()
|
|
|
|
assert result["status"] == "ok"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# export_conversations
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestExportConversations:
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_empty_list_returns_error(self, mock_request):
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={"conversations": []}
|
|
)
|
|
route, _db, _core = _build_route()
|
|
result = await route.export_conversations()
|
|
|
|
assert result["status"] == "error"
|
|
assert "不能为空" in str(result.get("message") or "")
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@patch("astrbot.dashboard.routes.conversation.send_file")
|
|
@pytest.mark.asyncio
|
|
async def test_export_success(self, mock_send_file, mock_request):
|
|
"""A valid export request returns a file response via send_file."""
|
|
mock_send_file.return_value = {"status": "ok", "_mock_file": True}
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={
|
|
"conversations": [
|
|
{"user_id": "u1", "cid": "c1"},
|
|
]
|
|
}
|
|
)
|
|
route, _db, core = _build_route()
|
|
mock_conv = MagicMock()
|
|
mock_conv.history = "[]"
|
|
mock_conv.title = "t1"
|
|
mock_conv.persona_id = None
|
|
mock_conv.platform_id = "webchat"
|
|
mock_conv.created_at = "2024-01-01"
|
|
mock_conv.updated_at = "2024-01-02"
|
|
core.conversation_manager.get_conversation = AsyncMock(return_value=mock_conv)
|
|
result = await route.export_conversations()
|
|
|
|
assert result["_mock_file"] is True
|
|
mock_send_file.assert_awaited_once()
|
|
|
|
@patch("astrbot.dashboard.routes.conversation.request")
|
|
@pytest.mark.asyncio
|
|
async def test_export_skips_items_missing_params(self, mock_request):
|
|
"""Items without user_id/cid are reported as failures."""
|
|
mock_request.get_json = AsyncMock(
|
|
return_value={
|
|
"conversations": [
|
|
{"user_id": "u1", "cid": "c1"},
|
|
{"user_id": "", "cid": ""}, # missing
|
|
]
|
|
}
|
|
)
|
|
route, _db, core = _build_route()
|
|
core.conversation_manager.get_conversation = AsyncMock(
|
|
side_effect=[
|
|
MagicMock(
|
|
history="[]",
|
|
title="t",
|
|
persona_id=None,
|
|
platform_id="w",
|
|
created_at="",
|
|
updated_at="",
|
|
),
|
|
]
|
|
)
|
|
with patch("astrbot.dashboard.routes.conversation.send_file") as mock_sf:
|
|
mock_sf.return_value = {"_mock": True}
|
|
result = await route.export_conversations()
|
|
|
|
assert result["_mock"] is True
|