qing
7f21b65428
- db.py: FTS5 改用 trigram 分词（中文可检索），旧 unicode61 表自动迁移回灌；search_memories 增加 LIKE 兜底（%/_ 转义、MATCH 异常防护） - conversation_compressor.py: should_compress 补齐 Token 阈值判断（原为死配置），Token 触发时按比例压缩早期消息 - ConversationCompressor 抽离至 subsystems，注入 llm_client 打破与 DeepSeekChat 循环引用；aibox.py 3517→3018 行 - 新增 21 项测试（中文检索/旧表迁移/LIKE 转义/token 估算/压缩流程/失败回退），全量 232 项通过
313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""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)
|