kplam
f92eff6f65
1. add `/api/v1/documents/{doc_id}/file` endpoint for downloading original document files
2. add file field to document detail API response based on metadata
3. add comprehensive tests for file download API and metadata integration
4. update API documentation in CLAUDE.md
334 lines
14 KiB
Python
334 lines
14 KiB
Python
"""管理闭环集成测试（Spec Task 7：集成验证）
|
||
|
||
两部分覆盖，均不修改 app/ 与 scripts/ 实现代码：
|
||
|
||
1. TestAdminClosedLoop：真实内存 Qdrant（location=":memory:"）+ TestClient 走完整管理闭环
|
||
列表 → 详情 → stats → 删除 → 删除后校验 → 幂等再删。
|
||
document.py 与 knowledge.py 的 _get_qdrant 单例均 monkeypatch 指向同一内存服务，
|
||
数据直接经 QdrantService.upsert_* 写入（L1/L2/L3/chunks 四层齐全）。
|
||
|
||
2. TestAdminPageContract：读取 app/static/admin.html 文本做静态断言，
|
||
校验页面 fetch 路径全部落在后端真实路由集合内、页面引用的响应字段名抽样
|
||
存在于后端模型/路由代码中，防止前后端契约漂移。
|
||
"""
|
||
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from fastapi.routing import APIRoute
|
||
from fastapi.testclient import TestClient
|
||
from qdrant_client import AsyncQdrantClient
|
||
|
||
from app.api.v1 import document as document_module
|
||
from app.api.v1 import knowledge as knowledge_module
|
||
from app.config import settings
|
||
from app.main import app
|
||
from app.services.qdrant import (
|
||
COLLECTION_CHUNKS,
|
||
COLLECTION_L1,
|
||
COLLECTION_L2,
|
||
COLLECTION_L3,
|
||
QdrantService,
|
||
)
|
||
|
||
DIM = settings.embedding_dimension
|
||
|
||
# 文档 A：类目「技术文档」，L2 x2 / L3 x3 / chunks x4
|
||
DOC_A = "doc-a-tech"
|
||
DOC_A_TITLE = "安装指南"
|
||
DOC_A_SUMMARY = "安装指南的一句话总结"
|
||
DOC_A_CATEGORY = "技术文档"
|
||
DOC_A_TAGS = ["安装", "运维"]
|
||
DOC_A_L2_PATHS = ["安装指南", "安装指南 / 环境准备"]
|
||
DOC_A_L3_PATHS = ["安装指南", "安装指南 / 环境准备", "安装指南 / 安装步骤"]
|
||
DOC_A_CHUNKS = 4
|
||
|
||
# 文档 B：类目「uncategorized」，L2 x1 / L3 x2 / chunks x3
|
||
DOC_B = "doc-b-misc"
|
||
DOC_B_TITLE = "随手记"
|
||
DOC_B_SUMMARY = "无法归类的随手记录"
|
||
DOC_B_CATEGORY = "uncategorized"
|
||
DOC_B_L2_PATHS = ["随手记"]
|
||
DOC_B_L3_PATHS = ["随手记", "随手记 / 杂项"]
|
||
DOC_B_CHUNKS = 3
|
||
|
||
# admin.html 与项目根路径
|
||
_PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
_ADMIN_HTML_PATH = _PROJECT_ROOT / "app" / "static" / "admin.html"
|
||
|
||
# 页面引用的响应字段抽样 -> 应包含该字段的后端模型/路由文件（相对项目根）
|
||
_CONTRACT_FIELD_FILES = {
|
||
"routed_categories": "app/models/search.py",
|
||
"fallback": "app/models/search.py",
|
||
"chunks_count": "app/models/document.py",
|
||
"deleted_total": "app/api/v1/document.py",
|
||
"next_offset": "app/api/v1/document.py",
|
||
"uncategorized_count": "app/api/v1/knowledge.py",
|
||
"documents_total": "app/api/v1/knowledge.py",
|
||
}
|
||
|
||
|
||
def _backend_paths() -> set[str]:
|
||
"""展平 app.routes，收集后端真实路由路径
|
||
|
||
兼容两种挂载形态：APIRoute 直接平铺，以及 include_router 产生的
|
||
嵌套代理对象（其路由在 original_router.routes 上）。
|
||
"""
|
||
paths: set[str] = set()
|
||
stack: list[Any] = list(app.routes)
|
||
while stack:
|
||
route = stack.pop()
|
||
if isinstance(route, APIRoute):
|
||
paths.add(route.path)
|
||
continue
|
||
stack.extend(getattr(route, "routes", []) or [])
|
||
inner = getattr(route, "original_router", None)
|
||
if inner is not None:
|
||
stack.extend(getattr(inner, "routes", []))
|
||
return paths
|
||
|
||
|
||
def _dense(seed: float) -> list[float]:
|
||
"""构造确定性 dense 向量：前 4 维取特征值，便于区分不同点"""
|
||
vec = [0.0] * DIM
|
||
vec[0] = seed
|
||
vec[1] = 1.0 - seed
|
||
vec[2] = seed * 0.5
|
||
vec[3] = 0.1
|
||
return vec
|
||
|
||
|
||
def _node(doc_id: str, section_path: str, category: str, tags: list[str], seed: float) -> dict[str, Any]:
|
||
"""构造一个 L2/L3 大纲节点"""
|
||
return {
|
||
"doc_id": doc_id,
|
||
"section_path": section_path,
|
||
"text": f"{section_path} 的节点内容",
|
||
"category": category,
|
||
"tags": tags,
|
||
"dense_vector": _dense(seed),
|
||
}
|
||
|
||
|
||
def _chunk(doc_id: str, index: int, title: str, category: str, tags: list[str]) -> dict[str, Any]:
|
||
"""构造一个原文 chunk"""
|
||
return {
|
||
"doc_id": doc_id,
|
||
"chunk_index": index,
|
||
"text": f"{doc_id} 的第 {index} 个 chunk",
|
||
"section_path": f"section-{index}",
|
||
"title": title,
|
||
"category": category,
|
||
"tags": tags,
|
||
"dense_vector": _dense(0.5 + index * 0.05),
|
||
}
|
||
|
||
|
||
async def _seed_documents(service: QdrantService) -> None:
|
||
"""写入两篇四层结构完整的文档：A「技术文档」、B「uncategorized」"""
|
||
await service.upsert_l1(
|
||
doc_id=DOC_A,
|
||
title=DOC_A_TITLE,
|
||
summary=DOC_A_SUMMARY,
|
||
category=DOC_A_CATEGORY,
|
||
tags=DOC_A_TAGS,
|
||
dense_vector=_dense(0.9),
|
||
)
|
||
await service.upsert_nodes(
|
||
COLLECTION_L2, [_node(DOC_A, path, DOC_A_CATEGORY, DOC_A_TAGS, 0.8) for path in DOC_A_L2_PATHS]
|
||
)
|
||
await service.upsert_nodes(
|
||
COLLECTION_L3, [_node(DOC_A, path, DOC_A_CATEGORY, DOC_A_TAGS, 0.7) for path in DOC_A_L3_PATHS]
|
||
)
|
||
await service.upsert_chunks(
|
||
[_chunk(DOC_A, i, DOC_A_TITLE, DOC_A_CATEGORY, DOC_A_TAGS) for i in range(DOC_A_CHUNKS)]
|
||
)
|
||
|
||
await service.upsert_l1(
|
||
doc_id=DOC_B,
|
||
title=DOC_B_TITLE,
|
||
summary=DOC_B_SUMMARY,
|
||
category=DOC_B_CATEGORY,
|
||
tags=[],
|
||
dense_vector=_dense(0.6),
|
||
)
|
||
await service.upsert_nodes(
|
||
COLLECTION_L2, [_node(DOC_B, path, DOC_B_CATEGORY, [], 0.55) for path in DOC_B_L2_PATHS]
|
||
)
|
||
await service.upsert_nodes(
|
||
COLLECTION_L3, [_node(DOC_B, path, DOC_B_CATEGORY, [], 0.5) for path in DOC_B_L3_PATHS]
|
||
)
|
||
await service.upsert_chunks([_chunk(DOC_B, i, DOC_B_TITLE, DOC_B_CATEGORY, []) for i in range(DOC_B_CHUNKS)])
|
||
|
||
|
||
class TestAdminClosedLoop:
|
||
"""文档管理 API + stats 的完整闭环（真实内存 Qdrant + TestClient）"""
|
||
|
||
async def test_document_admin_closed_loop(
|
||
self, monkeypatch: pytest.MonkeyPatch, admin_headers: dict[str, str]
|
||
) -> None:
|
||
service = QdrantService(client=AsyncQdrantClient(location=":memory:"))
|
||
await service.ensure_collections()
|
||
await _seed_documents(service)
|
||
|
||
# lifespan 的集合初始化替换为空操作；两个路由模块的 _get_qdrant 指向同一内存服务
|
||
async def _noop_ensure_collections(self: QdrantService) -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(QdrantService, "ensure_collections", _noop_ensure_collections)
|
||
monkeypatch.setattr(document_module, "_get_qdrant", lambda: service)
|
||
monkeypatch.setattr(knowledge_module, "_get_qdrant", lambda: service)
|
||
|
||
with TestClient(app) as client:
|
||
# 1. 列表：items 数=2，字段完整
|
||
resp = client.get("/api/v1/documents")
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["code"] == 0
|
||
data = body["data"]
|
||
assert set(data.keys()) == {"items", "next_offset"}
|
||
items = data["items"]
|
||
assert len(items) == 2
|
||
assert data["next_offset"] is None
|
||
by_id = {item["doc_id"]: item for item in items}
|
||
assert set(by_id) == {DOC_A, DOC_B}
|
||
item_a = by_id[DOC_A]
|
||
assert set(item_a.keys()) == {"doc_id", "title", "category", "tags", "summary"}
|
||
assert item_a["title"] == DOC_A_TITLE
|
||
assert item_a["category"] == DOC_A_CATEGORY
|
||
assert item_a["tags"] == DOC_A_TAGS
|
||
assert item_a["summary"] == DOC_A_SUMMARY
|
||
item_b = by_id[DOC_B]
|
||
assert item_b["category"] == DOC_B_CATEGORY
|
||
assert item_b["tags"] == []
|
||
|
||
# 2. 详情：l1/l2_nodes/l3_nodes/chunks_count/file 与写入一致
|
||
resp = client.get(f"/api/v1/documents/{DOC_A}")
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert set(data.keys()) == {"l1", "l2_nodes", "l3_nodes", "chunks_count", "file"}
|
||
# 文本入库无 metadata → file=None
|
||
assert data["file"] is None
|
||
l1 = data["l1"]
|
||
assert l1["doc_id"] == DOC_A
|
||
assert l1["title"] == DOC_A_TITLE
|
||
assert l1["text"] == DOC_A_SUMMARY
|
||
assert l1["category"] == DOC_A_CATEGORY
|
||
assert l1["tags"] == DOC_A_TAGS
|
||
assert len(data["l2_nodes"]) == len(DOC_A_L2_PATHS)
|
||
assert {n["section_path"] for n in data["l2_nodes"]} == set(DOC_A_L2_PATHS)
|
||
assert len(data["l3_nodes"]) == len(DOC_A_L3_PATHS)
|
||
assert {n["section_path"] for n in data["l3_nodes"]} == set(DOC_A_L3_PATHS)
|
||
assert data["chunks_count"] == DOC_A_CHUNKS
|
||
|
||
# 3. stats：documents_total=2、类目分布与四层点数正确
|
||
resp = client.get("/api/v1/knowledge/stats")
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert set(data.keys()) == {"collections", "categories", "uncategorized_count", "documents_total"}
|
||
assert data["documents_total"] == 2
|
||
assert data["categories"] == {DOC_A_CATEGORY: 1, DOC_B_CATEGORY: 1}
|
||
assert data["uncategorized_count"] == 1
|
||
assert data["collections"] == {
|
||
COLLECTION_L1: 2,
|
||
COLLECTION_L2: len(DOC_A_L2_PATHS) + len(DOC_B_L2_PATHS),
|
||
COLLECTION_L3: len(DOC_A_L3_PATHS) + len(DOC_B_L3_PATHS),
|
||
COLLECTION_CHUNKS: DOC_A_CHUNKS + DOC_B_CHUNKS,
|
||
}
|
||
|
||
# 4. 删除文档 A：deleted_total > 0，各集合删除数与写入一致
|
||
resp = client.delete(f"/api/v1/documents/{DOC_A}", headers=admin_headers)
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["code"] == 0
|
||
data = body["data"]
|
||
assert set(data.keys()) == {"doc_id", "deleted", "deleted_total"}
|
||
assert data["doc_id"] == DOC_A
|
||
assert data["deleted"] == {
|
||
COLLECTION_L1: 1,
|
||
COLLECTION_L2: len(DOC_A_L2_PATHS),
|
||
COLLECTION_L3: len(DOC_A_L3_PATHS),
|
||
COLLECTION_CHUNKS: DOC_A_CHUNKS,
|
||
}
|
||
assert data["deleted_total"] == 1 + len(DOC_A_L2_PATHS) + len(DOC_A_L3_PATHS) + DOC_A_CHUNKS > 0
|
||
|
||
# 5. 删除后校验：详情 1004、stats 只剩 1 篇、文档 B 不受影响
|
||
resp = client.get(f"/api/v1/documents/{DOC_A}")
|
||
body = resp.json()
|
||
assert body["code"] == 1004
|
||
assert body["data"] is None
|
||
|
||
resp = client.get("/api/v1/knowledge/stats")
|
||
data = resp.json()["data"]
|
||
assert data["documents_total"] == 1
|
||
assert data["categories"] == {DOC_B_CATEGORY: 1}
|
||
assert data["uncategorized_count"] == 1
|
||
assert data["collections"] == {
|
||
COLLECTION_L1: 1,
|
||
COLLECTION_L2: len(DOC_B_L2_PATHS),
|
||
COLLECTION_L3: len(DOC_B_L3_PATHS),
|
||
COLLECTION_CHUNKS: DOC_B_CHUNKS,
|
||
}
|
||
|
||
resp = client.get(f"/api/v1/documents/{DOC_B}")
|
||
body = resp.json()
|
||
assert body["code"] == 0
|
||
data = body["data"]
|
||
assert data["l1"]["doc_id"] == DOC_B
|
||
assert len(data["l2_nodes"]) == len(DOC_B_L2_PATHS)
|
||
assert len(data["l3_nodes"]) == len(DOC_B_L3_PATHS)
|
||
assert data["chunks_count"] == DOC_B_CHUNKS
|
||
|
||
resp = client.get("/api/v1/documents")
|
||
items = resp.json()["data"]["items"]
|
||
assert [item["doc_id"] for item in items] == [DOC_B]
|
||
|
||
# 6. 幂等再删：code=0 且 deleted_total=0，各集合删除数全 0
|
||
resp = client.delete(f"/api/v1/documents/{DOC_A}", headers=admin_headers)
|
||
body = resp.json()
|
||
assert body["code"] == 0
|
||
data = body["data"]
|
||
all_collections = (COLLECTION_L1, COLLECTION_L2, COLLECTION_L3, COLLECTION_CHUNKS)
|
||
assert data["deleted"] == {name: 0 for name in all_collections}
|
||
assert data["deleted_total"] == 0
|
||
|
||
|
||
class TestAdminPageContract:
|
||
"""admin.html 与后端 API 的契约一致性（静态断言，防契约漂移）"""
|
||
|
||
def test_fetch_paths_in_backend_routes(self) -> None:
|
||
"""HTML 中所有 /api/v1/ URL 字面量都在后端真实路由集合内"""
|
||
html = _ADMIN_HTML_PATH.read_text(encoding="utf-8")
|
||
# 提取所有以 /api/v1/ 开头的字符串字面量（覆盖 fetch()/api() 调用与路径变量）
|
||
literals = set(re.findall(r"""["'](/api/v1/[^"']*)["']""", html))
|
||
|
||
# 页面至少应覆盖这 5 条契约路径（缺失说明提取失效或页面能力回退）
|
||
expected = {
|
||
"/api/v1/search",
|
||
"/api/v1/documents",
|
||
"/api/v1/documents/", # 详情/删除：拼接 doc_id 的前缀形式
|
||
"/api/v1/knowledge/stats",
|
||
"/api/v1/knowledge/categories",
|
||
}
|
||
assert expected <= literals
|
||
|
||
backend_paths = _backend_paths()
|
||
for literal in sorted(literals):
|
||
path = literal.split("?", 1)[0] # 去掉查询串
|
||
if path in backend_paths:
|
||
continue
|
||
# 以 / 结尾的前缀（如 /api/v1/documents/）应对应带路径参数的路由
|
||
if path.endswith("/") and any(p.startswith(path) for p in backend_paths):
|
||
continue
|
||
pytest.fail(f"页面 fetch 路径不在后端路由集合内: {literal}")
|
||
|
||
@pytest.mark.parametrize(("field", "source_file"), sorted(_CONTRACT_FIELD_FILES.items()))
|
||
def test_response_fields_in_backend_code(self, field: str, source_file: str) -> None:
|
||
"""页面引用的响应字段名：既在 HTML 中出现，也在后端模型/路由代码中存在"""
|
||
html = _ADMIN_HTML_PATH.read_text(encoding="utf-8")
|
||
assert field in html, f"契约字段未在页面中引用: {field}"
|
||
source = (_PROJECT_ROOT / source_file).read_text(encoding="utf-8")
|
||
assert field in source, f"契约字段未在后端代码中定义: {field} ({source_file})"
|