147 lines
5.4 KiB
Python
147 lines
5.4 KiB
Python
import pytest
|
|
|
|
from app.document_service import DocumentService, DocumentNotFoundError
|
|
|
|
|
|
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_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
|