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

"""DocumentConverter 单元测试，不依赖真实 LibreOffice。"""

from __future__ import annotations

import subprocess
import sys
import threading
import time
from pathlib import Path

import pytest
import fitz

from dms.common.errors import (
    PreviewConversionFailedError,
    PreviewConversionTimeoutError,
    PreviewConverterUnavailableError,
)
from dms.services.document_converter import (
    FakeDocumentConverter,
    LibreOfficeDocumentConverter,
    _version_at_least_7_4,
)


def test_fake_converter_available_by_default() -> None:
    converter = FakeDocumentConverter()
    assert converter.available is True


def test_fake_converter_returns_pdf_bytes(tmp_path: Path) -> None:
    converter = FakeDocumentConverter()
    source = tmp_path / "test.docx"
    source.write_bytes(b"hello document")
    result = converter.convert_to_pdf(source, "abcd1234")
    assert result.startswith(b"%PDF-")
    with fitz.open(stream=result, filetype="pdf") as document:
        assert "hello document" in "".join(page.get_text() for page in document)
    assert len(converter.calls) == 1
    assert converter.calls[0] == (source, "abcd1234")


def test_fake_converter_unavailable_raises() -> None:
    converter = FakeDocumentConverter(enabled=False)
    assert converter.available is False
    source = Path("/nonexistent/test.docx")
    with pytest.raises(PreviewConverterUnavailableError):
        converter.convert_to_pdf(source, "hash")


def test_libreoffice_unavailable_when_disabled() -> None:
    converter = LibreOfficeDocumentConverter(
        executable="/usr/bin/soffice",
        timeout_seconds=60,
        max_concurrency=2,
        enabled=False,
    )
    assert converter.available is False


def test_version_at_least_7_4() -> None:
    assert _version_at_least_7_4("LibreOffice 7.4") is True
    assert _version_at_least_7_4("LibreOffice 7.6.7.2 (X86_64)") is True
    assert _version_at_least_7_4("LibreOffice 24.8.4.2 (X86_64)") is True
    assert _version_at_least_7_4("LibreOffice 25.2 (x86_64)") is True
    assert _version_at_least_7_4("LibreOffice 7.3.7.2") is False
    assert _version_at_least_7_4("LibreOffice 6.4.7.2") is False
    assert _version_at_least_7_4("") is False
    assert _version_at_least_7_4("no version here") is False


@pytest.mark.skipif(sys.platform != "win32", reason="soffice.com 仅 Windows 存在")
def test_version_probe_prefers_soffice_com_on_windows(tmp_path: Path) -> None:
    exe = tmp_path / "soffice.exe"
    exe.write_bytes(b"MZ")
    com = tmp_path / "soffice.com"
    com.write_bytes(b"MZ")
    converter = LibreOfficeDocumentConverter(
        executable=str(exe),
        timeout_seconds=5,
        max_concurrency=1,
    )
    assert converter._version_probe_executable() == str(com)


def test_version_probe_falls_back_when_com_missing(tmp_path: Path) -> None:
    exe = tmp_path / "soffice.exe"
    exe.write_bytes(b"MZ")
    converter = LibreOfficeDocumentConverter(
        executable=str(exe),
        timeout_seconds=5,
        max_concurrency=1,
    )
    assert converter._version_probe_executable() == str(exe)


class _CompletedVersion:
    def __init__(self, stdout: str) -> None:
        self.stdout = stdout
        self.stderr = ""


def test_single_page_detect_success_cached_permanently(
    tmp_path: Path, monkeypatch
) -> None:
    exe = tmp_path / "soffice.exe"
    exe.write_bytes(b"MZ")
    converter = LibreOfficeDocumentConverter(
        executable=str(exe),
        timeout_seconds=5,
        max_concurrency=1,
    )
    calls: list[tuple] = []

    def fake_run(*args, **kwargs):
        calls.append(args)
        return _CompletedVersion("LibreOffice 26.2.5.2")

    monkeypatch.setattr(subprocess, "run", fake_run)
    assert converter._supports_single_page_sheets() is True
    assert converter._supports_single_page_sheets() is True
    assert len(calls) == 1, "探测成功应永久缓存，不重复执行 --version"


def test_single_page_detect_failure_retried_after_ttl(
    tmp_path: Path, monkeypatch
) -> None:
    exe = tmp_path / "soffice.exe"
    exe.write_bytes(b"MZ")
    converter = LibreOfficeDocumentConverter(
        executable=str(exe),
        timeout_seconds=5,
        max_concurrency=1,
    )
    calls: list[tuple] = []
    state = {"fail": True}

    def fake_run(*args, **kwargs):
        calls.append(args)
        if state["fail"]:
            raise OSError("probe failed")
        return _CompletedVersion("LibreOffice 26.2.5.2")

    monkeypatch.setattr(subprocess, "run", fake_run)
    assert converter._supports_single_page_sheets() is False
    assert converter._supports_single_page_sheets() is False
    assert len(calls) == 1, "TTL 内的失败结果应被缓存，不重复探测"

    # TTL 过期后允许重试；恢复成功后转为永久缓存
    monkeypatch.setattr(
        "dms.services.document_converter._SINGLE_PAGE_DETECT_RETRY_SECONDS", 0.0
    )
    state["fail"] = False
    assert converter._supports_single_page_sheets() is True
    assert len(calls) == 2
    assert converter._supports_single_page_sheets() is True
    assert len(calls) == 2, "重试成功后应转为永久缓存"


def test_libreoffice_unavailable_when_executable_missing() -> None:
    converter = LibreOfficeDocumentConverter(
        executable="/definitely/not/exists/soffice",
        timeout_seconds=60,
        max_concurrency=2,
    )
    assert converter.available is False
    source = Path(__file__)
    with pytest.raises(PreviewConverterUnavailableError):
        converter.convert_to_pdf(source, "hash")


class FakeProcess:
    def __init__(self, returncode: int = 0, timeout: bool = False) -> None:
        self._returncode = returncode
        self._timeout = timeout

    def communicate(self, timeout: float | None = None) -> tuple[bytes, bytes]:
        if self._timeout:
            raise subprocess.TimeoutExpired(cmd=["soffice"], timeout=timeout or 1)
        return b"", b""

    def poll(self) -> int | None:
        return self._returncode

    def terminate(self) -> None:
        pass

    def wait(self, timeout: float | None = None) -> int:
        return self._returncode

    @property
    def returncode(self) -> int:
        return self._returncode


def test_libreoffice_conversion_fails_with_non_zero_exit(tmp_path: Path, monkeypatch) -> None:
    real_exe = tmp_path / "soffice.exe"
    real_exe.write_bytes(b"MZ")
    converter = LibreOfficeDocumentConverter(
        executable=str(real_exe),
        timeout_seconds=5,
        max_concurrency=1,
    )
    assert converter.available is True

    def fake_popen(*args, **kwargs):
        return FakeProcess(returncode=1)

    monkeypatch.setattr(subprocess, "Popen", fake_popen)

    source = tmp_path / "test.docx"
    source.write_bytes(b"not a real docx")
    with pytest.raises(PreviewConversionFailedError):
        converter.convert_to_pdf(source, "hash")


def test_libreoffice_timeout_cleans_up(tmp_path: Path, monkeypatch) -> None:
    real_exe = tmp_path / "soffice.exe"
    real_exe.write_bytes(b"MZ")
    converter = LibreOfficeDocumentConverter(
        executable=str(real_exe),
        timeout_seconds=1,
        max_concurrency=1,
    )
    assert converter.available is True

    def fake_popen(*args, **kwargs):
        return FakeProcess(timeout=True)

    monkeypatch.setattr(subprocess, "Popen", fake_popen)

    source = tmp_path / "test.docx"
    source.write_bytes(b"not a real docx")
    with pytest.raises(PreviewConversionTimeoutError):
        converter.convert_to_pdf(source, "hash")


def test_libreoffice_concurrent_same_hash_deduplicates(tmp_path: Path, monkeypatch) -> None:
    real_exe = tmp_path / "soffice.exe"
    real_exe.write_bytes(b"MZ")
    converter = LibreOfficeDocumentConverter(
        executable=str(real_exe),
        timeout_seconds=5,
        max_concurrency=2,
    )
    calls: list[tuple] = []
    lock = threading.Lock()
    ready = threading.Event()

    def fake_popen(*args, **kwargs):
        with lock:
            calls.append(args)
        ready.set()
        time.sleep(0.25)
        argv = args[0]
        out_dir = Path(argv[argv.index("--outdir") + 1])
        input_file = Path(argv[-1])
        out_dir.mkdir(parents=True, exist_ok=True)
        (out_dir / f"{input_file.stem}.pdf").write_bytes(b"%PDF-1.4 fake")
        return FakeProcess(returncode=0)

    monkeypatch.setattr(subprocess, "Popen", fake_popen)

    source = tmp_path / "test.docx"
    source.write_bytes(b"shared")
    results: list[bytes] = []

    def call() -> None:
        results.append(converter.convert_to_pdf(source, "same_hash"))

    t1 = threading.Thread(target=call)
    t2 = threading.Thread(target=call)
    t1.start()
    ready.wait(timeout=1)
    t2.start()
    t1.join(timeout=5)
    t2.join(timeout=5)

    assert len(results) == 2
    assert all(r == results[0] for r in results)
    assert len(calls) == 1, "同一文件哈希应只触发一次真实转换"