feat: wire documents and search routers into FastAPI app with e2e smoke test

This commit is contained in:
Tianyang 2026-08-01 02:36:58 +08:00
parent d758500548
commit 448a5cbe56
2 changed files with 92 additions and 1 deletions

View File

@ -2,7 +2,7 @@ from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware from fastapi.middleware.cors import CORSMiddleware
from app.config import settings from app.config import settings
from app.routers import documents from app.routers import documents, search
app = FastAPI(title="DocHub API") app = FastAPI(title="DocHub API")
@ -15,6 +15,12 @@ app.add_middleware(
) )
app.include_router(documents.router) app.include_router(documents.router)
app.include_router(search.router)
@app.get("/health")
def health():
return {"status": "ok"}
@app.get("/health") @app.get("/health")

View File

@ -1,6 +1,9 @@
import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from app.main import app from app.main import app
from app.document_service import DocumentNotFoundError
from app.dependencies import get_document_service, get_search_client
client = TestClient(app) client = TestClient(app)
@ -9,3 +12,85 @@ def test_health_returns_ok():
response = client.get("/health") response = client.get("/health")
assert response.status_code == 200 assert response.status_code == 200
assert response.json() == {"status": "ok"} assert response.json() == {"status": "ok"}
class FakeDocumentService:
def __init__(self):
self.storage = {}
def list_documents(self, path):
prefix = path if not path or path.endswith("/") else f"{path}/"
files = [
{"key": key, "name": key[len(prefix):], "size": len(data), "modifiedAt": "2026-08-01T10:00:00Z"}
for key, data in self.storage.items()
if key.startswith(prefix)
]
return {"folders": [], "files": files}
def get_document(self, key):
if key not in self.storage:
raise DocumentNotFoundError(key)
return {"key": key, "content": self.storage[key].decode("utf-8"), "contentType": "text/markdown"}
def save_document(self, key, content, content_type):
self.storage[key] = content
return {"key": key, "updatedAt": "2026-08-01T10:05:00Z"}
def delete_document(self, key):
self.storage.pop(key, None)
def upload_document(self, key, data, content_type):
self.storage[key] = data
return {"key": key, "name": key.rsplit("/", 1)[-1], "size": len(data)}
class FakeSearchClient:
def search(self, query):
return [
{
"key": "产品文档/架构设计.md",
"title": "架构设计.md",
"path": "产品文档",
"snippet": "全文<em>检索</em>示例",
}
]
@pytest.fixture
def e2e_client():
fake_service = FakeDocumentService()
fake_search = FakeSearchClient()
app.dependency_overrides[get_document_service] = lambda: fake_service
app.dependency_overrides[get_search_client] = lambda: fake_search
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()
def test_end_to_end_upload_get_list_search_delete_flow(e2e_client):
upload_response = e2e_client.post(
"/api/upload",
data={"path": "产品文档"},
files={"file": ("架构设计.md", "# 架构设计\n\nDocHub 支持全文检索。".encode("utf-8"), "text/markdown")},
)
assert upload_response.status_code == 201
key = upload_response.json()["key"]
assert key == "产品文档/架构设计.md"
get_response = e2e_client.get(f"/api/documents/{key}")
assert get_response.status_code == 200
assert "架构设计" in get_response.json()["content"]
list_response = e2e_client.get("/api/documents", params={"path": "产品文档"})
assert list_response.status_code == 200
assert any(f["key"] == key for f in list_response.json()["files"])
search_response = e2e_client.get("/api/search", params={"q": "检索"})
assert search_response.status_code == 200
assert search_response.json()["results"][0]["key"] == "产品文档/架构设计.md"
delete_response = e2e_client.delete(f"/api/documents/{key}")
assert delete_response.status_code == 204
missing_response = e2e_client.get(f"/api/documents/{key}")
assert missing_response.status_code == 404