Tianyang
9ef22f9f0a
- Default CORS_ORIGINS to http://localhost:5173 instead of "*" (still overridable) - Reject ".." and leading "/" in document keys/paths across get/save/delete/upload, mapped to HTTP 400 via new InvalidPathError - Add a delete button to DocumentEditor wired to deleteDocument + onDeleted, giving the existing DELETE API an actual UI entry point - Share a single lru_cache'd Meilisearch (and Minio) SDK client instance instead of constructing duplicates in dependencies.py - Add a startup check that creates the MinIO bucket if missing, tolerating MinIO being unreachable at boot without crashing the app
134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import app.main as main_module
|
|
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"}
|
|
|
|
|
|
def test_startup_creates_bucket_when_missing(monkeypatch):
|
|
fake_minio = MagicMock()
|
|
fake_minio.bucket_exists.return_value = False
|
|
monkeypatch.setattr(main_module, "_get_minio_sdk", lambda: fake_minio)
|
|
|
|
with TestClient(app):
|
|
pass
|
|
|
|
fake_minio.bucket_exists.assert_called()
|
|
fake_minio.make_bucket.assert_called()
|
|
|
|
|
|
def test_startup_does_not_create_bucket_when_it_already_exists(monkeypatch):
|
|
fake_minio = MagicMock()
|
|
fake_minio.bucket_exists.return_value = True
|
|
monkeypatch.setattr(main_module, "_get_minio_sdk", lambda: fake_minio)
|
|
|
|
with TestClient(app):
|
|
pass
|
|
|
|
fake_minio.make_bucket.assert_not_called()
|
|
|
|
|
|
def test_startup_does_not_crash_app_when_minio_is_unreachable(monkeypatch):
|
|
def raise_connection_error():
|
|
raise ConnectionError("minio unreachable")
|
|
|
|
monkeypatch.setattr(main_module, "_get_minio_sdk", raise_connection_error)
|
|
|
|
with TestClient(app) as test_client:
|
|
response = test_client.get("/health")
|
|
assert response.status_code == 200
|
|
|
|
|
|
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
|