# -*- 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()