"""Punkt 1: veraPDF-Binary wird geprüft — sonst vernichtet es die Originale. Mit `[verapdf].enabled = true` und falschem Pfad lieferte `run_verapdf()` für JEDE Datei False: OCR-Ergebnis nach error/, Original laut `original_on_success = "delete"` gelöscht. Ein Tippfehler im Pfad vernichtete so Scan für Scan die Vorlagen, während die Unit als "läuft" dastand. Zwei Absicherungen: 1. Der Preflight lässt den Dienst gar nicht erst starten (Exit 2). 2. `run_verapdf()` unterscheidet "nicht konform" (False) von "nicht aufrufbar" (`VeraPdfUnavailable`) — Letzteres entsorgt nichts. """ from __future__ import annotations import subprocess import sys from pathlib import Path from unittest.mock import patch import pytest from pdf_ocr_hotfolder.config import OcrConfig, OutputConfig, VeraPdfConfig from pdf_ocr_hotfolder.processor import ( VeraPdfUnavailable, process_pdf, resolve_verapdf_binary, run_verapdf, ) from pdf_ocr_hotfolder.service import ( HotfolderService, PreflightError, check_preflight, check_verapdf_binary, ) ORIGINAL = b"%PDF-1.4 original\n" def _executable(tmp_path: Path, name: str = "verapdf") -> Path: """Legt eine echte, ausführbare Datei an (wird nie wirklich aufgerufen).""" b = tmp_path / name b.write_text("#!/bin/sh\nexit 0\n") b.chmod(0o755) return b # ---------------- check_verapdf_binary ---------------- def test_disabled_verapdf_ignores_binary() -> None: """Solange veraPDF aus ist, darf ein unsinniger Pfad nichts blockieren.""" check_verapdf_binary(False, "/gibt/es/nicht/verapdf") def test_existing_executable_passes(tmp_path: Path) -> None: check_verapdf_binary(True, str(_executable(tmp_path))) def test_missing_binary_raises(tmp_path: Path) -> None: with pytest.raises(PreflightError) as exc: check_verapdf_binary(True, str(tmp_path / "tippfehler")) msg = str(exc.value) assert "verapdf" in msg.lower() assert "tippfehler" in msg def test_non_executable_binary_raises(tmp_path: Path) -> None: """Vorhanden, aber ohne x-Bit: genauso tödlich wie gar nicht vorhanden.""" b = tmp_path / "verapdf" b.write_text("#!/bin/sh\n") b.chmod(0o644) with pytest.raises(PreflightError, match="ausführbar"): check_verapdf_binary(True, str(b)) def test_empty_binary_raises() -> None: with pytest.raises(PreflightError, match=r"\[verapdf\].binary"): check_verapdf_binary(True, "") def test_resolve_finds_binary_in_path(tmp_path: Path, monkeypatch) -> None: """Ein nackter Name wird im PATH gesucht, nicht nur ein absoluter Pfad.""" _executable(tmp_path, "verapdf") monkeypatch.setenv("PATH", str(tmp_path)) assert resolve_verapdf_binary("verapdf") == str(tmp_path / "verapdf") def test_resolve_returns_none_for_empty() -> None: assert resolve_verapdf_binary("") is None # ---------------- check_preflight reicht die veraPDF-Prüfung durch ---------------- def test_check_preflight_checks_verapdf(tmp_path: Path) -> None: with patch("pdf_ocr_hotfolder.service.shutil.which", return_value="/usr/bin/fake"): with pytest.raises(PreflightError, match="verapdf"): check_preflight(verapdf_enabled=True, verapdf_binary=str(tmp_path / "weg")) def test_check_preflight_ok_with_verapdf(tmp_path: Path) -> None: with patch("pdf_ocr_hotfolder.service.shutil.which", return_value="/usr/bin/fake"): check_preflight(verapdf_enabled=True, verapdf_binary=str(_executable(tmp_path))) def test_run_once_aborts_on_broken_verapdf(tmp_config, tmp_path: Path) -> None: """Der Dienst startet nicht — statt Datei für Datei Originale zu löschen.""" tmp_config.verapdf = VeraPdfConfig(enabled=True, binary=str(tmp_path / "gibtsnicht")) service = HotfolderService(tmp_config) try: with patch("pdf_ocr_hotfolder.service.shutil.which", return_value="/usr/bin/fake"): with pytest.raises(PreflightError, match="verapdf"): service.run_once() finally: service._executor.shutdown(wait=False) def test_check_config_returns_2_for_broken_verapdf(tmp_path, tmp_config, monkeypatch, capsys) -> None: """--check-config meldet Exit 2 (der Updater wertet das aus).""" from pdf_ocr_hotfolder.__main__ import CHECK_ERROR, main cfg_file = tmp_path / "cfg.toml" cfg_file.write_text(f""" [paths] incoming = "{tmp_config.paths.incoming}" outgoing = "{tmp_config.paths.outgoing}" working = "{tmp_config.paths.working}" error = "{tmp_config.paths.error}" [verapdf] enabled = true binary = "{tmp_path / 'nicht-da'}" """) monkeypatch.setattr(sys, "argv", ["pdf-ocr-hotfolder", "--config", str(cfg_file), "--check-config"]) with patch("pdf_ocr_hotfolder.service.shutil.which", return_value="/usr/bin/fake"): assert main() == CHECK_ERROR assert "nicht-da" in capsys.readouterr().err def test_check_config_ok_with_working_verapdf(tmp_path, tmp_config, monkeypatch) -> None: from pdf_ocr_hotfolder.__main__ import CHECK_OK, main cfg_file = tmp_path / "cfg.toml" cfg_file.write_text(f""" [paths] incoming = "{tmp_config.paths.incoming}" outgoing = "{tmp_config.paths.outgoing}" working = "{tmp_config.paths.working}" error = "{tmp_config.paths.error}" [verapdf] enabled = true binary = "{_executable(tmp_path)}" """) monkeypatch.setattr(sys, "argv", ["pdf-ocr-hotfolder", "--config", str(cfg_file), "--check-config"]) with patch("pdf_ocr_hotfolder.service.shutil.which", return_value="/usr/bin/fake"): assert main() == CHECK_OK # ---------------- run_verapdf: Urteil vs. Nicht-Aufrufbarkeit ---------------- def _completed(returncode: int, stdout: str = "", stderr: str = ""): return subprocess.CompletedProcess([], returncode, stdout, stderr) def test_run_verapdf_pass(tmp_path: Path) -> None: cfg = VeraPdfConfig(enabled=True, binary=str(_executable(tmp_path))) with patch("pdf_ocr_hotfolder.processor.subprocess.run", return_value=_completed(0, "PASS /tmp/x.pdf\n")): assert run_verapdf(tmp_path / "x.pdf", cfg) is True def test_run_verapdf_fail_is_a_verdict(tmp_path: Path) -> None: """Echtes FAIL bleibt False — das ist ein inhaltliches Urteil.""" cfg = VeraPdfConfig(enabled=True, binary=str(_executable(tmp_path))) with patch("pdf_ocr_hotfolder.processor.subprocess.run", return_value=_completed(1, "FAIL /tmp/x.pdf\n")): assert run_verapdf(tmp_path / "x.pdf", cfg) is False def test_run_verapdf_missing_binary_raises(tmp_path: Path) -> None: """Fehlendes Programm ist KEIN FAIL mehr, sondern ein Fehler.""" cfg = VeraPdfConfig(enabled=True, binary=str(tmp_path / "weg")) with pytest.raises(VeraPdfUnavailable, match="weg"): run_verapdf(tmp_path / "x.pdf", cfg) def test_run_verapdf_timeout_raises(tmp_path: Path) -> None: cfg = VeraPdfConfig(enabled=True, binary=str(_executable(tmp_path))) with patch("pdf_ocr_hotfolder.processor.subprocess.run", side_effect=subprocess.TimeoutExpired("verapdf", 300)): with pytest.raises(VeraPdfUnavailable, match="nicht geantwortet"): run_verapdf(tmp_path / "x.pdf", cfg) def test_run_verapdf_oserror_raises(tmp_path: Path) -> None: cfg = VeraPdfConfig(enabled=True, binary=str(_executable(tmp_path))) with patch("pdf_ocr_hotfolder.processor.subprocess.run", side_effect=OSError("Exec format error")): with pytest.raises(VeraPdfUnavailable, match="nicht startbar"): run_verapdf(tmp_path / "x.pdf", cfg) def test_run_verapdf_without_verdict_raises(tmp_path: Path) -> None: """Startet der Wrapper nicht durch (fehlendes Java), steht kein Urteil da. Exit != 0 ohne PASS/FAIL in der Ausgabe darf nicht als "nicht konform" durchgehen — genau so würde der Original-Löschpfad wieder aufgehen. """ cfg = VeraPdfConfig(enabled=True, binary=str(_executable(tmp_path))) with patch("pdf_ocr_hotfolder.processor.subprocess.run", return_value=_completed(127, "", "java: command not found")): with pytest.raises(VeraPdfUnavailable, match="kein Urteil"): run_verapdf(tmp_path / "x.pdf", cfg) def test_run_verapdf_disabled_returns_true(tmp_path: Path) -> None: assert run_verapdf(tmp_path / "x.pdf", VeraPdfConfig(enabled=False)) is True # ---------------- process_pdf: nicht aufrufbares veraPDF entsorgt nichts ---------------- def _fake_ocr(src: Path, dst: Path, cfg: OcrConfig) -> None: dst.write_bytes(b"%PDF-1.4 OCRed\n") def _prepare(tmp_path: Path) -> dict: dirs = {name: tmp_path / name for name in ("incoming", "working", "outgoing", "error", "archive")} for d in dirs.values(): d.mkdir(parents=True, exist_ok=True) src = dirs["incoming"] / "scan.pdf" src.write_bytes(ORIGINAL) return {"src": src, **dirs} def _run_unavailable(env: dict, out_cfg: OutputConfig): with patch("pdf_ocr_hotfolder.processor.run_ocr", side_effect=_fake_ocr), \ patch("pdf_ocr_hotfolder.processor.run_verapdf", side_effect=VeraPdfUnavailable("Binary weg")): return process_pdf( src=env["src"], working_dir=env["working"], outgoing_dir=env["outgoing"], error_dir=env["error"], ocr_cfg=OcrConfig(), vera_cfg=VeraPdfConfig(enabled=True), output_cfg=out_cfg, ) def test_unavailable_verapdf_keeps_original_despite_delete(tmp_path: Path) -> None: """Der gefährliche Fall: original_on_success='delete' darf nicht greifen.""" env = _prepare(tmp_path) result = _run_unavailable(env, OutputConfig(name_mode="prefix", name_tag="OCR_", original_on_success="delete")) assert not result.success # Original ist NICHT weg, sondern in error/ gesichert gesichert = env["error"] / "scan.pdf" assert gesichert.exists() assert gesichert.read_bytes() == ORIGINAL assert not (env["working"] / "scan.pdf").exists() # Kein fertiges Ergebnis in outgoing/ assert list(env["outgoing"].iterdir()) == [] def test_unavailable_verapdf_result_is_not_a_fail_verdict(tmp_path: Path) -> None: """verapdf_passed bleibt None: es gab kein Urteil, nur einen Fehler.""" env = _prepare(tmp_path) result = _run_unavailable(env, OutputConfig(original_on_success="delete")) assert result.verapdf_passed is None assert "veraPDF" in result.error def test_unavailable_verapdf_also_keeps_ocr_result(tmp_path: Path) -> None: env = _prepare(tmp_path) _run_unavailable(env, OutputConfig(name_mode="prefix", name_tag="OCR_", original_on_success="delete")) assert (env["error"] / "__ocr_OCR_scan.pdf").exists() assert list(env["working"].iterdir()) == []