techadmin
a23a3968ef
Neue [output]-Section: - name_mode: prefix | suffix | none (suffix wird vor Extension eingefügt) - name_tag: verbatim einfügbarer String - original_on_success: delete | archive - archive_dir mit Kollisions-Schutz (Timestamp-Suffix) 20 neue Tests (50 insgesamt, alle grün). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
97 lines
3.6 KiB
Python
97 lines
3.6 KiB
Python
"""Tests für Issue #2: --once Modus muss Exit-Code != 0 bei Fehlern liefern."""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from pdf_ocr_hotfolder.processor import ProcessResult
|
|
from pdf_ocr_hotfolder.service import HotfolderService
|
|
|
|
|
|
def _fake_success(src: Path, working_dir, outgoing_dir, error_dir, **kwargs):
|
|
out = outgoing_dir / f"OCR_{src.name}"
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
out.write_bytes(b"%PDF-1.4 ocr\n")
|
|
src.unlink(missing_ok=True)
|
|
return ProcessResult(src, out, True)
|
|
|
|
|
|
def _fake_failure(src: Path, working_dir, outgoing_dir, error_dir, **kwargs):
|
|
error_dir.mkdir(parents=True, exist_ok=True)
|
|
dest = error_dir / src.name
|
|
src.rename(dest)
|
|
return ProcessResult(src, outgoing_dir / f"OCR_{src.name}", False,
|
|
error="fake ocr failure")
|
|
|
|
|
|
def _run(tmp_config, fake_process):
|
|
"""Helper: führt run_once() mit gemocktem process_pdf und preflight aus."""
|
|
with patch("pdf_ocr_hotfolder.service.check_preflight", return_value=None), \
|
|
patch("pdf_ocr_hotfolder.service.process_pdf", side_effect=fake_process), \
|
|
patch("pdf_ocr_hotfolder.service._wait_until_stable", return_value=True):
|
|
service = HotfolderService(tmp_config)
|
|
try:
|
|
return service.run_once()
|
|
finally:
|
|
service._executor.shutdown(wait=False)
|
|
|
|
|
|
def test_once_exit_0_when_no_files(tmp_config) -> None:
|
|
"""Szenario: Keine PDFs vorhanden → Exit 0."""
|
|
errors = _run(tmp_config, _fake_success)
|
|
assert errors == 0
|
|
|
|
|
|
def test_once_exit_0_when_all_success(tmp_config) -> None:
|
|
"""Szenario: Alle PDFs erfolgreich → Exit 0."""
|
|
(tmp_config.paths.incoming / "a.pdf").write_bytes(b"%PDF-1.4\n")
|
|
(tmp_config.paths.incoming / "b.pdf").write_bytes(b"%PDF-1.4\n")
|
|
|
|
errors = _run(tmp_config, _fake_success)
|
|
assert errors == 0
|
|
|
|
|
|
def test_once_exit_nonzero_when_all_fail(tmp_config) -> None:
|
|
"""Szenario: Alle PDFs fehlgeschlagen → Exit != 0 (Issue #2)."""
|
|
(tmp_config.paths.incoming / "a.pdf").write_bytes(b"%PDF-1.4\n")
|
|
(tmp_config.paths.incoming / "b.pdf").write_bytes(b"%PDF-1.4\n")
|
|
|
|
errors = _run(tmp_config, _fake_failure)
|
|
assert errors == 2
|
|
|
|
|
|
def test_once_exit_nonzero_when_some_fail(tmp_config) -> None:
|
|
"""Szenario: Teilweise fehlgeschlagen → Exit != 0."""
|
|
(tmp_config.paths.incoming / "ok.pdf").write_bytes(b"%PDF-1.4\n")
|
|
(tmp_config.paths.incoming / "bad.pdf").write_bytes(b"%PDF-1.4\n")
|
|
|
|
def mixed(src, *args, **kwargs):
|
|
if "bad" in src.name:
|
|
return _fake_failure(src, *args, **kwargs)
|
|
return _fake_success(src, *args, **kwargs)
|
|
|
|
errors = _run(tmp_config, mixed)
|
|
assert errors == 1
|
|
|
|
|
|
def test_counters_track_success_and_failure(tmp_config) -> None:
|
|
"""success_count und error_count sollen korrekt mitzählen."""
|
|
(tmp_config.paths.incoming / "ok.pdf").write_bytes(b"%PDF-1.4\n")
|
|
(tmp_config.paths.incoming / "bad.pdf").write_bytes(b"%PDF-1.4\n")
|
|
|
|
def mixed(src, *args, **kwargs):
|
|
if "bad" in src.name:
|
|
return _fake_failure(src, *args, **kwargs)
|
|
return _fake_success(src, *args, **kwargs)
|
|
|
|
with patch("pdf_ocr_hotfolder.service.check_preflight", return_value=None), \
|
|
patch("pdf_ocr_hotfolder.service.process_pdf", side_effect=mixed), \
|
|
patch("pdf_ocr_hotfolder.service._wait_until_stable", return_value=True):
|
|
service = HotfolderService(tmp_config)
|
|
try:
|
|
service.run_once()
|
|
assert service.success_count == 1
|
|
assert service.error_count == 1
|
|
finally:
|
|
service._executor.shutdown(wait=False)
|