orc-renaming/tests/test_ollama.py

import json
from datetime import date

import httpx

from orc_renaming.models import DocumentScope, DocumentType, ExtractionResult
from orc_renaming.ollama import OllamaAnalyzer, merge_results


def test_stammdaten_override_llm() -> None:
    rules = ExtractionResult(
        document_type=DocumentType.INVOICE,
        scope=DocumentScope.PROPERTY,
        document_date=date(2026, 7, 18),
        company="Kanonische-Firma",
        property_id="WH1",
        confidence=0.93,
        evidence=["Stammdatentreffer"],
    )
    llm = ExtractionResult(
        document_type=DocumentType.CORRESPONDENCE,
        scope=DocumentScope.PRIVATE,
        document_date=date(2026, 7, 17),
        company="Halluzinierte Firma",
        topic="Rechnung",
        confidence=0.99,
        evidence=["Modelltreffer"],
        source="ollama",
        model="qwen3.5:4B",
    )

    result = merge_results(rules, llm)
    assert result.document_type == DocumentType.INVOICE
    assert result.scope == DocumentScope.PROPERTY
    assert result.document_date == date(2026, 7, 18)
    assert result.company == "Kanonische-Firma"
    assert result.property_id == "WH1"
    assert result.source == "rules+ollama:qwen3.5:4B"


def test_ollama_structured_response(config) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/api/chat"
        body = json.loads(request.read())
        assert isinstance(body["format"], dict)
        assert body["options"]["temperature"] == 0
        assert body["options"]["presence_penalty"] == 0
        assert body["think"] is False
        assert body["options"]["num_ctx"] == 8192
        assert body["messages"][1]["images"] == ["aW1hZ2UtYnl0ZXM="]
        assert '"document_type"' in body["messages"][1]["content"]
        assert "Verbindliches JSON-Schema" in body["messages"][1]["content"]
        return httpx.Response(
            200,
            json={
                "message": {
                    "content": (
                        '{"document_type":"invoice","scope":"property",'
                        '"document_date":"2026-07-18","company":"Firma",'
                        '"property_id":"WH1","topic":null,"confidence":0.94,'
                        '"evidence":["Rechnung"]}'
                    )
                }
            },
        )

    analyzer = OllamaAnalyzer(config)
    analyzer.client.close()
    analyzer.client = httpx.Client(
        base_url="http://ollama.test",
        transport=httpx.MockTransport(handler),
    )
    result = analyzer.analyze(
        "Rechnung für Musterstraße 12",
        image_bytes=b"image-bytes",
    )
    analyzer.close()

    assert result.document_type == DocumentType.INVOICE
    assert result.property_id == "WH1"
    assert result.document_date == date(2026, 7, 18)
    assert result.model == "qwen3.5:4B"


def test_incomplete_ollama_json_is_kept_as_low_confidence(config) -> None:
    config.ollama.json_repair_attempts = 0

    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(
            200,
            json={
                "message": {
                    "content": '{"company":"Firma GmbH","confidence":0.94}'
                }
            },
        )

    analyzer = OllamaAnalyzer(config)
    analyzer.client.close()
    analyzer.client = httpx.Client(
        base_url="http://ollama.test",
        transport=httpx.MockTransport(handler),
    )
    result = analyzer.analyze("Brief")
    analyzer.close()

    assert result.company == "Firma GmbH"
    assert result.document_type == DocumentType.UNKNOWN
    assert result.confidence == 0.65
    assert any("unvollständig" in warning for warning in result.warnings)


def test_free_text_response_is_repaired_without_resending_image(config) -> None:
    calls: list[dict] = []

    def handler(request: httpx.Request) -> httpx.Response:
        body = json.loads(request.read())
        calls.append(body)
        if len(calls) == 1:
            return httpx.Response(
                200,
                json={
                    "message": {
                        "content": (
                            "Hier ist eine strukturierte Zusammenfassung: "
                            "Rechnung der Firma GmbH vom 18.07.2026 für WH1."
                        )
                    }
                },
            )
        return httpx.Response(
            200,
            json={
                "message": {
                    "content": (
                        '{"document_type":"invoice","scope":"property",'
                        '"document_date":"2026-07-18","company":"Firma GmbH",'
                        '"property_id":"WH1","topic":"","confidence":0.94,'
                        '"evidence":["Rechnung für WH1"]}'
                    )
                }
            },
        )

    analyzer = OllamaAnalyzer(config)
    analyzer.client.close()
    analyzer.client = httpx.Client(
        base_url="http://ollama.test",
        transport=httpx.MockTransport(handler),
    )
    result = analyzer.analyze(
        "Rechnung für Musterstraße 12",
        model="qwen3.5:9b",
        image_bytes=b"image-bytes",
    )
    analyzer.close()

    assert len(calls) == 2
    assert calls[0]["model"] == "qwen3.5:9b"
    assert calls[1]["model"] == "qwen3.5:4B"
    assert calls[0]["messages"][1]["images"] == ["aW1hZ2UtYnl0ZXM="]
    assert "images" not in calls[1]["messages"][1]
    assert "Hier ist eine strukturierte Zusammenfassung" in (
        calls[1]["messages"][1]["content"]
    )
    assert result.document_type == DocumentType.INVOICE
    assert result.company == "Firma GmbH"
    assert result.property_id == "WH1"
    assert result.confidence == 0.85
    assert result.model == "qwen3.5:9b+json-repair:qwen3.5:4B"
    assert any("normalisiert" in warning for warning in result.warnings)


def test_wrong_json_schema_triggers_repair(config) -> None:
    responses = iter(
        [
            {
                "message": {
                    "content": (
                        '{"document_type":"inspection_report",'
                        '"title":"Überprüfungsergebnis","date":"2025-05-14"}'
                    )
                }
            },
            {
                "message": {
                    "content": (
                        '{"document_type":"correspondence","scope":"private",'
                        '"document_date":"2025-05-14","company":"Kaminkehrer",'
                        '"property_id":"","topic":"Überprüfungsergebnis",'
                        '"confidence":0.88,"evidence":["KÜO"]}'
                    )
                }
            },
        ]
    )

    def handler(_request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=next(responses))

    analyzer = OllamaAnalyzer(config)
    analyzer.client.close()
    analyzer.client = httpx.Client(
        base_url="http://ollama.test",
        transport=httpx.MockTransport(handler),
    )
    result = analyzer.analyze("Überprüfungsergebnis gemäß KÜO")
    analyzer.close()

    assert result.document_type == DocumentType.CORRESPONDENCE
    assert result.topic == "Überprüfungsergebnis"
    assert result.confidence == 0.85