test: 更新 AgentScope 相关单元测试与集成测试

- 重命名 test_react_runner.py 为 test_runner.py
- 新增 test_utils.py 测试工具函数
- 更新现有测试用例适配新架构
This commit is contained in:
qzl
2026-03-16 16:11:06 +08:00
parent 36b104fa37
commit e55f12cdc1
15 changed files with 753 additions and 717 deletions
+67 -4
View File
@@ -45,6 +45,12 @@ async def test_snapshot_message_returns_raw_db_columns() -> None:
seq=7,
role=AgentChatMessageRole.TOOL,
content='{"offloaded":true}',
model_code=None,
tool_name=None,
input_tokens=0,
output_tokens=0,
cost=0,
latency_ms=None,
metadata_json={"tool_call_id": "call-1"},
created_at=now,
)
@@ -71,8 +77,7 @@ async def test_persist_user_message_sets_session_title_when_empty() -> None:
await repository.persist_user_message(
session_id=session_id,
run_id="run-1",
content_text=" 请帮我安排明天下午开会 ",
content=" 请帮我安排明天下午开会 ",
metadata=None,
)
@@ -94,10 +99,68 @@ async def test_persist_user_message_keeps_existing_session_title() -> None:
await repository.persist_user_message(
session_id=session_id,
run_id="run-2",
content_text="新的消息内容",
content="新的消息内容",
metadata=None,
)
assert session_row.title == "已有标题"
assert session_row.message_count == 2
class _ScalarRows:
def __init__(self, rows: list[object]) -> None:
self._rows = rows
def all(self) -> list[object]:
return self._rows
class _ExecuteRowsResult:
def __init__(self, rows: list[object]) -> None:
self._rows = rows
def scalars(self) -> _ScalarRows:
return _ScalarRows(self._rows)
class _FakeHistorySession:
def __init__(self) -> None:
self._execute_count = 0
async def execute(self, stmt): # noqa: ANN001
del stmt
self._execute_count += 1
if self._execute_count == 1:
return _ExecuteResult(datetime(2026, 3, 16, 11, 0, tzinfo=timezone.utc))
if self._execute_count == 2:
message = SimpleNamespace(
id=uuid4(),
seq=1,
role=AgentChatMessageRole.USER,
content="hello",
model_code=None,
tool_name=None,
input_tokens=0,
output_tokens=0,
cost=0,
latency_ms=None,
metadata_json=None,
created_at=datetime(2026, 3, 16, 11, 0, tzinfo=timezone.utc),
)
return _ExecuteRowsResult([message])
return _ExecuteResult(uuid4())
@pytest.mark.asyncio
async def test_get_history_day_uses_target_day_queries_only() -> None:
session = _FakeHistorySession()
repository = AgentRepository(session=session) # type: ignore[arg-type]
payload = await repository.get_history_day(session_id=str(uuid4()), before=None)
assert payload is not None
assert payload["day"] == "2026-03-16"
assert payload["hasMore"] is True
messages = payload["messages"]
assert isinstance(messages, list)
assert len(messages) == 1