kplam
bdd30f0a88
- deps.py: _create_redis_client 增加同步 ping 校验，Redis 不可达时正确降级为内存模式 - main.py: lifespan 复用 deps 的 UserStore 单例，避免 admin 与 API 请求实例不一致 - 前端: 强制改密弹窗（must_change_password 用户）、兼容新旧登录返回格式、markPasswordChanged - 新增 NAS SSH 部署脚本与批处理测试；gitignore 前端构建产物
340 lines
14 KiB
Python
340 lines
14 KiB
Python
"""批量上传与入库进度集成验证
|
||
|
||
在真实内存 Qdrant + FakeOllama + 确定性向量环境下，全链路验证：
|
||
- POST /documents/upload-batch 批量上传闭环（全部成功 / 部分失败）
|
||
- GET /documents/tasks 任务列表（按 updated_at 降序、每项含 filename/doc_id）
|
||
- POST /documents/{doc_id}/reingest 重新入库（成功 / 边界 1001/1004）
|
||
|
||
复用 tests/test_e2e_integration.py 的 FakeOllama / DeterministicEmbedding / FakeCache；
|
||
upload_dir 用 tmp_path 真实落盘，monkeypatch 模块级 _task_manager / _qdrant 单例。
|
||
"""
|
||
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from qdrant_client import AsyncQdrantClient
|
||
|
||
from app.api.v1 import document as document_module
|
||
from app.config import Settings
|
||
from app.core.chunker import Chunker
|
||
from app.core.classifier import Classifier
|
||
from app.core.ingest_tasks import IngestTaskManager, IngestTaskStatus
|
||
from app.core.ingestion import Ingester
|
||
from app.core.query_parser import QueryParser
|
||
from app.core.retriever import Retriever
|
||
from app.core.sparse import SparseEncoder
|
||
from app.core.summarizer import Summarizer
|
||
from app.main import app
|
||
from app.models.knowledge import load_taxonomy
|
||
from app.services.qdrant import COLLECTION_CHUNKS, QdrantService
|
||
from tests.test_e2e_integration import (
|
||
FAKE_L1_SUMMARY,
|
||
DeterministicEmbedding,
|
||
FakeCache,
|
||
FakeOllama,
|
||
)
|
||
|
||
|
||
async def _make_env() -> tuple[QdrantService, Ingester, Retriever]:
|
||
"""构建集成环境：真实组件 + 内存 Qdrant + FakeOllama + 确定性向量"""
|
||
qdrant = QdrantService(client=AsyncQdrantClient(location=":memory:"))
|
||
await qdrant.ensure_collections()
|
||
ollama = FakeOllama()
|
||
taxonomy = load_taxonomy()
|
||
embedding = DeterministicEmbedding()
|
||
ingester = Ingester(
|
||
summarizer=Summarizer(ollama=ollama), # type: ignore[arg-type]
|
||
classifier=Classifier(ollama=ollama, taxonomy=taxonomy), # type: ignore[arg-type]
|
||
chunker=Chunker(),
|
||
embedding=embedding, # type: ignore[arg-type]
|
||
sparse=SparseEncoder(),
|
||
qdrant=qdrant,
|
||
)
|
||
retriever = Retriever(
|
||
qdrant=qdrant,
|
||
query_parser=QueryParser(ollama=ollama, taxonomy=taxonomy, cache=FakeCache()), # type: ignore[arg-type]
|
||
embedding=embedding, # type: ignore[arg-type]
|
||
sparse_encoder=SparseEncoder(),
|
||
)
|
||
return qdrant, ingester, retriever
|
||
|
||
|
||
def _patch_app(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
manager: IngestTaskManager,
|
||
qdrant: QdrantService,
|
||
retriever: Retriever,
|
||
upload_dir: Path,
|
||
) -> None:
|
||
"""替换模块级单例：任务管理器 / Qdrant / Retriever / upload_dir，lifespan 建集合改空操作"""
|
||
|
||
async def _noop_ensure_collections(self: QdrantService) -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(QdrantService, "ensure_collections", _noop_ensure_collections)
|
||
monkeypatch.setattr(document_module, "_task_manager", manager)
|
||
monkeypatch.setattr(document_module, "_qdrant", qdrant)
|
||
monkeypatch.setattr(document_module.settings, "upload_dir", str(upload_dir))
|
||
|
||
|
||
def _md_content(name: str) -> bytes:
|
||
"""构造可触发完整三级总结的 .md 文件内容（不同 name 保证文本不同，避免去重）"""
|
||
paragraph = f"这是 {name} 章节的正文内容，包含足够信息量用于测试切分与向量化流程。" * 20
|
||
text = f"# {name} 安装指南\n{paragraph}\n\n## 环境准备\n{paragraph}\n\n## 安装步骤\n{paragraph}"
|
||
return text.encode("utf-8")
|
||
|
||
|
||
class TestBatchUploadIntegration:
|
||
"""批量上传闭环：3 文件全成功 → 任务列表 → 文档详情含原始文件信息"""
|
||
|
||
async def test_batch_upload_full_loop(
|
||
self,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
admin_headers: dict[str, str],
|
||
) -> None:
|
||
qdrant, ingester, retriever = await _make_env()
|
||
manager = IngestTaskManager(ingester, None, Settings())
|
||
_patch_app(monkeypatch, manager, qdrant, retriever, tmp_path / "uploads")
|
||
|
||
with TestClient(app) as client:
|
||
# 1. 批量上传 3 个 .md：202 + tasks 含 3 个 + failed 空
|
||
resp = client.post(
|
||
"/api/v1/documents/upload-batch",
|
||
files=[
|
||
("files", ("a.md", _md_content("Alpha"), "text/markdown")),
|
||
("files", ("b.md", _md_content("Beta"), "text/markdown")),
|
||
("files", ("c.md", _md_content("Gamma"), "text/markdown")),
|
||
],
|
||
headers=admin_headers,
|
||
)
|
||
assert resp.status_code == 202
|
||
data = resp.json()["data"]
|
||
assert len(data["tasks"]) == 3
|
||
assert data["failed"] == []
|
||
task_ids = [t["task_id"] for t in data["tasks"]]
|
||
filenames = [t["filename"] for t in data["tasks"]]
|
||
assert filenames == ["a.md", "b.md", "c.md"]
|
||
|
||
# 2. 逐个等待终态：全部 done
|
||
doc_ids: list[str] = []
|
||
for tid in task_ids:
|
||
final = await manager.wait_done(tid)
|
||
assert final["status"] == IngestTaskStatus.DONE
|
||
doc_ids.append(final["result"]["document_id"])
|
||
|
||
# 3. GET /documents/tasks：items 含 3 项，按 updated_at 降序，每项含 filename/doc_id
|
||
resp_list = client.get(
|
||
"/api/v1/documents/tasks?limit=10", headers=admin_headers
|
||
)
|
||
body_list = resp_list.json()
|
||
assert body_list["code"] == 0
|
||
items = body_list["data"]["items"]
|
||
assert body_list["data"]["total"] == 3
|
||
assert len(items) == 3
|
||
# 按 updated_at 降序
|
||
updated = [it["updated_at"] for it in items]
|
||
assert updated == sorted(updated, reverse=True)
|
||
# 每项含 filename 与 doc_id
|
||
for it in items:
|
||
assert it["filename"] in {"a.md", "b.md", "c.md"}
|
||
assert it["doc_id"] in doc_ids
|
||
|
||
# 4. GET /documents/{doc_id}：file 字段非 null（有原始文件信息）
|
||
resp_detail = client.get(f"/api/v1/documents/{doc_ids[0]}")
|
||
body_detail = resp_detail.json()
|
||
assert body_detail["code"] == 0
|
||
file_info = body_detail["data"]["file"]
|
||
assert file_info is not None
|
||
assert file_info["filename"] == filenames[0]
|
||
assert file_info["size_bytes"] > 0
|
||
assert file_info["url"] == f"/api/v1/documents/{doc_ids[0]}/file"
|
||
|
||
|
||
class TestBatchUploadPartialFailure:
|
||
"""部分失败：1 个 .unsupported 扩展名进 failed，其余 2 个成功"""
|
||
|
||
async def test_batch_upload_partial_failure(
|
||
self,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
admin_headers: dict[str, str],
|
||
) -> None:
|
||
qdrant, ingester, retriever = await _make_env()
|
||
manager = IngestTaskManager(ingester, None, Settings())
|
||
_patch_app(monkeypatch, manager, qdrant, retriever, tmp_path / "uploads")
|
||
|
||
with TestClient(app) as client:
|
||
resp = client.post(
|
||
"/api/v1/documents/upload-batch",
|
||
files=[
|
||
("files", ("ok1.md", _md_content("Ok1"), "text/markdown")),
|
||
("files", ("ok2.md", _md_content("Ok2"), "text/markdown")),
|
||
("files", ("bad.unsupported", b"whatever", "application/octet-stream")),
|
||
],
|
||
headers=admin_headers,
|
||
)
|
||
assert resp.status_code == 202
|
||
data = resp.json()["data"]
|
||
assert len(data["tasks"]) == 2
|
||
assert len(data["failed"]) == 1
|
||
assert data["failed"][0]["filename"] == "bad.unsupported"
|
||
assert "不支持" in data["failed"][0]["error"]
|
||
# 2 个成功任务均可进入终态 done
|
||
for t in data["tasks"]:
|
||
final = await manager.wait_done(t["task_id"])
|
||
assert final["status"] == IngestTaskStatus.DONE
|
||
|
||
|
||
class TestReingestIntegration:
|
||
"""重新入库：成功路径 + 边界（无原始文件 / 文档不存在 / 文件已删）"""
|
||
|
||
async def test_reingest_success(
|
||
self,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
admin_headers: dict[str, str],
|
||
) -> None:
|
||
qdrant, ingester, retriever = await _make_env()
|
||
manager = IngestTaskManager(ingester, None, Settings())
|
||
_patch_app(monkeypatch, manager, qdrant, retriever, tmp_path / "uploads")
|
||
|
||
with TestClient(app) as client:
|
||
# 先上传 1 个文件，拿到 doc_id
|
||
resp_up = client.post(
|
||
"/api/v1/documents/upload-batch",
|
||
files=[("files", ("re.md", _md_content("Reingest"), "text/markdown"))],
|
||
headers=admin_headers,
|
||
)
|
||
assert resp_up.status_code == 202
|
||
up_data = resp_up.json()["data"]
|
||
assert len(up_data["tasks"]) == 1
|
||
old_task_id = up_data["tasks"][0]["task_id"]
|
||
old_final = await manager.wait_done(old_task_id)
|
||
old_doc_id = old_final["result"]["document_id"]
|
||
|
||
# 重新入库：202 + task_id
|
||
resp_re = client.post(
|
||
f"/api/v1/documents/{old_doc_id}/reingest", headers=admin_headers
|
||
)
|
||
assert resp_re.status_code == 202
|
||
re_body = resp_re.json()
|
||
assert re_body["code"] == 0
|
||
new_task_id = re_body["data"]["task_id"]
|
||
assert new_task_id
|
||
|
||
# 等待终态：done
|
||
new_final = await manager.wait_done(new_task_id)
|
||
assert new_final["status"] == IngestTaskStatus.DONE
|
||
new_doc_id = new_final["result"]["document_id"]
|
||
|
||
# 旧 doc_id 数据已删：GET /documents/{old_doc_id} → 1004
|
||
resp_old = client.get(f"/api/v1/documents/{old_doc_id}")
|
||
assert resp_old.json()["code"] == 1004
|
||
|
||
# 新 doc_id 详情：L1 summary 存在，file 非 null（保留原文件信息）
|
||
resp_new = client.get(f"/api/v1/documents/{new_doc_id}")
|
||
body_new = resp_new.json()
|
||
assert body_new["code"] == 0
|
||
assert body_new["data"]["l1"]["text"] == FAKE_L1_SUMMARY
|
||
assert body_new["data"]["file"] is not None
|
||
assert body_new["data"]["file"]["filename"] == "re.md"
|
||
|
||
# 旧 chunks 已删：chunks 集合中 old_doc_id 点数为 0
|
||
old_count = await qdrant.client.count(
|
||
collection_name=COLLECTION_CHUNKS,
|
||
count_filter=qdrant.build_filter(doc_ids=[old_doc_id]),
|
||
exact=True,
|
||
)
|
||
assert old_count.count == 0
|
||
|
||
async def test_reingest_text_only_doc_returns_1001(
|
||
self,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
admin_headers: dict[str, str],
|
||
) -> None:
|
||
"""纯文本入库的 doc_id → reingest → 1001（无原始文件）"""
|
||
qdrant, ingester, retriever = await _make_env()
|
||
manager = IngestTaskManager(ingester, None, Settings())
|
||
_patch_app(monkeypatch, manager, qdrant, retriever, tmp_path / "uploads")
|
||
|
||
with TestClient(app) as client:
|
||
# 纯文本入库（无原始文件落盘记录）
|
||
resp_doc = client.post(
|
||
"/api/v1/documents",
|
||
json={
|
||
"text": "# 纯文本\n这是一段纯文本入库的内容，用于测试 reingest 边界。",
|
||
"title": "纯文本",
|
||
},
|
||
headers=admin_headers,
|
||
)
|
||
assert resp_doc.status_code == 202
|
||
task_id = resp_doc.json()["data"]["task_id"]
|
||
final = await manager.wait_done(task_id)
|
||
doc_id = final["result"]["document_id"]
|
||
|
||
# reingest → 1001
|
||
resp_re = client.post(
|
||
f"/api/v1/documents/{doc_id}/reingest", headers=admin_headers
|
||
)
|
||
body = resp_re.json()
|
||
assert body["code"] == 1001
|
||
assert "无原始文件" in body["message"]
|
||
|
||
async def test_reingest_doc_not_found_returns_1004(
|
||
self,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
admin_headers: dict[str, str],
|
||
) -> None:
|
||
"""不存在的 doc_id → reingest → 1004"""
|
||
qdrant, ingester, retriever = await _make_env()
|
||
manager = IngestTaskManager(ingester, None, Settings())
|
||
_patch_app(monkeypatch, manager, qdrant, retriever, tmp_path / "uploads")
|
||
|
||
with TestClient(app) as client:
|
||
resp = client.post(
|
||
"/api/v1/documents/nonexistent-doc/reingest", headers=admin_headers
|
||
)
|
||
body = resp.json()
|
||
assert body["code"] == 1004
|
||
|
||
async def test_reingest_file_deleted_returns_1004(
|
||
self,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
tmp_path: Path,
|
||
admin_headers: dict[str, str],
|
||
) -> None:
|
||
"""文件被删（删 upload_dir 下落盘文件后）→ reingest → 1004"""
|
||
qdrant, ingester, retriever = await _make_env()
|
||
manager = IngestTaskManager(ingester, None, Settings())
|
||
_patch_app(monkeypatch, manager, qdrant, retriever, tmp_path / "uploads")
|
||
|
||
with TestClient(app) as client:
|
||
# 上传 1 个文件
|
||
resp_up = client.post(
|
||
"/api/v1/documents/upload-batch",
|
||
files=[("files", ("del.md", _md_content("Deleted"), "text/markdown"))],
|
||
headers=admin_headers,
|
||
)
|
||
assert resp_up.status_code == 202
|
||
tid = resp_up.json()["data"]["tasks"][0]["task_id"]
|
||
final = await manager.wait_done(tid)
|
||
doc_id = final["result"]["document_id"]
|
||
|
||
# 删除 upload_dir 下落盘的原始文件
|
||
meta = await qdrant.get_l1_metadata(doc_id)
|
||
assert meta is not None
|
||
raw_path = Path(meta["raw_file_path"])
|
||
assert raw_path.is_file()
|
||
raw_path.unlink()
|
||
|
||
# reingest → 1004（文件不存在）
|
||
resp_re = client.post(
|
||
f"/api/v1/documents/{doc_id}/reingest", headers=admin_headers
|
||
)
|
||
body = resp_re.json()
|
||
assert body["code"] == 1004
|