DocHub/backend/tests/test_document_service.py

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)

    def walk_files(self):
        return list(self.stored.keys())


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
        self.cleared = False

    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 delete_all_documents(self):
        self.cleared = True


def test_list_documents_sorts_files_naturally_by_numeric_prefix():
    file_client = FakeFileClient()
    file_client.entries = [
        {"key": "8.10 系统设计.md", "is_dir": False, "size": 1, "modifiedAt": None},
        {"key": "8.1 GMP调度.md", "is_dir": False, "size": 1, "modifiedAt": None},
        {"key": "8.2 Channel.md", "is_dir": False, "size": 1, "modifiedAt": None},
    ]
    service = DocumentService(file_client, FakeSearchClient())

    result = service.list_documents("")

    assert [f["name"] for f in result["files"]] == [
        "8.1 GMP调度.md",
        "8.2 Channel.md",
        "8.10 系统设计.md",
    ]


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


def test_reindex_all_clears_index_then_reindexes_every_supported_file():
    file_client = FakeFileClient()
    file_client.stored["产品文档/架构设计.md"] = "# 架构设计".encode("utf-8")
    file_client.stored["产品文档/图片.png"] = b"\x89PNG\r\n\x1a\n"
    search_client = FakeSearchClient()
    service = DocumentService(file_client, search_client)

    result = service.reindex_all()

    assert search_client.cleared
    assert result == {"indexed": 1, "failed": 0, "total": 2}
    assert search_client.indexed[0]["key"] == "产品文档/架构设计.md"


def test_reindex_all_counts_failures_without_raising():
    file_client = FakeFileClient()
    file_client.stored["产品文档/架构设计.md"] = "# 架构设计".encode("utf-8")
    search_client = FakeSearchClient(fail_on_index=True)
    service = DocumentService(file_client, search_client)

    result = service.reindex_all()

    assert result == {"indexed": 0, "failed": 1, "total": 1}