document-management/backend/tests/dms/services/test_preview_service.py

"""PreviewService 单元测试，不依赖数据库或真实 LibreOffice。"""

from __future__ import annotations

from pathlib import Path

import pytest

from dms.services.document_converter import FakeDocumentConverter
from dms.services.preview_service import PreviewService, preview_cache_key
from dms.storage.paths import UnsafeStoragePathError, preview_cache_path


@pytest.fixture
def storage_root(tmp_path: Path) -> Path:
    root = tmp_path / "dms-storage"
    (root / "preview").mkdir(parents=True)
    (root / "original").mkdir(parents=True)
    return root


@pytest.fixture
def preview_service(storage_root: Path) -> PreviewService:
    return PreviewService(storage_root, FakeDocumentConverter())


def test_preview_cache_path_with_hash_prefix(storage_root: Path) -> None:
    path = preview_cache_path("abcdef123456", storage_root)
    assert path.name == "abcdef123456.pdf"
    assert path.parent.name == "ab"
    assert path.parent.parent.name == "preview"


def test_preview_cache_path_rejects_invalid_hash(storage_root: Path) -> None:
    with pytest.raises(UnsafeStoragePathError):
        preview_cache_path("", storage_root)


def test_validate_cached_pdf_accepts_valid_pdf(
    preview_service: PreviewService, storage_root: Path
) -> None:
    cache = preview_cache_path("validhash", storage_root)
    cache.parent.mkdir(parents=True, exist_ok=True)
    # 生成 fitz 可真正打开的合法最小 PDF，避免仅写入魔数导致验证失败
    import fitz

    doc = fitz.open()
    doc.new_page()
    doc.save(str(cache))
    doc.close()
    assert preview_service.validate_cached_pdf(cache) is True


def test_validate_cached_pdf_rejects_non_pdf(
    preview_service: PreviewService, storage_root: Path
) -> None:
    cache = preview_cache_path("nonpdfhash", storage_root)
    cache.parent.mkdir(parents=True, exist_ok=True)
    cache.write_bytes(b"not a pdf")
    assert preview_service.validate_cached_pdf(cache) is False


def test_validate_cached_pdf_rejects_path_outside_preview(
    preview_service: PreviewService, tmp_path: Path
) -> None:
    outside = tmp_path / "outside.pdf"
    outside.write_bytes(b"%PDF-1.4")
    assert preview_service.validate_cached_pdf(outside) is False


class FakeDocument:
    def __init__(
        self,
        file_extension: str,
        file_hash: str,
        file_relative_path: str,
    ) -> None:
        self.file_extension = file_extension
        self.file_hash = file_hash
        self.file_relative_path = file_relative_path


def test_preview_path_for_pdf_returns_original(
    preview_service: PreviewService, storage_root: Path
) -> None:
    original = storage_root / "original" / "sample.pdf"
    original.write_bytes(b"%PDF-1.4")
    document = FakeDocument("pdf", "pdfhash", "original/sample.pdf")
    path = preview_service.preview_path_for(document)
    assert path == original


def test_preview_path_for_docx_generates_cache(
    preview_service: PreviewService, storage_root: Path
) -> None:
    original = storage_root / "original" / "sample.docx"
    original.write_bytes(b"fake docx content")
    document = FakeDocument("docx", "docxhash", "original/sample.docx")
    path = preview_service.preview_path_for(document)
    assert path.name == "docxhash.pdf"
    assert path.exists()
    assert path.read_bytes().startswith(b"%PDF-")


def test_preview_path_for_wps_generates_cache(
    preview_service: PreviewService, storage_root: Path
) -> None:
    original = storage_root / "original" / "sample.wps"
    original.write_bytes(b"fake wps content")
    document = FakeDocument("wps", "wpshash", "original/sample.wps")
    path = preview_service.preview_path_for(document)
    assert path.name == "wpshash.pdf"
    assert path.exists()
    assert path.read_bytes().startswith(b"%PDF-")


def test_preview_path_for_doc_uses_cache_on_second_call(
    preview_service: PreviewService, storage_root: Path
) -> None:
    original = storage_root / "original" / "sample.doc"
    original.write_bytes(b"fake doc content")
    document = FakeDocument("doc", "dochash", "original/sample.doc")
    first = preview_service.preview_path_for(document)
    mtime = first.stat().st_mtime
    second = preview_service.preview_path_for(document)
    assert second == first
    assert second.stat().st_mtime == mtime


def test_preview_cache_key_spreadsheet_gets_version_suffix() -> None:
    assert preview_cache_key("abcdef123456", "xlsx", True) == "abcdef123456-sps2"
    assert preview_cache_key("abcdef123456", "xlsx", False) == "abcdef123456-sps0"
    assert preview_cache_key("abcdef123456", "XLS", True) == "abcdef123456-sps2"
    assert preview_cache_key("abcdef123456", ".et", False) == "abcdef123456-sps0"
    assert preview_cache_key("abcdef123456", "docx", True) == "abcdef123456"
    assert preview_cache_key("abcdef123456", "wps", False) == "abcdef123456"


def test_preview_path_for_xlsx_uses_suffixed_cache(
    preview_service: PreviewService, storage_root: Path
) -> None:
    original = storage_root / "original" / "sample.xlsx"
    original.write_bytes(b"fake xlsx content")
    document = FakeDocument("xlsx", "xlshash", "original/sample.xlsx")
    path = preview_service.preview_path_for(document)
    # 表格类缓存键带策略版本后缀（Fake 转换器不支持单页 → 分页键 sps0）
    assert path.name == "xlshash-sps0.pdf"
    assert path.exists()
    assert path.read_bytes().startswith(b"%PDF-")


def test_preview_reuses_valid_cache_when_replace_occupied(
    preview_service: PreviewService,
    storage_root: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """并发场景：os.replace 因目标被先完成方读取占用而失败时，幂等复用有效缓存。"""
    import fitz
    import os as os_module

    original = storage_root / "original" / "sample.docx"
    original.write_bytes(b"fake docx content")
    document = FakeDocument("docx", "occupiedhash", "original/sample.docx")
    cache = preview_cache_path("occupiedhash", storage_root)

    # 预置无效缓存，确保前置校验不命中、走转换写缓存路径
    cache.parent.mkdir(parents=True, exist_ok=True)
    cache.write_bytes(b"not a pdf")

    def fake_replace(src: Path, dst: Path) -> None:
        # 模拟并发先完成方：已写好有效缓存且目标正被其读取
        # （Windows 下共享模式不含删除权限 → 替换失败抛 PermissionError）
        doc = fitz.open()
        doc.new_page()
        doc.save(str(dst))
        doc.close()
        raise PermissionError(13, "另一个程序正在使用此文件，进程无法访问")

    monkeypatch.setattr(os_module, "replace", fake_replace)
    path = preview_service.preview_path_for(document)
    assert path == cache
    assert preview_service.validate_cached_pdf(cache) is True
    # 临时文件已清理，不残留 .tmp
    assert list(cache.parent.glob("*.tmp")) == []