cb75b9b274
多库检索与存储修复: - RRF 融合去重改用 (collection, chunk_id) 复合键，修复同名文件结果被吞 - DocStore 存储路径加 collection 前缀，修复跨库同名切片数据覆盖 - search_multiple 去重改用复合键 - chunk_id 解析改用 rsplit 兼容下划线文件名 上传与版本管理修复: - 同名文件上传改为覆盖模式，自动清理旧切片 - 修复首次上传不创建版本记录 - 修复覆盖上传版本号回退到 v1 - sync ADDED 分支改用动态版本号生成 - _generate_version_id 改为基于全部版本递增 - 废止/恢复操作同步 SQLite 版本记录 - mark_document_as_superseded 改为仅更新 SQLite 删除清理修复: - 删除文档时同步清理 SQLite 版本记录和变更日志 - 删除向量库时同步清理该库所有版本记录 - cleanup 改为清理 SQLite 记录而非 ChromaDB 测试: - test_version_management.py: 27 条版本管理单元测试 - test_edge_cases.py: 28 条边界用例测试 - test_upload_dedup.py: 5 条上传去重测试 - e2e_risk_test.py: 27 条端到端风险测试 文档: - 新增风险边界问题修复注意事项.md（面向后端的对接文档） - 新增向量库边界风险分析.md - 更新多篇现有文档
438 lines
14 KiB
Python
438 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
文档版本管理修复验证测试
|
||
|
||
测试覆盖：
|
||
1. mark_document_as_superseded 只更新 SQLite（不操作 ChromaDB）
|
||
2. deprecate_document 同步 SQLite 版本记录
|
||
3. restore_document 同步 SQLite 版本记录
|
||
4. 版本历史查询一致性
|
||
5. cleanup_superseded_versions 清理 SQLite 记录
|
||
6. 上传覆盖创建版本记录（逻辑验证）
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import sqlite3
|
||
import tempfile
|
||
import shutil
|
||
from datetime import datetime, timedelta
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
passed = 0
|
||
failed = 0
|
||
|
||
|
||
def check(name, condition, detail=""):
|
||
global passed, failed
|
||
if condition:
|
||
print(f" [PASS] {name}")
|
||
passed += 1
|
||
else:
|
||
print(f" [FAIL] {name} {detail}")
|
||
failed += 1
|
||
|
||
|
||
# ======================================================================
|
||
# 辅助：创建临时 SQLite 数据库
|
||
# ======================================================================
|
||
|
||
def create_test_db():
|
||
"""创建内存中的测试数据库"""
|
||
conn = sqlite3.connect(":memory:")
|
||
cursor = conn.cursor()
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS document_versions (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
document_id TEXT NOT NULL,
|
||
collection TEXT,
|
||
version TEXT NOT NULL DEFAULT 'v1',
|
||
content_hash TEXT,
|
||
status TEXT NOT NULL DEFAULT 'active',
|
||
effective_date DATE,
|
||
expiry_date DATE,
|
||
deprecated_date DATETIME,
|
||
deprecated_reason TEXT,
|
||
deprecated_by TEXT,
|
||
change_summary TEXT,
|
||
changed_sections TEXT,
|
||
supersedes TEXT,
|
||
chunk_count INTEGER DEFAULT 0,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
created_by TEXT,
|
||
UNIQUE(document_id, collection, version)
|
||
)
|
||
''')
|
||
cursor.execute('''
|
||
CREATE TABLE IF NOT EXISTS version_change_logs (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
document_id TEXT NOT NULL,
|
||
collection TEXT,
|
||
old_version TEXT,
|
||
new_version TEXT,
|
||
old_status TEXT,
|
||
new_status TEXT,
|
||
change_type TEXT NOT NULL,
|
||
reason TEXT,
|
||
changed_by TEXT,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
conn.commit()
|
||
return conn
|
||
|
||
|
||
def insert_version(conn, collection, doc_id, version, status="active"):
|
||
"""插入版本记录"""
|
||
conn.execute(
|
||
"""INSERT OR REPLACE INTO document_versions
|
||
(document_id, collection, version, status, created_at)
|
||
VALUES (?, ?, ?, ?, ?)""",
|
||
(doc_id, collection, version, status, datetime.now().isoformat())
|
||
)
|
||
conn.commit()
|
||
|
||
|
||
def get_versions(conn, collection, doc_id):
|
||
"""查询版本记录"""
|
||
cursor = conn.execute(
|
||
"""SELECT version, status, deprecated_date, deprecated_reason
|
||
FROM document_versions
|
||
WHERE collection = ? AND document_id = ?
|
||
ORDER BY created_at DESC""",
|
||
(collection, doc_id)
|
||
)
|
||
return cursor.fetchall()
|
||
|
||
|
||
def get_change_logs(conn, collection, doc_id):
|
||
"""查询变更日志"""
|
||
cursor = conn.execute(
|
||
"""SELECT change_type, old_version, new_version, old_status, new_status
|
||
FROM version_change_logs
|
||
WHERE collection = ? AND document_id = ?
|
||
ORDER BY created_at DESC""",
|
||
(collection, doc_id)
|
||
)
|
||
return cursor.fetchall()
|
||
|
||
|
||
# ======================================================================
|
||
# 测试 1：mark_document_as_superseded 参数签名验证
|
||
# ======================================================================
|
||
print("\n=== 测试 1：mark_document_as_superseded 新签名 ===")
|
||
|
||
from knowledge.manager import KnowledgeBaseManager
|
||
import inspect
|
||
|
||
sig = inspect.signature(KnowledgeBaseManager.mark_document_as_superseded)
|
||
params = list(sig.parameters.keys())
|
||
|
||
check("新签名包含 filename 参数",
|
||
"filename" in params,
|
||
f"参数列表: {params}")
|
||
|
||
check("新签名包含 new_version 参数",
|
||
"new_version" in params,
|
||
f"参数列表: {params}")
|
||
|
||
check("旧参数 old_filename 已移除",
|
||
"old_filename" not in params,
|
||
f"参数列表: {params}")
|
||
|
||
check("旧参数 new_filename 已移除",
|
||
"new_filename" not in params,
|
||
f"参数列表: {params}")
|
||
|
||
|
||
# ======================================================================
|
||
# 测试 2：deprecate_document 同步 SQLite（mock 验证）
|
||
# ======================================================================
|
||
print("\n=== 测试 2：deprecate_document SQLite 同步逻辑 ===")
|
||
|
||
from knowledge.document import DocumentMixin
|
||
|
||
|
||
class MockDocumentManager(DocumentMixin):
|
||
"""模拟文档管理器"""
|
||
def __init__(self):
|
||
self.mock_collection = MagicMock()
|
||
self._bm25_indexes = {}
|
||
|
||
def get_collection(self, kb_name):
|
||
return self.mock_collection
|
||
|
||
def rebuild_bm25_index(self, kb_name):
|
||
pass
|
||
|
||
|
||
# 模拟场景：3 个 chunk，active 状态
|
||
mgr = MockDocumentManager()
|
||
mgr.mock_collection.get.return_value = {
|
||
'ids': ['rule_0', 'rule_1', 'rule_2'],
|
||
'metadatas': [
|
||
{'source': 'rule.txt', 'status': 'active', 'version': 'v1'},
|
||
{'source': 'rule.txt', 'status': 'active', 'version': 'v1'},
|
||
{'source': 'rule.txt', 'status': 'active', 'version': 'v1'},
|
||
]
|
||
}
|
||
|
||
# 模拟 SQLite 连接
|
||
test_conn = create_test_db()
|
||
insert_version(test_conn, "test_kb", "rule.txt", "v1", "active")
|
||
|
||
with patch('data.db.get_connection', return_value=test_conn):
|
||
with patch('knowledge.document_versions.get_version_query') as mock_vq:
|
||
mock_vq_inst = MagicMock()
|
||
mock_vq_inst.get_active_version.return_value = MagicMock(version="v1")
|
||
mock_vq.return_value = mock_vq_inst
|
||
|
||
result = mgr.deprecate_document("test_kb", "rule.txt", reason="test deprecate")
|
||
|
||
check("deprecate 返回 success",
|
||
result.get("success") is True,
|
||
f"result: {result}")
|
||
|
||
check("deprecate 标记 3 个 chunks",
|
||
result.get("deprecated_chunks") == 3,
|
||
f"deprecated_chunks: {result.get('deprecated_chunks')}")
|
||
|
||
# 验证 SQLite 中的状态是否更新
|
||
versions = get_versions(test_conn, "test_kb", "rule.txt")
|
||
v1_status = versions[0][1] if versions else None
|
||
check("SQLite v1 状态改为 deprecated",
|
||
v1_status == "deprecated",
|
||
f"status: {v1_status}")
|
||
|
||
v1_deprecated_date = versions[0][2] if versions else None
|
||
check("SQLite deprecated_date 有值",
|
||
v1_deprecated_date is not None,
|
||
f"deprecated_date: {v1_deprecated_date}")
|
||
|
||
|
||
# ======================================================================
|
||
# 测试 3：restore_document 同步 SQLite（mock 验证）
|
||
# ======================================================================
|
||
print("\n=== 测试 3：restore_document SQLite 同步逻辑 ===")
|
||
|
||
mgr2 = MockDocumentManager()
|
||
mgr2.mock_collection.get.return_value = {
|
||
'ids': ['rule_0', 'rule_1'],
|
||
'metadatas': [
|
||
{'source': 'rule.txt', 'status': 'deprecated'},
|
||
{'source': 'rule.txt', 'status': 'deprecated'},
|
||
]
|
||
}
|
||
|
||
with patch('data.db.get_connection', return_value=test_conn):
|
||
with patch('knowledge.document_versions.get_version_query') as mock_vq:
|
||
mock_vq_inst = MagicMock()
|
||
mock_vq_inst.get_document_history.return_value = [
|
||
MagicMock(status=MagicMock(value='deprecated'), version='v1')
|
||
]
|
||
mock_vq.return_value = mock_vq_inst
|
||
|
||
result = mgr2.restore_document("test_kb", "rule.txt")
|
||
|
||
check("restore 返回 success",
|
||
result.get("success") is True,
|
||
f"result: {result}")
|
||
|
||
check("restore 恢复 2 个 chunks",
|
||
result.get("restored_chunks") == 2,
|
||
f"restored_chunks: {result.get('restored_chunks')}")
|
||
|
||
versions = get_versions(test_conn, "test_kb", "rule.txt")
|
||
v1_status = versions[0][1] if versions else None
|
||
check("SQLite v1 状态恢复为 active",
|
||
v1_status == "active",
|
||
f"status: {v1_status}")
|
||
|
||
|
||
# ======================================================================
|
||
# 测试 4：cleanup_superseded_versions 清理 SQLite 记录
|
||
# ======================================================================
|
||
print("\n=== 测试 4：cleanup_superseded_versions SQLite 清理 ===")
|
||
|
||
from knowledge.cleanup import cleanup_superseded_versions
|
||
|
||
test_conn2 = create_test_db()
|
||
# 插入一条过期的 superseded 记录
|
||
old_date = (datetime.now() - timedelta(days=30)).isoformat()
|
||
test_conn2.execute(
|
||
"""INSERT INTO document_versions
|
||
(document_id, collection, version, status, deprecated_date)
|
||
VALUES (?, ?, ?, ?, ?)""",
|
||
("old_report.pdf", "public_kb", "v1", "superseded", old_date)
|
||
)
|
||
# 插入一条近期的 superseded 记录
|
||
recent_date = (datetime.now() - timedelta(days=2)).isoformat()
|
||
test_conn2.execute(
|
||
"""INSERT INTO document_versions
|
||
(document_id, collection, version, status, deprecated_date)
|
||
VALUES (?, ?, ?, ?, ?)""",
|
||
("new_report.pdf", "public_kb", "v1", "superseded", recent_date)
|
||
)
|
||
# 插入一条旧日志
|
||
test_conn2.execute(
|
||
"""INSERT INTO version_change_logs
|
||
(document_id, collection, change_type, created_at)
|
||
VALUES (?, ?, ?, ?)""",
|
||
("old_report.pdf", "public_kb", "supersede", old_date)
|
||
)
|
||
test_conn2.commit()
|
||
|
||
with patch('data.db.get_connection', return_value=test_conn2):
|
||
cleaned = cleanup_superseded_versions(days_to_keep=7)
|
||
|
||
check("清理了 1 条过期 superseded 记录",
|
||
cleaned == 1,
|
||
f"cleaned: {cleaned}")
|
||
|
||
# 验证剩余记录
|
||
remaining = test_conn2.execute(
|
||
"SELECT document_id, status FROM document_versions"
|
||
).fetchall()
|
||
check("近期 superseded 记录保留",
|
||
any(r[0] == "new_report.pdf" and r[1] == "superseded" for r in remaining),
|
||
f"remaining: {remaining}")
|
||
|
||
check("过期 superseded 记录已删除",
|
||
not any(r[0] == "old_report.pdf" for r in remaining),
|
||
f"remaining: {remaining}")
|
||
|
||
# 验证日志也被清理
|
||
logs_remaining = test_conn2.execute(
|
||
"SELECT COUNT(*) FROM version_change_logs"
|
||
).fetchone()[0]
|
||
check("过期变更日志已清理",
|
||
logs_remaining == 0,
|
||
f"logs_remaining: {logs_remaining}")
|
||
|
||
|
||
# ======================================================================
|
||
# 测试 5：_filter_deprecated_chunks 过滤逻辑不变
|
||
# ======================================================================
|
||
print("\n=== 测试 5：_filter_deprecated_chunks 过滤逻辑 ===")
|
||
|
||
from core.engine import RAGEngine
|
||
|
||
# 模拟搜索结果
|
||
mock_results = {
|
||
'ids': [['id1', 'id2', 'id3', 'id4']],
|
||
'documents': [['doc1', 'doc2', 'doc3', 'doc4']],
|
||
'metadatas': [[
|
||
{'source': 'a.txt', 'status': 'active'},
|
||
{'source': 'b.txt', 'status': 'deprecated'},
|
||
{'source': 'c.txt'}, # 无 status 字段，默认 active
|
||
{'source': 'd.txt', 'status': 'superseded'},
|
||
]],
|
||
'distances': [[0.1, 0.2, 0.3, 0.4]],
|
||
}
|
||
|
||
# 直接调用实例方法（传 None 作为 self，方法中未使用 self）
|
||
filter_fn = RAGEngine._filter_deprecated_chunks
|
||
filtered = filter_fn(None, mock_results)
|
||
|
||
check("过滤后保留 2 条（active + 无status）",
|
||
len(filtered['ids'][0]) == 2,
|
||
f"ids: {filtered['ids'][0]}")
|
||
|
||
check("保留的 id 是 id1 和 id3",
|
||
filtered['ids'][0] == ['id1', 'id3'],
|
||
f"ids: {filtered['ids'][0]}")
|
||
|
||
check("deprecated 被过滤",
|
||
'id2' not in filtered['ids'][0],
|
||
"")
|
||
|
||
check("superseded 被过滤",
|
||
'id4' not in filtered['ids'][0],
|
||
"")
|
||
|
||
|
||
# ======================================================================
|
||
# 测试 6：upload 覆盖版本记录逻辑验证
|
||
# ======================================================================
|
||
print("\n=== 测试 6：upload 覆盖 + sync 版本记录逻辑（代码路径验证）===")
|
||
|
||
# 验证 document_routes.py 中 replaced 分支有 superseded 标记代码
|
||
routes_file = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"api", "document_routes.py"
|
||
)
|
||
with open(routes_file, 'r', encoding='utf-8') as f:
|
||
source = f.read()
|
||
|
||
check("upload_document 中有 get_version_query 调用",
|
||
"get_version_query" in source,
|
||
"")
|
||
|
||
check("upload_document 中有 superseded 状态更新",
|
||
"status='superseded'" in source,
|
||
"")
|
||
|
||
check("upload_document replaced 分支标记旧版本",
|
||
"重新上传覆盖" in source,
|
||
"")
|
||
|
||
# 验证 sync.py ADDED 分支使用自动版本号（不硬编码 v1）
|
||
sync_file2 = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"knowledge", "sync.py"
|
||
)
|
||
with open(sync_file2, 'r', encoding='utf-8') as f:
|
||
sync_src = f.read()
|
||
|
||
check("sync ADDED 分支使用 _generate_version_id",
|
||
"_generate_version_id" in sync_src,
|
||
"")
|
||
|
||
check("sync ADDED 版本记录使用动态版本号",
|
||
"version=new_version" in sync_src,
|
||
"")
|
||
|
||
|
||
# ======================================================================
|
||
# 测试 7：sync.py MODIFIED 流程调用正确性
|
||
# ======================================================================
|
||
print("\n=== 测试 7：sync.py MODIFIED 流程参数验证 ===")
|
||
|
||
sync_file = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"knowledge", "sync.py"
|
||
)
|
||
with open(sync_file, 'r', encoding='utf-8') as f:
|
||
sync_source = f.read()
|
||
|
||
check("MODIFIED 分支中 mark 调用包含 new_version 参数",
|
||
"new_version=new_version" in sync_source,
|
||
"")
|
||
|
||
# 在 MODIFIED 分支内验证调用顺序：先获取 old_version 再生成 new_version
|
||
modified_marker = "elif change.change_type == ChangeType.MODIFIED:"
|
||
mod_start = sync_source.index(modified_marker)
|
||
modified_section = sync_source[mod_start:mod_start + 2000]
|
||
check("MODIFIED 分支先获取 old_version 再生成 new_version",
|
||
"_get_current_version" in modified_section and
|
||
"_generate_version_id" in modified_section and
|
||
modified_section.index("_get_current_version") < modified_section.index("_generate_version_id"),
|
||
f"section contains: _get_current_version={'_get_current_version' in modified_section}, _generate_version_id={'_generate_version_id' in modified_section}")
|
||
|
||
check("MODIFIED 分支有 create_version_record 调用",
|
||
"create_version_record" in sync_source,
|
||
"")
|
||
|
||
|
||
# ======================================================================
|
||
# 汇总
|
||
# ======================================================================
|
||
print(f"\n{'='*60}")
|
||
print(f"测试完成: {passed} 通过, {failed} 失败, 共 {passed + failed} 条")
|
||
print(f"{'='*60}")
|
||
|
||
if failed > 0:
|
||
sys.exit(1)
|