pdf-ocr-hotfolder/tests/test_outgoing_collision.py

"""Namens-Kollision in outgoing/ darf kein Ergebnis mehr überschreiben.

`process_pdf()` beendete mit `shutil.move(work_out, final_out)`. Lag dort
bereits eine Datei desselben Namens (Scanner liefert denselben Dateinamen ein
zweites Mal, oder das Vorgängerergebnis wurde noch nicht abgeholt), war das
ältere Ergebnis kommentarlos weg. Jetzt gilt derselbe Zeitstempel-Ausweg wie
im Archiv.
"""
from __future__ import annotations

import logging
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 _collision_free_path, process_pdf

OLD = b"%PDF-1.4 altes ergebnis\n"
ORIGINAL = b"%PDF-1.4 original\n"


def _fake_ocr(src: Path, dst: Path, cfg: OcrConfig) -> None:
    dst.write_bytes(b"%PDF-1.4 OCRed\n" + src.read_bytes())


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(env: dict, out_cfg: OutputConfig):
    with patch("pdf_ocr_hotfolder.processor.run_ocr", side_effect=_fake_ocr):
        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=False),
            output_cfg=out_cfg,
        )


# ---------------- Kollision in outgoing/ ----------------

def test_existing_result_is_not_overwritten(tmp_path: Path) -> None:
    """Beide Dateien müssen hinterher existieren."""
    env = _prepare(tmp_path)
    (env["outgoing"] / "OCR_scan.pdf").write_bytes(OLD)

    result = _run(env, OutputConfig(name_mode="prefix", name_tag="OCR_",
                                    original_on_success="delete"))

    assert result.success
    # Altes Ergebnis unverändert
    assert (env["outgoing"] / "OCR_scan.pdf").read_bytes() == OLD
    # Neues Ergebnis unter Zeitstempel-Namen daneben
    neu = [p for p in env["outgoing"].glob("OCR_scan_*.pdf")]
    assert len(neu) == 1
    assert neu[0].read_bytes() == b"%PDF-1.4 OCRed\n" + ORIGINAL
    assert len(list(env["outgoing"].iterdir())) == 2


def test_result_output_points_to_written_file(tmp_path: Path) -> None:
    """ProcessResult.output muss den TATSÄCHLICH geschriebenen Pfad tragen.

    Sonst melden Uploads und die E-Mail-Benachrichtigung die falsche (nämlich
    die fremde, ältere) Datei.
    """
    env = _prepare(tmp_path)
    (env["outgoing"] / "OCR_scan.pdf").write_bytes(OLD)

    result = _run(env, OutputConfig(name_mode="prefix", name_tag="OCR_",
                                    original_on_success="delete"))

    assert result.output.exists()
    assert result.output.name != "OCR_scan.pdf"
    assert result.output.parent == env["outgoing"]
    assert result.output.read_bytes() != OLD


def test_collision_logs_warning_with_both_names(tmp_path: Path, caplog) -> None:
    env = _prepare(tmp_path)
    (env["outgoing"] / "OCR_scan.pdf").write_bytes(OLD)

    with caplog.at_level(logging.WARNING, logger="pdf_ocr_hotfolder.processor"):
        result = _run(env, OutputConfig(name_mode="prefix", name_tag="OCR_",
                                        original_on_success="delete"))

    text = caplog.text
    assert "OCR_scan.pdf" in text
    assert result.output.name in text
    assert "überschrieben" in text


def test_no_collision_keeps_plain_name(tmp_path: Path, caplog) -> None:
    """Ohne Kollision bleibt alles wie bisher — kein Suffix, keine Warnung."""
    env = _prepare(tmp_path)

    with caplog.at_level(logging.WARNING, logger="pdf_ocr_hotfolder.processor"):
        result = _run(env, OutputConfig(name_mode="prefix", name_tag="OCR_",
                                        original_on_success="delete"))

    assert result.output == env["outgoing"] / "OCR_scan.pdf"
    assert result.output.exists()
    assert "überschrieben" not in caplog.text


def test_collision_with_name_mode_none(tmp_path: Path) -> None:
    """name_mode='none': Ergebnis heißt wie das Original — Kollision ist dort
    der Normalfall, nicht die Ausnahme."""
    env = _prepare(tmp_path)
    (env["outgoing"] / "scan.pdf").write_bytes(OLD)

    result = _run(env, OutputConfig(name_mode="none", name_tag="",
                                    original_on_success="delete"))

    assert result.success
    assert (env["outgoing"] / "scan.pdf").read_bytes() == OLD
    assert result.output.name.startswith("scan_")
    assert result.output.suffix == ".pdf"


def test_original_is_still_disposed_after_collision(tmp_path: Path) -> None:
    """Der Ausweichname darf die Entsorgung des Originals nicht aushebeln."""
    env = _prepare(tmp_path)
    (env["outgoing"] / "OCR_scan.pdf").write_bytes(OLD)

    _run(env, OutputConfig(name_mode="prefix", name_tag="OCR_",
                           original_on_success="archive",
                           archive_dir=str(env["archive"])))

    assert list(env["working"].iterdir()) == []
    assert (env["archive"] / "scan.pdf").read_bytes() == ORIGINAL


# ---------------- _collision_free_path ----------------

def test_collision_free_path_passes_through_free_name(tmp_path: Path) -> None:
    dest = tmp_path / "frei.pdf"
    assert _collision_free_path(dest) == dest


def test_collision_free_path_appends_timestamp(tmp_path: Path) -> None:
    dest = tmp_path / "belegt.pdf"
    dest.write_bytes(b"x")
    out = _collision_free_path(dest)
    assert out != dest
    assert out.name.startswith("belegt_")
    assert out.suffix == ".pdf"
    assert not out.exists()


def test_collision_free_path_counts_up_within_same_second(tmp_path: Path) -> None:
    """Zwei Ergebnisse in derselben Sekunde (mehrere Worker) kollidieren sonst
    erneut — und der move überschriebe wieder still."""
    dest = tmp_path / "belegt.pdf"
    dest.write_bytes(b"x")

    first = _collision_free_path(dest)
    first.write_bytes(b"y")
    with patch("pdf_ocr_hotfolder.processor.datetime") as dt:
        # Zeitstempel einfrieren: erzwingt denselben Namen wie `first`
        dt.now.return_value.strftime.return_value = first.stem.split("_", 1)[1]
        second = _collision_free_path(dest)

    assert second != first
    assert not second.exists()
    assert second.suffix == ".pdf"


@pytest.mark.parametrize("name", ["ohne_extension", "zwei.punkte.pdf"])
def test_collision_free_path_keeps_extension(tmp_path: Path, name: str) -> None:
    dest = tmp_path / name
    dest.write_bytes(b"x")
    out = _collision_free_path(dest)
    assert out.suffix == dest.suffix
    assert out.name != dest.name