orc-renaming/tests/test_pipeline.py

from __future__ import annotations

from datetime import date
from pathlib import Path

import orc_renaming.pipeline as pipeline_module
from orc_renaming.models import (
    DocumentScope,
    DocumentType,
    ExtractionResult,
    ProcessingRecord,
)
from orc_renaming.normalize import result_is_complete
from orc_renaming.pipeline import Pipeline
from orc_renaming.webdav import RemoteFile


class FakeWebDAV:
    uploads: list[tuple[str, bool]]
    moves: list[tuple[str, str]]
    folders: list[str]

    def __init__(self, _config) -> None:
        self.uploads = []
        self.moves = []
        self.folders = []

    def close(self) -> None:
        pass

    def list_pdfs(self, _folder: str) -> list[RemoteFile]:
        return [
            RemoteFile(
                path="/Scanner/Eingang/scan001.pdf",
                name="scan001.pdf",
            )
        ]

    def download(self, _remote_path: str, local_path: Path) -> None:
        local_path.write_bytes(b"%PDF fake")

    def exists(self, _remote_path: str) -> bool:
        return False

    def ensure_folder(self, folder: str) -> None:
        self.folders.append(folder)

    def upload(
        self, local_path: Path, remote_path: str, overwrite: bool = False
    ) -> None:
        assert local_path.exists()
        self.uploads.append((remote_path, overwrite))

    def move(self, source: str, destination: str, overwrite: bool = False) -> None:
        assert not overwrite
        self.moves.append((source, destination))


OCR_TEXT = """
Stadtwerke Musterstadt GmbH
Rechnung
Rechnungsnummer 2026-123
Rechnungsdatum: 18.07.2026
Lieferstelle Musterstraße 12
Vertragskonto 47110815
Rechnungsbetrag 123,45 EUR
"""


def _prepare(monkeypatch) -> None:
    monkeypatch.setattr(pipeline_module, "WebDAVClient", FakeWebDAV)
    monkeypatch.setattr(pipeline_module, "extract_pdf_text", lambda *_args, **_kwargs: OCR_TEXT)
    monkeypatch.setattr(pipeline_module, "file_sha256", lambda _path: "abc123")


def test_dry_run_writes_only_local_report(config, monkeypatch) -> None:
    _prepare(monkeypatch)
    config.processing.dry_run = True

    with Pipeline(config) as pipeline:
        counters = pipeline.run()
        assert pipeline.webdav.uploads == [
            ("/Scanner/Eingang/umbenannt/protokoll.xlsx", True)
        ]
        assert pipeline.webdav.moves == []

    assert counters == {"dry-run": 1}
    assert (config.processing.state_directory / "protokoll.xlsx").exists()


def test_live_run_uploads_pdf_report_and_archives(config, monkeypatch) -> None:
    _prepare(monkeypatch)
    config.processing.dry_run = False

    with Pipeline(config) as pipeline:
        counters = pipeline.run()
        uploads = list(pipeline.webdav.uploads)
        moves = list(pipeline.webdav.moves)
        folders = list(pipeline.webdav.folders)

    assert counters == {"success": 1}
    assert (
        "/Scanner/Eingang/umbenannt/260718_RE_Stadtwerke-Musterstadt-WH1.pdf",
        False,
    ) in uploads
    assert (
        "/Scanner/Eingang/umbenannt/protokoll.xlsx",
        True,
    ) in uploads
    assert moves == [
        (
            "/Scanner/Eingang/scan001.pdf",
            "/Scanner/Eingang/verarbeitet-originale/scan001.pdf",
        )
    ]
    assert config.nextcloud.review_folder in folders


def test_empty_run_does_not_write_or_upload_report(config, monkeypatch) -> None:
    _prepare(monkeypatch)

    with Pipeline(config) as pipeline:
        pipeline.webdav.list_pdfs = lambda _folder: []
        counters = pipeline.run()
        uploads = list(pipeline.webdav.uploads)

    assert counters == {}
    assert uploads == []
    assert not (config.processing.state_directory / "protokoll.xlsx").exists()


def test_skipped_files_do_not_rewrite_report(config, monkeypatch) -> None:
    _prepare(monkeypatch)
    report_path = config.processing.state_directory / "protokoll.xlsx"

    with Pipeline(config) as pipeline:
        pipeline.ledger.add(
            ProcessingRecord(
                remote_path="/Scanner/Eingang/scan001.pdf",
                original_name="scan001.pdf",
                checksum="abc123",
                status="success",
            )
        )
        pipeline.ledger.export_xlsx(report_path)
        report_before = report_path.read_bytes()

        counters = pipeline.run()
        uploads = list(pipeline.webdav.uploads)
        report_after = report_path.read_bytes()

    assert counters == {"skipped": 1}
    assert uploads == []
    assert report_after == report_before


def test_ollama_fallback_is_only_used_after_incomplete_primary(
    config, monkeypatch
) -> None:
    _prepare(monkeypatch)
    config.ollama.enabled = True
    config.ollama.model = "fast:4b"
    config.ollama.fallback_model = "accurate:9b"

    class FakeOllama:
        def __init__(self, _config) -> None:
            self.calls: list[str] = []

        def close(self) -> None:
            pass

        def analyze(
            self,
            _text: str,
            model: str,
            image_bytes: bytes | None = None,
        ) -> ExtractionResult:
            assert image_bytes is None
            self.calls.append(model)
            if model == "fast:4b":
                return ExtractionResult(
                    document_type=DocumentType.CORRESPONDENCE,
                    document_date=date(2026, 7, 18),
                    company="Firma",
                    confidence=0.90,
                    model=model,
                    source="ollama",
                )
            return ExtractionResult(
                document_type=DocumentType.CORRESPONDENCE,
                scope=DocumentScope.PRIVATE,
                document_date=date(2026, 7, 18),
                company="Firma",
                topic="Vertragsänderung",
                confidence=0.92,
                model=model,
                source="ollama",
            )

    monkeypatch.setattr(pipeline_module, "OllamaAnalyzer", FakeOllama)
    with Pipeline(config) as pipeline:
        result = pipeline.analyze_text("Nicht regelbasiert erkennbarer Brieftext")
        calls = list(pipeline.ollama.calls)

    assert calls == ["fast:4b", "accurate:9b"]
    assert result_is_complete(result)
    assert result.model == "fast:4b -> accurate:9b"