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