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

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

from __future__ import annotations

import subprocess
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,
)


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_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, "同一文件哈希应只触发一次真实转换"