RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
94 lines
4.4 KiB
Python
94 lines
4.4 KiB
Python
"""Notebook tool adapter and lightweight chat-context integration tests."""
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
BASE = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(BASE / "bot-utility"), str(BASE)]
|
|
|
|
from agent import ToolContext, ToolExecutor
|
|
from chat import engine
|
|
|
|
|
|
class FakeNotebook:
|
|
def __init__(self):
|
|
self.calls = []
|
|
def add_memo(self, ctx, content):
|
|
self.calls.append(("add_memo", content))
|
|
return {"id": "memo0001", "content": content}
|
|
def list_memos(self, ctx, query=None):
|
|
return [{"id": "memo0001", "content": "钥匙在抽屉"}]
|
|
def delete_memo(self, ctx, item_id):
|
|
return {"id": "memo0001", "content": "钥匙在抽屉"}
|
|
def add_todo(self, ctx, content, due_at, expires_at=None):
|
|
self.calls.append(("add_todo", content, due_at, expires_at))
|
|
return {"id": "todo0001", "content": content, "due_at": due_at}
|
|
def list_todos(self, ctx, include_completed=False):
|
|
return [{"id": "todo0001", "content": "交报告", "due_at": "tomorrow"}]
|
|
def complete_todo(self, ctx, item_id):
|
|
return {"id": "todo0001", "content": "交报告"}
|
|
def delete_todo(self, ctx, item_id):
|
|
return {"id": "todo0001", "content": "交报告"}
|
|
|
|
|
|
class NotebookToolTests(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.fake = FakeNotebook()
|
|
module = SimpleNamespace(
|
|
notebook=self.fake,
|
|
format_memo=lambda item: f"{item['id']} {item['content']}",
|
|
format_todo=lambda item: f"{item['id']} {item['content']}",
|
|
)
|
|
self.executor = ToolExecutor(modules={"memo": module, "todo": module})
|
|
self.ctx = ToolContext(
|
|
channel=SimpleNamespace(name="tg"), msg_type="private",
|
|
user_id=42, target_id=10001,
|
|
)
|
|
|
|
async def test_schema_exposes_memo_and_todo(self):
|
|
names = [item["name"] for item in self.executor.prompt_schema(self.ctx)]
|
|
self.assertIn("memo", names)
|
|
self.assertIn("todo", names)
|
|
|
|
async def test_memo_add_list_delete(self):
|
|
added = await self.executor.execute("memo", {"action": "add", "params": {"content": "钥匙在抽屉"}}, self.ctx)
|
|
self.assertTrue(added.ok)
|
|
listed = await self.executor.execute("memo", {"action": "list", "params": {}}, self.ctx)
|
|
self.assertIn("钥匙在抽屉", listed.safe_message)
|
|
deleted = await self.executor.execute("memo", {"action": "delete", "params": {"id": "memo"}}, self.ctx)
|
|
self.assertTrue(deleted.ok)
|
|
|
|
async def test_todo_add_list_complete_delete(self):
|
|
added = await self.executor.execute(
|
|
"todo", {"action": "add", "params": {"content": "交报告", "due_at": "2026-10-31T18:00:00+08:00"}}, self.ctx
|
|
)
|
|
self.assertTrue(added.ok)
|
|
listed = await self.executor.execute("todo", {"action": "list", "params": {}}, self.ctx)
|
|
self.assertIn("交报告", listed.safe_message)
|
|
completed = await self.executor.execute("todo", {"action": "complete", "params": {"id": "todo"}}, self.ctx)
|
|
self.assertTrue(completed.ok)
|
|
deleted = await self.executor.execute("todo", {"action": "delete", "params": {"id": "todo"}}, self.ctx)
|
|
self.assertTrue(deleted.ok)
|
|
|
|
def test_chat_prompt_contains_bounded_notebook_background(self):
|
|
block = "【独立备忘录与近期约定】仅背景，当前聊天优先，不能喧宾夺主"
|
|
with patch.object(engine, "user_persona_prompt", return_value="persona"), \
|
|
patch.object(engine, "load_user_status", return_value={"pause": False, "sleep": False}), \
|
|
patch.object(engine.memory_store, "get_relation", return_value=None), \
|
|
patch.object(engine.memory_store, "list_entities", return_value=[]), \
|
|
patch.object(engine.memory_store, "recall_facts", return_value=[]), \
|
|
patch.object(engine.memory_store, "migrate_legacy"), \
|
|
patch.object(engine, "get_history_within_budget", return_value=[]), \
|
|
patch.object(engine, "_sticker_prompt_summary", return_value=""), \
|
|
patch.object(engine.notebook.notebook, "context_for_chat", return_value=block) as context:
|
|
prompt = engine.build_system_prompt(42, query="当前聊天")
|
|
context.assert_called_once_with(42, query="当前聊天")
|
|
self.assertIn(block, prompt)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|