"""subsystems.db 单元测试 覆盖： - MemoryDatabase: 增删查 / FTS5 全文检索 / 批量删除 - MemoDatabase: 增删查 / 待执行 / 重复任务 / 即时任务 / 统计 每个测试用独立临时 sqlite 文件，互不污染。 """ from datetime import datetime, timedelta import pytest from subsystems import MemoryDatabase, MemoDatabase from common import constants @pytest.fixture def mem_db(tmp_path): """独立临时文件的 MemoryDatabase。""" db_path = str(tmp_path / "test_memory.db") return MemoryDatabase(db_path=db_path) @pytest.fixture def memo_db(tmp_path): """独立临时文件的 MemoDatabase。""" db_path = str(tmp_path / "test_memo.db") return MemoDatabase(db_path=db_path) # ==================== MemoryDatabase ==================== def test_memory_add_and_get(mem_db): mid = mem_db.add_memory( topic="测试主题", summary="测试摘要", content="详细内容", tags=["t1", "t2"] ) assert isinstance(mid, int) and mid > 0 m = mem_db.get_memory_by_id(mid) assert m is not None assert m["topic"] == "测试主题" assert m["summary"] == "测试摘要" assert m["tags"] == ["t1", "t2"] def test_memory_search_by_keyword(mem_db): mem_db.add_memory(topic="Python learning", summary="learn Python basics", content="variables and functions") mem_db.add_memory(topic="Go learning", summary="learn Go basics", content="goroutine") results = mem_db.search_memories(["Python"]) assert len(results) == 1 assert "Python" in results[0]["topic"] def test_memory_search_match_all(mem_db): mem_db.add_memory(topic="Python database", summary="sqlite usage", content="FTS5 search") mem_db.add_memory(topic="Python network", summary="requests usage", content="HTTP request") # match_all=True：两个关键词都要命中 results = mem_db.search_memories(["Python", "sqlite"], match_all=True) assert len(results) == 1 assert "database" in results[0]["topic"] # match_all=False：任一关键词命中 results = mem_db.search_memories(["Python", "sqlite"], match_all=False) assert len(results) == 2 def test_memory_search_empty_keywords_returns_empty(mem_db): assert mem_db.search_memories([]) == [] def test_memory_search_chinese_keyword(mem_db): """中文关键词应可检索（trigram 分词或 LIKE 兜底）。""" mem_db.add_memory(topic="天气查询", summary="今天天气晴朗", content="气温25度") mem_db.add_memory(topic="会议安排", summary="下午三点开会", content="会议室A") results = mem_db.search_memories(["天气"]) assert len(results) == 1 assert results[0]["topic"] == "天气查询" def test_memory_search_chinese_match_all(mem_db): """中文多关键词 match_all：全部命中才返回。""" mem_db.add_memory(topic="天气查询", summary="今天天气晴朗", content="气温25度") mem_db.add_memory(topic="天气与会议", summary="开会前先查天气", content="会议室A") results = mem_db.search_memories(["天气", "开会"], match_all=True) assert len(results) == 1 assert results[0]["topic"] == "天气与会议" def test_memory_search_like_escape_wildcards(mem_db): """LIKE 兜底路径中 %/_ 应被转义为字面字符，不当作通配符。""" import sqlite3 mem_db.add_memory(topic="progress", summary="100% done") mem_db.add_memory(topic="task_a", summary="plain text") mem_db.add_memory(topic="taskXa", summary="wildcard trap") def like_search(kw): with sqlite3.connect(mem_db.db_path) as conn: conn.row_factory = sqlite3.Row return mem_db._search_via_like(conn.cursor(), [kw], match_all=False, limit=100) # "%" 为字面字符："100%" 只命中 progress，不当作任意串通配 results = like_search("100%") assert [r["topic"] for r in results] == ["progress"] # "_" 为字面字符："ask_" 只命中 task_a；若未转义，taskXa 也会被 %ask_% 误匹配 results = like_search("ask_") assert [r["topic"] for r in results] == ["task_a"] def test_memory_fts_migration_from_unicode61(tmp_path): """旧 unicode61 FTS 表在重新打开时应自动迁移为 trigram 并回灌数据。""" import sqlite3 db_path = str(tmp_path / "legacy.db") # 模拟旧库：unicode61 建表 + 写入数据 conn = sqlite3.connect(db_path) conn.execute('''CREATE TABLE memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, topic TEXT NOT NULL, summary TEXT NOT NULL, content TEXT, start_time TEXT NOT NULL, end_time TEXT NOT NULL, created_at TEXT NOT NULL, tags TEXT, importance INTEGER DEFAULT 1)''') conn.execute('''CREATE VIRTUAL TABLE memories_fts USING fts5(topic, summary, content, tags, content=memories)''') conn.execute("INSERT INTO memories (topic, summary, start_time, end_time, created_at) " "VALUES ('天气查询', '今天天气晴朗', 't', 't', 't')") conn.execute("INSERT INTO memories_fts (rowid, topic, summary) VALUES (1, '天气查询', '今天天气晴朗')") conn.commit() conn.close() # 重新打开 → 自动迁移 → 中文可搜 db = MemoryDatabase(db_path=db_path) results = db.search_memories(["天气"]) assert len(results) == 1 assert results[0]["topic"] == "天气查询" # 迁移后 FTS 表应为 trigram 分词（环境支持时） conn = sqlite3.connect(db_path) sql = conn.execute("SELECT sql FROM sqlite_master WHERE name='memories_fts'").fetchone()[0] fts_count = conn.execute("SELECT COUNT(*) FROM memories_fts").fetchone()[0] conn.close() assert fts_count == 1 # 数据已回灌 try: probe = sqlite3.connect(":memory:") probe.execute("CREATE VIRTUAL TABLE t USING fts5(x, tokenize='trigram')") probe.close() assert "trigram" in sql except sqlite3.OperationalError: pass # 环境不支持 trigram，保持 unicode61 属预期行为 def test_memory_delete(mem_db): mid = mem_db.add_memory(topic="待删除", summary="摘要") assert mem_db.delete_memory(mid) is True assert mem_db.get_memory_by_id(mid) is None # 再次删除应返回 False assert mem_db.delete_memory(mid) is False def test_memory_batch_delete(mem_db): ids = [mem_db.add_memory(topic=f"t{i}", summary=f"s{i}") for i in range(3)] deleted = mem_db.delete_memories_by_ids(ids) assert deleted == 3 for mid in ids: assert mem_db.get_memory_by_id(mid) is None def test_memory_batch_delete_empty_list(mem_db): assert mem_db.delete_memories_by_ids([]) == 0 def test_memory_get_recent(mem_db): for i in range(5): mem_db.add_memory(topic=f"recent_{i}", summary=f"s{i}") recent = mem_db.get_recent_memories(limit=3) assert len(recent) == 3 def test_memory_nonexistent_id_returns_none(mem_db): assert mem_db.get_memory_by_id(99999) is None # ==================== MemoDatabase ==================== def test_memo_add_and_get(memo_db): reminder = (datetime.now() + timedelta(days=1)).strftime(constants.DATETIME_FORMAT) mid = memo_db.add_memo(title="测试任务", content="内容", reminder_time=reminder) assert isinstance(mid, int) and mid > 0 m = memo_db.get_memo(mid) assert m is not None assert m["title"] == "测试任务" assert m["is_completed"] == 0 def test_memo_pending_filter_by_time(memo_db): """未到期的任务不应出现在 pending 列表中。""" future = (datetime.now() + timedelta(days=1)).strftime(constants.DATETIME_FORMAT) past = (datetime.now() - timedelta(days=1)).strftime(constants.DATETIME_FORMAT) memo_db.add_memo(title="未来任务", reminder_time=future) memo_db.add_memo(title="过期任务", reminder_time=past) pending = memo_db.get_pending_memos() titles = [m["title"] for m in pending] assert "过期任务" in titles assert "未来任务" not in titles def test_memo_complete(memo_db): reminder = (datetime.now() + timedelta(days=1)).strftime(constants.DATETIME_FORMAT) mid = memo_db.add_memo(title="一次性任务", reminder_time=reminder) assert memo_db.complete_memo(mid) is True m = memo_db.get_memo(mid) assert m["is_completed"] == 1 def test_memo_complete_nonexistent_returns_false(memo_db): assert memo_db.complete_memo(99999) is False def test_memo_immediate(memo_db): """即时任务应被 get_immediate_memos 返回。""" mid = memo_db.add_memo(title="即时任务", is_immediate=True) imm = memo_db.get_immediate_memos() assert any(m["id"] == mid for m in imm) def test_memo_mark_as_triggered_completes_one_shot(memo_db): """一次性任务触发后应标记完成。""" past = (datetime.now() - timedelta(days=1)).strftime(constants.DATETIME_FORMAT) mid = memo_db.add_memo(title="一次性", reminder_time=past) nxt = memo_db.mark_as_triggered(mid) assert nxt is None # 一次性任务无下次提醒 m = memo_db.get_memo(mid) assert m["is_completed"] == 1 assert m["trigger_count"] == 1 def test_memo_mark_as_triggered_daily_repeat(memo_db): """daily 重复任务触发后应计算下次提醒时间。""" past = (datetime.now() - timedelta(days=1)).strftime(constants.DATETIME_FORMAT) mid = memo_db.add_memo(title="每日任务", reminder_time=past, repeat_type="daily") nxt = memo_db.mark_as_triggered(mid) assert nxt is not None # 下次应在未来 next_dt = datetime.strptime(nxt, constants.DATETIME_FORMAT) assert next_dt > datetime.now() - timedelta(minutes=1) m = memo_db.get_memo(mid) assert m["is_completed"] == 0 # 重复任务不完成 def test_memo_stats(memo_db): past = (datetime.now() - timedelta(days=1)).strftime(constants.DATETIME_FORMAT) future = (datetime.now() + timedelta(days=1)).strftime(constants.DATETIME_FORMAT) memo_db.add_memo(title="过期", reminder_time=past) memo_db.add_memo(title="未来", reminder_time=future) memo_db.add_memo(title="即时", is_immediate=True) stats = memo_db.get_memo_stats() assert stats["total"] == 3 assert stats["pending"] == 3 assert stats["overdue"] == 1 # 只有过期且非即时的 assert stats["immediate"] == 1 def test_memo_get_all_excludes_completed_by_default(memo_db): reminder = (datetime.now() + timedelta(days=1)).strftime(constants.DATETIME_FORMAT) mid = memo_db.add_memo(title="待完成", reminder_time=reminder) memo_db.complete_memo(mid) all_memos = memo_db.get_all_memos(include_completed=False) assert all(m["is_completed"] == 0 for m in all_memos) all_with_completed = memo_db.get_all_memos(include_completed=True) assert any(m["id"] == mid for m in all_with_completed) def test_memo_get_pending_tasks_for_user(memo_db): past = (datetime.now() - timedelta(days=1)).strftime(constants.DATETIME_FORMAT) memo_db.add_memo(title="用户任务", reminder_time=past) tasks = memo_db.get_pending_tasks_for_user() assert len(tasks) >= 1 assert all("id" in t and "title" in t and "is_overdue" in t for t in tasks) assert any(t["title"] == "用户任务" and t["is_overdue"] for t in tasks) def test_memo_add_user_task_disabled(monkeypatch, memo_db): """当 MEMO_ALLOW_USER_TASKS=False 时应抛异常。""" monkeypatch.setattr(type(constants), "MEMO_ALLOW_USER_TASKS", property(lambda self: False)) with pytest.raises(Exception, match="禁用"): memo_db.add_user_task(title="x") def test_memo_custom_repeat_minutes(memo_db): """custom 重复 + minutes 单位应正确计算下次。""" past = (datetime.now() - timedelta(minutes=1)).strftime(constants.DATETIME_FORMAT) mid = memo_db.add_memo( title="每5分钟", reminder_time=past, repeat_type="custom", repeat_interval=5, repeat_interval_unit="minutes" ) nxt = memo_db.mark_as_triggered(mid) assert nxt is not None next_dt = datetime.strptime(nxt, constants.DATETIME_FORMAT) # 下次约在 5 分钟后 delta = next_dt - datetime.now() assert timedelta(minutes=4) < delta < timedelta(minutes=6)