Tianyang
4c19fad561
Single-node deployment doesn't benefit from S3 abstraction overhead; LocalFileClient keeps the same interface DocumentService depends on, backed by a STORAGE_DIR volume mount instead of a separate Garage service.
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 FakeFileClient:
|
|
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():
|
|
file_client = FakeFileClient()
|
|
file_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(file_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():
|
|
file_client = FakeFileClient()
|
|
file_client.stored["产品文档/架构设计.md"] = "# 架构设计\n\nDocHub 使用 Garage 存储文档。".encode("utf-8")
|
|
service = DocumentService(file_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(FakeFileClient(), FakeSearchClient())
|
|
|
|
with pytest.raises(DocumentNotFoundError):
|
|
service.get_document("不存在/文档.md")
|
|
|
|
|
|
def test_get_document_returns_base64_content_for_binary_data_without_raising():
|
|
file_client = FakeFileClient()
|
|
binary_data = b"\x89PNG\r\n\x1a\n\x00\x01\xff\xfe"
|
|
file_client.stored["产品文档/图片.png"] = binary_data
|
|
service = DocumentService(file_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():
|
|
file_client = FakeFileClient()
|
|
file_client.stored["产品文档/架构设计.md"] = "# 架构设计".encode("utf-8")
|
|
service = DocumentService(file_client, FakeSearchClient())
|
|
|
|
result = service.get_document("产品文档/架构设计.md")
|
|
|
|
assert result["encoding"] == "utf-8"
|
|
|
|
|
|
def test_save_document_writes_to_garage_then_indexes_content():
|
|
file_client = FakeFileClient()
|
|
search_client = FakeSearchClient()
|
|
service = DocumentService(file_client, search_client)
|
|
content = "# 架构设计\n\nDocHub 使用 **Garage** 存储文档，使用 Meilisearch 做全文检索。".encode("utf-8")
|
|
|
|
result = service.save_document("产品文档/架构设计.md", content, "text/markdown")
|
|
|
|
assert file_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():
|
|
file_client = FakeFileClient()
|
|
search_client = FakeSearchClient(fail_on_index=True)
|
|
service = DocumentService(file_client, search_client)
|
|
content = b"# doc"
|
|
|
|
result = service.save_document("产品文档/架构设计.md", content, "text/markdown")
|
|
|
|
assert file_client.stored["产品文档/架构设计.md"] == content
|
|
assert result["key"] == "产品文档/架构设计.md"
|
|
|
|
|
|
def test_delete_document_removes_from_garage_and_search_index():
|
|
file_client = FakeFileClient()
|
|
file_client.stored["产品文档/旧文档.md"] = b"old content"
|
|
search_client = FakeSearchClient()
|
|
service = DocumentService(file_client, search_client)
|
|
|
|
service.delete_document("产品文档/旧文档.md")
|
|
|
|
assert "产品文档/旧文档.md" not in file_client.stored
|
|
assert search_client.deleted == ["产品文档/旧文档.md"]
|
|
|
|
|
|
def test_delete_document_does_not_fail_when_index_delete_fails():
|
|
file_client = FakeFileClient()
|
|
file_client.stored["产品文档/旧文档.md"] = b"old content"
|
|
search_client = FakeSearchClient(fail_on_delete=True)
|
|
service = DocumentService(file_client, search_client)
|
|
|
|
service.delete_document("产品文档/旧文档.md")
|
|
|
|
assert "产品文档/旧文档.md" not in file_client.stored
|
|
|
|
|
|
def test_upload_document_writes_file_and_returns_metadata():
|
|
file_client = FakeFileClient()
|
|
service = DocumentService(file_client, FakeSearchClient())
|
|
data = b"# uploaded doc"
|
|
|
|
result = service.upload_document("产品文档/新上传.md", data, "text/markdown")
|
|
|
|
assert result == {"key": "产品文档/新上传.md", "name": "新上传.md", "size": len(data)}
|
|
assert file_client.stored["产品文档/新上传.md"] == data
|
|
|
|
|
|
def test_upload_document_rejects_unsupported_file_extension():
|
|
file_client = FakeFileClient()
|
|
service = DocumentService(file_client, FakeSearchClient())
|
|
|
|
with pytest.raises(UnsupportedFileTypeError):
|
|
service.upload_document("产品文档/病毒.exe", b"data", "application/octet-stream")
|
|
|
|
assert "产品文档/病毒.exe" not in file_client.stored
|
|
|
|
|
|
def test_upload_document_rejects_oversized_file():
|
|
file_client = FakeFileClient()
|
|
service = DocumentService(file_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 file_client.stored
|
|
|
|
|
|
def test_get_document_rejects_path_traversal_key():
|
|
service = DocumentService(FakeFileClient(), FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.get_document("../../etc/passwd")
|
|
|
|
|
|
def test_save_document_rejects_path_traversal_key():
|
|
file_client = FakeFileClient()
|
|
service = DocumentService(file_client, FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.save_document("../../etc/passwd", b"data", "text/plain")
|
|
|
|
assert "../../etc/passwd" not in file_client.stored
|
|
|
|
|
|
def test_delete_document_rejects_path_traversal_key():
|
|
service = DocumentService(FakeFileClient(), FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.delete_document("../../etc/passwd")
|
|
|
|
|
|
def test_upload_document_rejects_path_traversal_key():
|
|
file_client = FakeFileClient()
|
|
service = DocumentService(file_client, FakeSearchClient())
|
|
|
|
with pytest.raises(InvalidPathError):
|
|
service.upload_document("../../etc/passwd", b"data", "text/plain")
|
|
|
|
assert "../../etc/passwd" not in file_client.stored
|
|
|
|
|
|
def test_upload_document_accepts_allowed_image_extension():
|
|
file_client = FakeFileClient()
|
|
service = DocumentService(file_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 file_client.stored["产品文档/图片.png"] == data
|