97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.main import app
|
|
from app.document_service import DocumentNotFoundError
|
|
from app.dependencies import get_document_service, get_search_client
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def test_health_returns_ok():
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
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
|