107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
"""AgentLoop 已持久化用户轮次入口的回归测试。"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
from unittest.mock import patch
|
|
from uuid import uuid4
|
|
|
|
from core.loop import AgentLoop
|
|
|
|
|
|
class _Session:
|
|
def __init__(self, messages=None):
|
|
self.messages = list(messages or [])
|
|
self.appended = []
|
|
self.append_artifacts = []
|
|
|
|
def append(self, message, *, artifact_refs=None):
|
|
self.messages.append(message)
|
|
self.appended.append(message)
|
|
self.append_artifacts.append(artifact_refs)
|
|
return uuid4()
|
|
|
|
|
|
def _loop(session: _Session) -> AgentLoop:
|
|
loop = AgentLoop(
|
|
llm=MagicMock(),
|
|
executor=MagicMock(),
|
|
session=session,
|
|
capabilities=SimpleNamespace(max_iterations=1),
|
|
user_id=uuid4(),
|
|
working_dir=Path("."),
|
|
cancel_check=lambda: True,
|
|
)
|
|
loop._maybe_fold_context = MagicMock()
|
|
return loop
|
|
|
|
|
|
class PersistedTurnTests(unittest.TestCase):
|
|
def test_persisted_turn_does_not_append_duplicate_user_message(self) -> None:
|
|
session = _Session([{"role": "user", "content": "已落库"}])
|
|
result = _loop(session).run_persisted_turn()
|
|
|
|
self.assertEqual(result, "[cancelled]")
|
|
self.assertEqual(session.appended, [])
|
|
self.assertEqual(session.messages, [{"role": "user", "content": "已落库"}])
|
|
|
|
def test_persisted_turn_requires_latest_user_message(self) -> None:
|
|
session = _Session([{"role": "assistant", "content": "旧回复"}])
|
|
|
|
with self.assertRaisesRegex(RuntimeError, "latest message"):
|
|
_loop(session).run_persisted_turn()
|
|
|
|
def test_legacy_run_still_appends_user_message(self) -> None:
|
|
session = _Session()
|
|
result = _loop(session).run("新消息")
|
|
|
|
self.assertEqual(result, "[cancelled]")
|
|
self.assertEqual(
|
|
session.appended,
|
|
[{"role": "user", "content": "新消息"}],
|
|
)
|
|
|
|
def test_final_assistant_persists_explicit_artifact_list(self) -> None:
|
|
session = _Session([{"role": "user", "content": "生成报告"}])
|
|
loop = AgentLoop(
|
|
llm=MagicMock(),
|
|
executor=MagicMock(),
|
|
session=session,
|
|
capabilities=SimpleNamespace(
|
|
max_iterations=1, family="test", variant="model",
|
|
input_cny_per_mtoken=0, output_cny_per_mtoken=0,
|
|
cache_hit_cny_per_mtoken=0,
|
|
),
|
|
user_id=uuid4(),
|
|
working_dir=Path("."),
|
|
)
|
|
loop._maybe_fold_context = MagicMock()
|
|
loop._pending_artifact_refs = [{
|
|
"version": 1, "scope": "working_dir", "path": "report.pdf",
|
|
}]
|
|
# _run resets turn state; emulate a published tool by restoring the pending ref when
|
|
# the final model response is received.
|
|
response = SimpleNamespace(
|
|
choices=[SimpleNamespace(message=SimpleNamespace(
|
|
content="已完成", tool_calls=None,
|
|
))],
|
|
usage=None,
|
|
)
|
|
loop._stream_llm = MagicMock(return_value=(response, False))
|
|
original_fold = loop._maybe_fold_context
|
|
original_fold.side_effect = lambda: loop._pending_artifact_refs.append({
|
|
"version": 1, "scope": "working_dir", "path": "report.pdf",
|
|
})
|
|
with patch("core.loop.record_chat_usage"):
|
|
result = loop.run_persisted_turn()
|
|
self.assertEqual(result, "已完成")
|
|
self.assertEqual(session.append_artifacts[-1], [{
|
|
"version": 1, "scope": "working_dir", "path": "report.pdf",
|
|
}])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|