govoplan-xrechnung/tests/test_validation.py

from __future__ import annotations

from datetime import UTC, datetime
import hashlib
from pathlib import Path

import pytest

from govoplan_xrechnung.backend.manifest import get_manifest
from govoplan_xrechnung.backend.validation import (
    InboundInvoice,
    KoSITValidationProfile,
    XRechnungValidationError,
    configuration_tree_sha256,
    create_validated_handoff,
    interpret_kosit_result,
    verify_profile,
)


def _profile(tmp_path: Path, *, minimum_steps: int = 2) -> KoSITValidationProfile:
    java = tmp_path / "java"
    java.write_bytes(b"#!/bin/sh\n")
    java.chmod(0o700)
    jar = tmp_path / "validator.jar"
    jar.write_bytes(b"pinned validator")
    config = tmp_path / "config"
    config.mkdir()
    scenarios = config / "scenarios.xml"
    scenarios.write_text("<scenarios/>", encoding="utf-8")
    resources = config / "resources"
    resources.mkdir()
    (resources / "rules.xsl").write_text("<stylesheet/>", encoding="utf-8")
    return KoSITValidationProfile(
        profile_id="xrechnung-explicit-test",
        xrechnung_version="explicit-test-only",
        validator_version="validator-test",
        configuration_version="configuration-test",
        java_executable=java.resolve(),
        validator_jar=jar.resolve(),
        validator_jar_sha256=hashlib.sha256(jar.read_bytes()).hexdigest(),
        configuration_root=config.resolve(),
        configuration_tree_sha256=configuration_tree_sha256(config),
        scenarios_file=scenarios.resolve(),
        minimum_validation_steps=minimum_steps,
    )


def _report(*, valid: bool, step_count: int = 2) -> bytes:
    assessment = "accept" if valid else "reject"
    flag = "true" if valid else "false"
    steps = "".join(
        f'<rep:validationStepResult id="step-{index}" valid="{flag}" />'
        for index in range(step_count)
    )
    return (
        f'<rep:report xmlns:rep="http://www.xoev.de/de/validator/varl/1" valid="{flag}" varlVersion="1.0.0">'
        f"<rep:scenarioMatched>{steps}</rep:scenarioMatched>"
        f"<rep:assessment><rep:{assessment}/></rep:assessment>"
        "</rep:report>"
    ).encode()


def _invoice() -> InboundInvoice:
    return InboundInvoice(
        tenant_id="tenant-a",
        source_reference="mail:message-1:attachment-1",
        document=b"<Invoice/>",
        received_at=datetime(2026, 8, 23, tzinfo=UTC),
    )


def test_profile_verifies_exact_engine_and_complete_configuration_tree(tmp_path: Path) -> None:
    profile = _profile(tmp_path)
    first = verify_profile(profile)
    assert len(first) == 64

    (profile.configuration_root / "resources" / "rules.xsl").write_text("changed", encoding="utf-8")
    with pytest.raises(XRechnungValidationError, match="tree digest"):
        verify_profile(profile)


def test_complete_valid_report_can_create_digest_bound_handoff(tmp_path: Path) -> None:
    profile = _profile(tmp_path)
    invoice = _invoice()
    result = interpret_kosit_result(
        profile=profile,
        profile_sha256=verify_profile(profile),
        document_sha256=invoice.document_sha256,
        return_code=0,
        runner_output=b"INFO validation completed",
        report=_report(valid=True),
    )

    handoff = create_validated_handoff(invoice, result)

    assert result.technical_outcome == "complete"
    assert result.conformance == "valid"
    assert result.assessment == "accept"
    assert handoff.document_sha256 == invoice.document_sha256
    assert len(handoff.handoff_sha256) == 64


def test_semantically_invalid_report_is_complete_but_cannot_handoff(tmp_path: Path) -> None:
    profile = _profile(tmp_path)
    invoice = _invoice()
    result = interpret_kosit_result(
        profile=profile,
        profile_sha256=verify_profile(profile),
        document_sha256=invoice.document_sha256,
        return_code=0,
        runner_output=b"INFO validation completed",
        report=_report(valid=False),
    )

    assert result.technical_outcome == "complete"
    assert result.conformance == "invalid"
    assert result.assessment == "reject"
    with pytest.raises(XRechnungValidationError, match="technically complete"):
        create_validated_handoff(invoice, result)


@pytest.mark.parametrize(
    ("return_code", "output", "report", "expected"),
    [
        (1, b"", _report(valid=True), "failed"),
        (0, b"ERROR Transformation failed", _report(valid=True), "incomplete"),
        (0, b"", _report(valid=True, step_count=1), "incomplete"),
        (0, b"", None, "incomplete"),
    ],
)
def test_technical_failures_never_trust_a_valid_looking_report(
    tmp_path: Path,
    return_code: int,
    output: bytes,
    report: bytes | None,
    expected: str,
) -> None:
    profile = _profile(tmp_path)
    result = interpret_kosit_result(
        profile=profile,
        profile_sha256=verify_profile(profile),
        document_sha256="a" * 64,
        return_code=return_code,
        runner_output=output,
        report=report,
    )

    assert result.technical_outcome == expected
    assert result.conformance == "unknown"
    assert result.assessment == "unknown"
    assert result.handoff_allowed is False


def test_manifest_does_not_select_an_active_standard_version() -> None:
    manifest = get_manifest()
    assert manifest.version == "0.1.20"
    assert "none is activated by default" in manifest.architecture.known_limits[0].lower()