RRRosmontis
34089d0042
Include utility modules, shared channels, prompts, read-only skills, wardrobe memory and regression tests. Externalize legacy credentials and exclude private runtime state.
297 lines
13 KiB
Python
297 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
import asyncio
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
BACKEND = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(BACKEND / "bot-utility"), str(BACKEND)]
|
|
|
|
import notebook as notebook_module
|
|
from notebook import HELP_TEXT, Notebook, SHANGHAI, format_memo, format_todo
|
|
|
|
|
|
NOW = datetime(2030, 1, 2, 12, 0, 0, tzinfo=SHANGHAI).timestamp()
|
|
HOUR = 3600
|
|
DAY = 24 * HOUR
|
|
|
|
|
|
def private_ctx(uid=1, channel="telegram"):
|
|
return SimpleNamespace(user_id=uid, channel=channel, msg_type="private", target_id=uid)
|
|
|
|
|
|
def group_ctx(target_id=100, channel="telegram", msg_type="group", uid=1):
|
|
return SimpleNamespace(
|
|
user_id=uid,
|
|
channel=SimpleNamespace(name=channel),
|
|
msg_type=msg_type,
|
|
target_id=target_id,
|
|
)
|
|
|
|
|
|
def iso_at(epoch):
|
|
return datetime.fromtimestamp(epoch, SHANGHAI).isoformat()
|
|
|
|
|
|
class NotebookTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tempdir = tempfile.TemporaryDirectory()
|
|
self.db_path = os.path.join(self.tempdir.name, "notebook-test.db")
|
|
self.nb = Notebook(self.db_path)
|
|
self.ctx = private_ctx(1)
|
|
|
|
def tearDown(self):
|
|
self.nb.close()
|
|
self.tempdir.cleanup()
|
|
|
|
def add_todo(self, content, due, expires=None, created_at=NOW, ctx=None):
|
|
with mock.patch("notebook.time.time", return_value=created_at):
|
|
record = self.nb.add_todo(
|
|
ctx or self.ctx,
|
|
content,
|
|
iso_at(due),
|
|
None if expires is None else iso_at(expires),
|
|
)
|
|
self.assertEqual(record["content"], content)
|
|
self.assertEqual(record["status"], "active")
|
|
return record["id"]
|
|
|
|
def test_memo_add_query_delete_and_content_limit(self):
|
|
with mock.patch("notebook.time.time", side_effect=[NOW, NOW + 1]):
|
|
first_record = self.nb.add_memo(self.ctx, "购买牛奶")
|
|
second_record = self.nb.add_memo(self.ctx, "购买咖啡豆")
|
|
first = first_record["id"]
|
|
second = second_record["id"]
|
|
self.assertEqual(first_record["content"], "购买牛奶")
|
|
self.assertEqual(first_record["created_at"], NOW)
|
|
self.assertRegex(first, r"^[0-9a-f]{8}$")
|
|
self.assertNotEqual(first, second)
|
|
self.assertEqual([m["id"] for m in self.nb.list_memos(self.ctx)], [second, first])
|
|
self.assertEqual(
|
|
[m["id"] for m in self.nb.list_memos(self.ctx, "牛奶")], [first]
|
|
)
|
|
self.assertEqual(self.nb.list_memos(self.ctx, limit=0), [])
|
|
with self.assertRaises(ValueError):
|
|
self.nb.add_memo(self.ctx, "x" * 2001)
|
|
self.assertEqual(self.nb.delete_memo(self.ctx, first[:4]), first_record)
|
|
with self.assertRaises(ValueError):
|
|
self.nb.delete_memo(self.ctx, first)
|
|
self.assertEqual([m["id"] for m in self.nb.list_memos(self.ctx)], [second])
|
|
with sqlite3.connect(self.db_path) as con:
|
|
self.assertEqual(
|
|
con.execute("SELECT COUNT(*) FROM memo_notes WHERE id=?", (first,)).fetchone()[0],
|
|
0,
|
|
)
|
|
|
|
def test_memo_permissions_and_scope_isolation(self):
|
|
private_id = self.nb.add_memo(private_ctx(1, channel="one"), "私聊记录")["id"]
|
|
# 私聊只按 user_id 隔离，channel/target_id 不参与私聊作用域。
|
|
self.assertEqual(
|
|
self.nb.list_memos(private_ctx(1, channel="two"))[0]["id"], private_id
|
|
)
|
|
with self.assertRaises(ValueError):
|
|
self.nb.delete_memo(private_ctx(2), private_id)
|
|
|
|
group_one = group_ctx(target_id=100, uid=1)
|
|
same_group_other_sender = group_ctx(target_id=100, uid=999)
|
|
group_two = group_ctx(target_id=200, uid=1)
|
|
group_record = self.nb.add_memo(group_one, "群记录")
|
|
group_id = group_record["id"]
|
|
self.assertEqual(self.nb.list_memos(same_group_other_sender)[0]["id"], group_id)
|
|
self.assertEqual(self.nb.list_memos(group_two), [])
|
|
with self.assertRaises(ValueError):
|
|
self.nb.delete_memo(group_two, group_id)
|
|
with self.assertRaises(ValueError):
|
|
self.nb.delete_memo(group_ctx(channel="other"), group_id)
|
|
self.assertEqual(
|
|
self.nb.delete_memo(same_group_other_sender, group_id), group_record
|
|
)
|
|
|
|
def test_id_prefix_must_be_unique_within_scope(self):
|
|
with mock.patch(
|
|
"notebook.secrets.token_hex", side_effect=["aaaaaaa1", "aaaaaaa2"]
|
|
):
|
|
first = self.nb.add_memo(self.ctx, "一")
|
|
self.nb.add_memo(self.ctx, "二")
|
|
with self.assertRaisesRegex(ValueError, "不唯一"):
|
|
self.nb.delete_memo(self.ctx, "aaaa")
|
|
# 其他会话中相同前缀不应影响本会话解析。
|
|
other = private_ctx(2)
|
|
with self.assertRaises(ValueError):
|
|
self.nb.delete_memo(other, "aaaa")
|
|
self.assertEqual(self.nb.delete_memo(self.ctx, first["id"]), first)
|
|
|
|
def test_todo_due_default_expiry_and_validation(self):
|
|
due = NOW + 2 * HOUR
|
|
item_id = self.add_todo("提交报告", due)
|
|
item = self.nb.list_todos(self.ctx)[0]
|
|
self.assertEqual(item["id"], item_id)
|
|
self.assertEqual(item["due_at"], due)
|
|
self.assertEqual(item["expires_at"], due + DAY)
|
|
self.assertEqual(item["status"], "active")
|
|
self.assertIsNone(item["completed_at"])
|
|
|
|
explicit_id = self.add_todo("显式过期", NOW + 3 * HOUR, NOW + 4 * HOUR)
|
|
explicit = next(t for t in self.nb.list_todos(self.ctx) if t["id"] == explicit_id)
|
|
self.assertEqual(explicit["expires_at"], NOW + 4 * HOUR)
|
|
with self.assertRaisesRegex(ValueError, "未来"):
|
|
self.add_todo("过去", NOW - 1)
|
|
with self.assertRaisesRegex(ValueError, "晚于"):
|
|
self.add_todo("过期等于到期", NOW + HOUR, NOW + HOUR)
|
|
with self.assertRaises(ValueError):
|
|
self.nb.add_todo(self.ctx, "无效时间", "not-a-time")
|
|
with self.assertRaises(ValueError):
|
|
self.add_todo("x" * 2001, NOW + HOUR)
|
|
|
|
def test_complete_delete_and_permissions(self):
|
|
item_id = self.add_todo("待办权限", NOW + HOUR)
|
|
stranger = private_ctx(2)
|
|
with self.assertRaises(ValueError):
|
|
self.nb.complete_todo(stranger, item_id)
|
|
with self.assertRaises(ValueError):
|
|
self.nb.delete_todo(stranger, item_id)
|
|
self.assertEqual(self.nb.list_todos(self.ctx)[0]["status"], "active")
|
|
|
|
with mock.patch("notebook.time.time", return_value=NOW + 10):
|
|
completed_record = self.nb.complete_todo(self.ctx, item_id[:5])
|
|
self.assertEqual(completed_record["status"], "completed")
|
|
with self.assertRaises(ValueError):
|
|
self.nb.complete_todo(self.ctx, item_id)
|
|
self.assertEqual(self.nb.list_todos(self.ctx), [])
|
|
completed = self.nb.list_todos(self.ctx, include_completed=True)[0]
|
|
self.assertEqual(completed["status"], "completed")
|
|
self.assertEqual(completed["completed_at"], NOW + 10)
|
|
self.assertEqual(self.nb.delete_todo(self.ctx, item_id)["id"], item_id)
|
|
self.assertEqual(self.nb.list_todos(self.ctx, include_completed=True), [])
|
|
with sqlite3.connect(self.db_path) as con:
|
|
self.assertEqual(con.execute("SELECT COUNT(*) FROM todo_items").fetchone()[0], 0)
|
|
|
|
def test_cleanup_physically_deletes_only_expired_active(self):
|
|
expired_id = self.add_todo("过期未完成", NOW + HOUR) # expires = NOW + 25h
|
|
completed_id = self.add_todo("过期已完成", NOW + HOUR)
|
|
active_id = self.add_todo("仍有效", NOW + 48 * HOUR)
|
|
self.assertTrue(self.nb.complete_todo(self.ctx, completed_id))
|
|
|
|
# 等于 expires_at 即过期；completed 即使过期也保留。
|
|
self.assertEqual(self.nb.cleanup_expired(NOW + 25 * HOUR), 1)
|
|
with sqlite3.connect(self.db_path) as con:
|
|
rows = dict(con.execute("SELECT id,status FROM todo_items").fetchall())
|
|
self.assertNotIn(expired_id, rows)
|
|
self.assertEqual(rows[completed_id], "completed")
|
|
self.assertEqual(rows[active_id], "active")
|
|
self.assertEqual(self.nb.cleanup_expired(NOW + 25 * HOUR), 0)
|
|
|
|
def test_expired_todo_cannot_be_completed_to_escape_deletion(self):
|
|
item_id = self.add_todo(
|
|
"过期后不能完成", NOW - 2 * HOUR, NOW - HOUR, created_at=NOW - 2 * DAY
|
|
)
|
|
with mock.patch("notebook.time.time", return_value=NOW):
|
|
with self.assertRaises(ValueError):
|
|
self.nb.complete_todo(self.ctx, item_id)
|
|
with sqlite3.connect(self.db_path) as con:
|
|
self.assertEqual(
|
|
con.execute("SELECT COUNT(*) FROM todo_items WHERE id=?", (item_id,)).fetchone()[0],
|
|
0,
|
|
)
|
|
|
|
def test_context_limits_and_prominent_background_rule(self):
|
|
for index in range(4):
|
|
with mock.patch("notebook.time.time", return_value=NOW + index):
|
|
self.nb.add_memo(self.ctx, f"备忘录{index}")
|
|
for index in range(4):
|
|
self.add_todo(f"待办{index}", NOW + (index + 1) * HOUR)
|
|
|
|
context = self.nb.context_for_chat(1, now=NOW)
|
|
self.assertIn("⚠️ 重要规则", context)
|
|
self.assertIn("这些仅是背景", context)
|
|
self.assertIn("只有用户询问或直接相关时才可参考", context)
|
|
self.assertIn("不主动重复、汇报或改变话题", context)
|
|
self.assertIn("当前聊天优先，不能喧宾夺主", context)
|
|
self.assertNotIn("备忘录0", context)
|
|
for index in range(1, 4):
|
|
self.assertIn(f"备忘录{index}", context)
|
|
for index in range(3):
|
|
self.assertIn(f"待办{index}", context)
|
|
self.assertNotIn("待办3", context)
|
|
self.assertEqual(context.count("｜到期 "), 3)
|
|
|
|
def test_context_overdue_72_hour_rule_query_and_physical_cleanup(self):
|
|
created = NOW - 10 * DAY
|
|
self.add_todo("逾期任务", NOW - HOUR, NOW + DAY, created)
|
|
self.add_todo("边界任务", NOW + 72 * HOUR, NOW + 73 * HOUR, created)
|
|
self.add_todo("过远任务", NOW + 73 * HOUR, NOW + 74 * HOUR, created)
|
|
expired_id = self.add_todo("过期任务", NOW - 2 * HOUR, NOW - HOUR, created)
|
|
completed_id = self.add_todo("已完成任务", NOW + HOUR, NOW + DAY, created)
|
|
self.assertTrue(self.nb.complete_todo(self.ctx, completed_id))
|
|
|
|
context = self.nb.context_for_chat(1, now=NOW)
|
|
self.assertIn("【已逾期】逾期任务", context)
|
|
self.assertIn("边界任务", context)
|
|
self.assertNotIn("过远任务", context)
|
|
self.assertNotIn("过期任务", context)
|
|
self.assertNotIn("已完成任务", context)
|
|
with sqlite3.connect(self.db_path) as con:
|
|
self.assertEqual(
|
|
con.execute("SELECT COUNT(*) FROM todo_items WHERE id=?", (expired_id,)).fetchone()[0],
|
|
0,
|
|
)
|
|
self.assertIn("逾期任务", self.nb.context_for_chat(1, "逾期", now=NOW))
|
|
self.assertIn("逾期任务", self.nb.context_for_chat(1, "不相关的整句聊天", now=NOW))
|
|
|
|
def test_context_private_isolation_and_empty_result(self):
|
|
self.nb.add_memo(self.ctx, "用户一的记录")
|
|
self.nb.add_memo(group_ctx(target_id=100, uid=1), "群记录")
|
|
self.assertEqual(self.nb.context_for_chat(2, now=NOW), "")
|
|
self.assertNotIn("群记录", self.nb.context_for_chat(1, now=NOW))
|
|
|
|
def test_restart_persistence(self):
|
|
memo_id = self.nb.add_memo(self.ctx, "持久备忘录")
|
|
todo_id = self.add_todo("持久待办", NOW + DAY)
|
|
self.nb.close()
|
|
|
|
restarted = Notebook(self.db_path)
|
|
try:
|
|
self.assertEqual(restarted.list_memos(self.ctx)[0]["id"], memo_id["id"])
|
|
self.assertEqual(restarted.list_todos(self.ctx)[0]["id"], todo_id)
|
|
finally:
|
|
restarted.close()
|
|
|
|
def test_formatters_and_help_distinguish_reminders(self):
|
|
with mock.patch("notebook.time.time", return_value=NOW):
|
|
memo_id = self.nb.add_memo(self.ctx, "格式")
|
|
memo = self.nb.list_memos(self.ctx)[0]
|
|
self.assertIn(memo_id["id"], format_memo(memo))
|
|
todo_id = self.add_todo("格式待办", NOW + HOUR)
|
|
todo = self.nb.list_todos(self.ctx)[0]
|
|
self.assertIn(todo_id, format_todo(todo, now=NOW))
|
|
self.assertIn("2030-01-02 13:00", format_todo(todo, now=NOW))
|
|
self.assertIn("已逾期", format_todo(todo, now=NOW + 2 * HOUR))
|
|
self.assertIn("不会到点发送通知", HELP_TEXT)
|
|
self.assertIn("定时提醒", HELP_TEXT)
|
|
|
|
|
|
class CleanupWorkerTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_worker_catches_cleanup_exception_and_continues(self):
|
|
nb = Notebook(":memory:")
|
|
with (
|
|
mock.patch.object(
|
|
nb, "cleanup_expired", side_effect=[RuntimeError("db busy"), 0]
|
|
) as cleanup,
|
|
mock.patch(
|
|
"notebook.asyncio.sleep", side_effect=[None, asyncio.CancelledError]
|
|
),
|
|
):
|
|
with self.assertRaises(asyncio.CancelledError):
|
|
await nb.cleanup_worker(interval=0)
|
|
self.assertEqual(cleanup.call_count, 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|