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
243 lines
8.6 KiB
Python
243 lines
8.6 KiB
Python
import pytest
|
|
|
|
from app.document_service import (
|
|
DocumentService,
|
|
DocumentNotFoundError,
|
|
FileTooLargeError,
|
|
InvalidPathError,
|
|
MAX_UPLOAD_SIZE_BYTES,
|
|
UnsupportedFileTypeError,
|
|
)
|
|
|
|
|
|
class FakeMinioClient:
|
|
def __init__(self):
|
|
self.entries = []
|
|
self.stored = {}
|
|
self.deleted_keys = []
|
|
self.put_calls = []
|
|
|
|
def list_objects(self, prefix):
|
|
return [e for e in self.entries if e["key"].startswith(prefix)]
|
|
|
|
def get_object(self, key):
|
|
if key not in self.stored:
|
|
raise FileNotFoundError(key)
|
|
return self.stored[key]
|
|
|
|
def put_object(self, key, data, content_type):
|
|
self.stored[key] = data
|
|
self.put_calls.append((key, content_type))
|
|
|
|
def delete_object(self, key):
|
|
if key not in self.stored:
|
|
raise FileNotFoundError(key)
|
|
del self.stored[key]
|
|
self.deleted_keys.append(key)
|
|
|
|
|
|
class FakeSearchClient:
|
|
def __init__(self, fail_on_index=False, fail_on_delete=False):
|
|
self.indexed = []
|
|
self.deleted = []
|
|
self.fail_on_index = fail_on_index
|
|
self.fail_on_delete = fail_on_delete
|
|
|
|
def index_document(self, key, path, title, content):
|
|
if self.fail_on_index:
|
|
raise RuntimeError("meilisearch unavailable")
|
|
self.indexed.append({"key": key, "path": path, "title": title, "content": content})
|
|
|
|
def delete_document(self, key):
|
|
if self.fail_on_delete:
|
|
raise RuntimeError("meilisearch unavailable")
|
|
self.deleted.append(key)
|
|
|
|
|
|
def test_list_documents_at_root_returns_folders_and_files():
|
|
minio_client = FakeMinioClient()
|
|
minio_client.entries = [
|
|
{"key": "产品文档/", "is_dir": True, "size": 0, "modifiedAt": None},
|
|
{"key": "团队规范/", "is_dir": True, "size": 0, "modifiedAt": None},
|
|
{"key": "README.md", "is_dir": False, "size": 128, "modifiedAt": "2026-08-01T09:00:00Z"},
|
|
]
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
|
|
result = service.list_documents("")
|
|
|
|
assert result["folders"] == ["产品文档", "团队规范"]
|
|
assert result["files"] == [
|
|
{"key": "README.md", "name": "README.md", "size": 128, "modifiedAt": "2026-08-01T09:00:00Z"},
|
|
]
|
|
|
|
|
|
def test_get_document_returns_content_and_content_type():
|
|
minio_client = FakeMinioClient()
|
|
minio_client.stored["产品文档/架构设计.md"] = "# 架构设计\n\nDocHub 使用 MinIO 存储文档。".encode("utf-8")
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
|
|
result = service.get_document("产品文档/架构设计.md")
|
|
|
|
assert result["key"] == "产品文档/架构设计.md"
|
|
assert result["contentType"] == "text/markdown"
|
|
assert "架构设计" in result["content"]
|
|
|
|
|
|
def test_get_document_raises_not_found_for_missing_key():
|
|
service = DocumentService(FakeMinioClient(), FakeSearchClient())
|
|
|
|
with pytest.raises(DocumentNotFoundError):
|
|
service.get_document("不存在/文档.md")
|
|
|
|
|
|
def test_get_document_returns_base64_content_for_binary_data_without_raising():
|
|
minio_client = FakeMinioClient()
|
|
binary_data = b"\x89PNG\r\n\x1a\n\x00\x01\xff\xfe"
|
|
minio_client.stored["产品文档/图片.png"] = binary_data
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
|
|
result = service.get_document("产品文档/图片.png")
|
|
|
|
assert result["encoding"] == "base64"
|
|
import base64
|
|
assert base64.b64decode(result["content"]) == binary_data
|
|
|
|
|
|
def test_get_document_returns_utf8_encoding_for_text_content():
|
|
minio_client = FakeMinioClient()
|
|
minio_client.stored["产品文档/架构设计.md"] = "# 架构设计".encode("utf-8")
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
|
|
result = service.get_document("产品文档/架构设计.md")
|
|
|
|
assert result["encoding"] == "utf-8"
|
|
|
|
|
|
def test_save_document_writes_to_minio_then_indexes_content():
|
|
minio_client = FakeMinioClient()
|
|
search_client = FakeSearchClient()
|
|
service = DocumentService(minio_client, search_client)
|
|
content = "# 架构设计\n\nDocHub 使用 **MinIO** 存储文档，使用 Meilisearch 做全文检索。".encode("utf-8")
|
|
|
|
result = service.save_document("产品文档/架构设计.md", content, "text/markdown")
|
|
|
|
assert minio_client.stored["产品文档/架构设计.md"] == content
|
|
assert search_client.indexed[0]["key"] == "产品文档/架构设计.md"
|
|
assert search_client.indexed[0]["path"] == "产品文档"
|
|
assert search_client.indexed[0]["title"] == "架构设计.md"
|
|
assert "MinIO" in search_client.indexed[0]["content"]
|
|
assert "#" not in search_client.indexed[0]["content"]
|
|
assert "key" in result and "updatedAt" in result
|
|
|
|
|
|
def test_save_document_does_not_fail_when_index_update_fails():
|
|
minio_client = FakeMinioClient()
|
|
search_client = FakeSearchClient(fail_on_index=True)
|
|
service = DocumentService(minio_client, search_client)
|
|
content = b"# doc"
|
|
|
|
result = service.save_document("产品文档/架构设计.md", content, "text/markdown")
|
|
|
|
assert minio_client.stored["产品文档/架构设计.md"] == content
|
|
assert result["key"] == "产品文档/架构设计.md"
|
|
|
|
|
|
def test_delete_document_removes_from_minio_and_search_index():
|
|
minio_client = FakeMinioClient()
|
|
minio_client.stored["产品文档/旧文档.md"] = b"old content"
|
|
search_client = FakeSearchClient()
|
|
service = DocumentService(minio_client, search_client)
|
|
|
|
service.delete_document("产品文档/旧文档.md")
|
|
|
|
assert "产品文档/旧文档.md" not in minio_client.stored
|
|
assert search_client.deleted == ["产品文档/旧文档.md"]
|
|
|
|
|
|
def test_delete_document_does_not_fail_when_index_delete_fails():
|
|
minio_client = FakeMinioClient()
|
|
minio_client.stored["产品文档/旧文档.md"] = b"old content"
|
|
search_client = FakeSearchClient(fail_on_delete=True)
|
|
service = DocumentService(minio_client, search_client)
|
|
|
|
service.delete_document("产品文档/旧文档.md")
|
|
|
|
assert "产品文档/旧文档.md" not in minio_client.stored
|
|
|
|
|
|
def test_upload_document_writes_file_and_returns_metadata():
|
|
minio_client = FakeMinioClient()
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
data = b"# uploaded doc"
|
|
|
|
result = service.upload_document("产品文档/新上传.md", data, "text/markdown")
|
|
|
|
assert result == {"key": "产品文档/新上传.md", "name": "新上传.md", "size": len(data)}
|
|
assert minio_client.stored["产品文档/新上传.md"] == data
|
|
|
|
|
|
def test_upload_document_rejects_unsupported_file_extension():
|
|
minio_client = FakeMinioClient()
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
|
|
with pytest.raises(UnsupportedFileTypeError):
|
|
service.upload_document("产品文档/病毒.exe", b"data", "application/octet-stream")
|
|
|
|
assert "产品文档/病毒.exe" not in minio_client.stored
|
|
|
|
|
|
def test_upload_document_rejects_oversized_file():
|
|
minio_client = FakeMinioClient()
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
data = b"x" * (MAX_UPLOAD_SIZE_BYTES + 1)
|
|
|
|
with pytest.raises(FileTooLargeError):
|
|
service.upload_document("产品文档/大文件.md", data, "text/markdown")
|
|
|
|
assert "产品文档/大文件.md" not in minio_client.stored
|
|
|
|
|
|
def test_get_document_rejects_path_traversal_key():
|
|
service = DocumentService(FakeMinioClient(), FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.get_document("../../etc/passwd")
|
|
|
|
|
|
def test_save_document_rejects_path_traversal_key():
|
|
minio_client = FakeMinioClient()
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.save_document("../../etc/passwd", b"data", "text/plain")
|
|
|
|
assert "../../etc/passwd" not in minio_client.stored
|
|
|
|
|
|
def test_delete_document_rejects_path_traversal_key():
|
|
service = DocumentService(FakeMinioClient(), FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.delete_document("../../etc/passwd")
|
|
|
|
|
|
def test_upload_document_rejects_path_traversal_key():
|
|
minio_client = FakeMinioClient()
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.upload_document("../../etc/passwd", b"data", "text/plain")
|
|
|
|
assert "../../etc/passwd" not in minio_client.stored
|
|
|
|
|
|
def test_upload_document_accepts_allowed_image_extension():
|
|
minio_client = FakeMinioClient()
|
|
service = DocumentService(minio_client, FakeSearchClient())
|
|
data = b"\x89PNG\r\n\x1a\n"
|
|
|
|
result = service.upload_document("产品文档/图片.png", data, "image/png")
|
|
|
|
assert result == {"key": "产品文档/图片.png", "name": "图片.png", "size": len(data)}
|
|
assert minio_client.stored["产品文档/图片.png"] == data
|