from __future__ import annotations from pathlib import Path from types import SimpleNamespace import pytest from deerflow.utils import file_conversion @pytest.fixture(autouse=True) def local_parser(monkeypatch): # Parser-selection unit tests inject readers locally. Production always # sends this same top-level parser entry point to spawned worker processes. monkeypatch.setattr( file_conversion, "_run_conversion_in_process", lambda path, converter: file_conversion._convert_document_worker(str(path), converter), ) def test_legacy_word_is_normalized_before_markitdown(monkeypatch, tmp_path: Path) -> None: source = tmp_path / "legacy.doc" source.write_bytes(b"legacy") monkeypatch.setattr(file_conversion.shutil, "which", lambda _name: "/usr/bin/soffice") monkeypatch.setattr( file_conversion, "_convert_with_markitdown", lambda path: f"converted:{path.suffix}", ) def fake_run(command, **_kwargs): output_dir = Path(command[command.index("--outdir") + 1]) (output_dir / "legacy.docx").write_bytes(b"normalized") return SimpleNamespace(returncode=0, stderr="") monkeypatch.setattr(file_conversion.subprocess, "run", fake_run) assert file_conversion._do_convert(source, "auto") == "converted:.docx" def test_modern_office_file_does_not_start_libreoffice(monkeypatch, tmp_path: Path) -> None: source = tmp_path / "modern.pptx" source.write_bytes(b"modern") monkeypatch.setattr( file_conversion, "_convert_with_markitdown", lambda path: f"direct:{path.suffix}", ) monkeypatch.setattr( file_conversion.subprocess, "run", lambda *_args, **_kwargs: (_ for _ in ()).throw(AssertionError("unexpected LibreOffice call")), ) assert file_conversion._do_convert(source, "auto") == "direct:.pptx" def test_legacy_office_falls_back_when_libreoffice_is_missing(monkeypatch, tmp_path: Path) -> None: source = tmp_path / "legacy.xls" source.write_bytes(b"legacy") monkeypatch.setattr(file_conversion.shutil, "which", lambda _name: None) monkeypatch.setattr( file_conversion, "_convert_with_markitdown", lambda path: f"fallback:{path.suffix}", ) assert file_conversion._do_convert(source, "auto") == "fallback:.xls"