Tianyang
360e9e5a14
The backend no longer mentions MinIO anywhere: MinioClient -> GarageClient (minio_client.py -> garage_client.py), MINIO_* env vars -> GARAGE_*, and the `minio` Python package is replaced with `boto3` (AWS's generic S3 client) since Garage is the only storage backend this project targets now. S3 API semantics are unchanged, so document_service.py's behavior is identical — this is a naming and dependency swap, not a logic change. Also updates test fixtures/names (FakeMinioClient -> FakeGarageClient, FakeMinioSDK -> FakeS3SDK) and storage/README.md, which no longer frames Garage as a MinIO replacement (that migration note has served its purpose now that no MinIO reference remains in the codebase). Backend: 54/54 tests passing. Frontend: 38/38 tests passing (unaffected, verified for regressions since it talks to the backend only through the unchanged REST API).
243 lines
8.7 KiB
Python
243 lines
8.7 KiB
Python
import pytest
|
|
|
|
from app.document_service import (
|
|
DocumentService,
|
|
DocumentNotFoundError,
|
|
FileTooLargeError,
|
|
InvalidPathError,
|
|
MAX_UPLOAD_SIZE_BYTES,
|
|
UnsupportedFileTypeError,
|
|
)
|
|
|
|
|
|
class FakeGarageClient:
|
|
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():
|
|
garage_client = FakeGarageClient()
|
|
garage_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(garage_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():
|
|
garage_client = FakeGarageClient()
|
|
garage_client.stored["产品文档/架构设计.md"] = "# 架构设计\n\nDocHub 使用 Garage 存储文档。".encode("utf-8")
|
|
service = DocumentService(garage_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(FakeGarageClient(), FakeSearchClient())
|
|
|
|
with pytest.raises(DocumentNotFoundError):
|
|
service.get_document("不存在/文档.md")
|
|
|
|
|
|
def test_get_document_returns_base64_content_for_binary_data_without_raising():
|
|
garage_client = FakeGarageClient()
|
|
binary_data = b"\x89PNG\r\n\x1a\n\x00\x01\xff\xfe"
|
|
garage_client.stored["产品文档/图片.png"] = binary_data
|
|
service = DocumentService(garage_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():
|
|
garage_client = FakeGarageClient()
|
|
garage_client.stored["产品文档/架构设计.md"] = "# 架构设计".encode("utf-8")
|
|
service = DocumentService(garage_client, FakeSearchClient())
|
|
|
|
result = service.get_document("产品文档/架构设计.md")
|
|
|
|
assert result["encoding"] == "utf-8"
|
|
|
|
|
|
def test_save_document_writes_to_garage_then_indexes_content():
|
|
garage_client = FakeGarageClient()
|
|
search_client = FakeSearchClient()
|
|
service = DocumentService(garage_client, search_client)
|
|
content = "# 架构设计\n\nDocHub 使用 **Garage** 存储文档，使用 Meilisearch 做全文检索。".encode("utf-8")
|
|
|
|
result = service.save_document("产品文档/架构设计.md", content, "text/markdown")
|
|
|
|
assert garage_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 "Garage" 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():
|
|
garage_client = FakeGarageClient()
|
|
search_client = FakeSearchClient(fail_on_index=True)
|
|
service = DocumentService(garage_client, search_client)
|
|
content = b"# doc"
|
|
|
|
result = service.save_document("产品文档/架构设计.md", content, "text/markdown")
|
|
|
|
assert garage_client.stored["产品文档/架构设计.md"] == content
|
|
assert result["key"] == "产品文档/架构设计.md"
|
|
|
|
|
|
def test_delete_document_removes_from_garage_and_search_index():
|
|
garage_client = FakeGarageClient()
|
|
garage_client.stored["产品文档/旧文档.md"] = b"old content"
|
|
search_client = FakeSearchClient()
|
|
service = DocumentService(garage_client, search_client)
|
|
|
|
service.delete_document("产品文档/旧文档.md")
|
|
|
|
assert "产品文档/旧文档.md" not in garage_client.stored
|
|
assert search_client.deleted == ["产品文档/旧文档.md"]
|
|
|
|
|
|
def test_delete_document_does_not_fail_when_index_delete_fails():
|
|
garage_client = FakeGarageClient()
|
|
garage_client.stored["产品文档/旧文档.md"] = b"old content"
|
|
search_client = FakeSearchClient(fail_on_delete=True)
|
|
service = DocumentService(garage_client, search_client)
|
|
|
|
service.delete_document("产品文档/旧文档.md")
|
|
|
|
assert "产品文档/旧文档.md" not in garage_client.stored
|
|
|
|
|
|
def test_upload_document_writes_file_and_returns_metadata():
|
|
garage_client = FakeGarageClient()
|
|
service = DocumentService(garage_client, FakeSearchClient())
|
|
data = b"# uploaded doc"
|
|
|
|
result = service.upload_document("产品文档/新上传.md", data, "text/markdown")
|
|
|
|
assert result == {"key": "产品文档/新上传.md", "name": "新上传.md", "size": len(data)}
|
|
assert garage_client.stored["产品文档/新上传.md"] == data
|
|
|
|
|
|
def test_upload_document_rejects_unsupported_file_extension():
|
|
garage_client = FakeGarageClient()
|
|
service = DocumentService(garage_client, FakeSearchClient())
|
|
|
|
with pytest.raises(UnsupportedFileTypeError):
|
|
service.upload_document("产品文档/病毒.exe", b"data", "application/octet-stream")
|
|
|
|
assert "产品文档/病毒.exe" not in garage_client.stored
|
|
|
|
|
|
def test_upload_document_rejects_oversized_file():
|
|
garage_client = FakeGarageClient()
|
|
service = DocumentService(garage_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 garage_client.stored
|
|
|
|
|
|
def test_get_document_rejects_path_traversal_key():
|
|
service = DocumentService(FakeGarageClient(), FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.get_document("../../etc/passwd")
|
|
|
|
|
|
def test_save_document_rejects_path_traversal_key():
|
|
garage_client = FakeGarageClient()
|
|
service = DocumentService(garage_client, FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.save_document("../../etc/passwd", b"data", "text/plain")
|
|
|
|
assert "../../etc/passwd" not in garage_client.stored
|
|
|
|
|
|
def test_delete_document_rejects_path_traversal_key():
|
|
service = DocumentService(FakeGarageClient(), FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.delete_document("../../etc/passwd")
|
|
|
|
|
|
def test_upload_document_rejects_path_traversal_key():
|
|
garage_client = FakeGarageClient()
|
|
service = DocumentService(garage_client, FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.upload_document("../../etc/passwd", b"data", "text/plain")
|
|
|
|
assert "../../etc/passwd" not in garage_client.stored
|
|
|
|
|
|
def test_upload_document_accepts_allowed_image_extension():
|
|
garage_client = FakeGarageClient()
|
|
service = DocumentService(garage_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 garage_client.stored["产品文档/图片.png"] == data
|