66 lines
1.9 KiB
Python
66 lines
1.9 KiB
Python
"""AgentLoop 已持久化用户轮次入口的回归测试。"""
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock
|
|
from uuid import uuid4
|
|
|
|
from core.loop import AgentLoop
|
|
|
|
|
|
class _Session:
|
|
def __init__(self, messages=None):
|
|
self.messages = list(messages or [])
|
|
self.appended = []
|
|
|
|
def append(self, message):
|
|
self.messages.append(message)
|
|
self.appended.append(message)
|
|
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": "新消息"}],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|