- vstore 独立安装包(8094):解析/嵌入/加密 vault/faiss 检索,存储路径支持本地 file 与 http(s):// URL,启动握手校验,集群化预留(embed 独立模块) - BFF docgen 路由:上传入库/generate/编辑/导出/下载/主库管理,意图护栏(与前端同词典),vstore 失败兜底,quota 507 与 TTL 惰性清理 - 前端:自然触发意图路由、拖拽上传 chips、能力徽标、右侧文档 TAB(WYSIWYG 编辑/prompt 修改/PDF+DOCX 导出下载)、右栏拖宽+手柄仿左栏 - 批量导入工具 docgen_import.py(并发 8、导入即加密+删源明文) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
111 lines
3.3 KiB
Python
111 lines
3.3 KiB
Python
"""test_api.py —— FastAPI 端点(TestClient + 假 embed,不碰真 ollama)。"""
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
from starlette.testclient import TestClient
|
||
|
||
from vstore import vault as vault_mod
|
||
from vstore.api import app
|
||
from vstore.settings import VstorePaths, set_paths
|
||
|
||
|
||
@pytest.fixture
|
||
def client(tmp_path, embed_fn):
|
||
set_paths(VstorePaths(root=str(tmp_path)))
|
||
with TestClient(app) as c:
|
||
yield c
|
||
|
||
|
||
def test_health(client):
|
||
resp = client.get("/api/vstore/health")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["ok"] is True
|
||
assert data["count"] == 0
|
||
|
||
|
||
def test_ingest_then_search(client):
|
||
resp = client.post(
|
||
"/api/vstore/ingest",
|
||
files={"file": ("report.txt", "小麦育种年度报告内容。".encode("utf-8"), "text/plain")},
|
||
data={"name": "年度报告", "category": "农业", "description": "小麦"},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["ok"] is True
|
||
item = data["item"]
|
||
assert item["document_id"]
|
||
assert item["name"] == "年度报告"
|
||
|
||
resp2 = client.get("/api/vstore/search", params={"q": "小麦"})
|
||
assert resp2.status_code == 200
|
||
data2 = resp2.json()
|
||
assert data2["ok"] is True
|
||
assert len(data2["items"]) >= 1
|
||
assert data2["items"][0]["id"] == item["document_id"]
|
||
|
||
|
||
def test_ingest_empty_file_rejected(client):
|
||
resp = client.post(
|
||
"/api/vstore/ingest",
|
||
files={"file": ("a.txt", b"", "text/plain")},
|
||
data={"name": "x"},
|
||
)
|
||
assert resp.status_code == 422
|
||
data = resp.json()
|
||
assert data["ok"] is False
|
||
assert data["error"]["code"] == "VALIDATION"
|
||
|
||
|
||
def test_document_roundtrip(client):
|
||
doc_id = client.post(
|
||
"/api/vstore/ingest",
|
||
files={"file": ("d.txt", "文档正文内容。".encode("utf-8"))},
|
||
data={"name": "d"},
|
||
).json()["item"]["document_id"]
|
||
|
||
resp = client.get(f"/api/vstore/documents/{doc_id}")
|
||
assert resp.status_code == 200
|
||
doc = resp.json()["document"]
|
||
assert "文档正文内容" in doc["paragraphs"][0]
|
||
|
||
# style(非 docx → 空 hints)
|
||
resp_style = client.get(f"/api/vstore/documents/{doc_id}/style")
|
||
assert resp_style.status_code == 200
|
||
assert resp_style.json()["style_hints"] == {}
|
||
|
||
|
||
def test_document_404(client):
|
||
assert client.get("/api/vstore/documents/nope").status_code == 404
|
||
assert client.get("/api/vstore/documents/nope/style").status_code == 404
|
||
|
||
|
||
def test_delete_document(client):
|
||
doc_id = client.post(
|
||
"/api/vstore/ingest",
|
||
files={"file": ("d.txt", b"content")},
|
||
data={"name": "d"},
|
||
).json()["item"]["document_id"]
|
||
resp = client.delete(f"/api/vstore/documents/{doc_id}")
|
||
assert resp.json() == {"ok": True, "removed": True}
|
||
assert client.get(f"/api/vstore/documents/{doc_id}").status_code == 404
|
||
|
||
|
||
def test_rebuild_index(client):
|
||
client.post(
|
||
"/api/vstore/ingest",
|
||
files={"file": ("a.txt", b"aaa")},
|
||
data={"name": "a"},
|
||
)
|
||
resp = client.post("/api/vstore/rebuild-index")
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["ok"] is True
|
||
assert data["count"] == 1
|
||
|
||
|
||
def test_search_params_bounded(client):
|
||
resp = client.get("/api/vstore/search", params={"q": "", "top_k": 999})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["ok"] is True
|