leocrm/tests/test_einvoice_generator.py

"""E-Invoice (EN16931 / XRechnung CII) format layer + AI block suggestions — Phase L4/L5.

L5 (Format-Layer): inline Rechnungsdaten → valides CII-XML. Das
Verkaufsmodul kommt später und dockt über den Contract-Hook
``einvoice_data()`` an (POST /einvoice/render-for).

L4: KI-Vorschlag für Block-Kompositionen. llm_complete wird in allen
Tests gemockt — niemals echte LLM-Calls (Budget, Deterministik).
"""

from __future__ import annotations

import json
import uuid as uuid_mod
from collections.abc import AsyncGenerator
from unittest.mock import AsyncMock, MagicMock, patch
import xml.etree.ElementTree as ET

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, AsyncSession

from app.core.db import close_engine, reset_engine_for_testing
from app.core.permission_registry import init_permission_registry
from app.core.service_container import get_container
from app.main import create_app
from app.plugins.builtins.permissions import PermissionsPlugin
from app.plugins.builtins.report_generator import ReportGeneratorPlugin
from app.plugins.registry import reset_registry_for_testing
from app.services.plugin_service import reset_plugin_service_for_testing

from tests.conftest import (
    ORIGIN_HEADER,
    login_client,
    seed_tenant_and_users,
)

DOC_BASE = "/api/v1/reports"


@pytest_asyncio.fixture
async def docs_app(engine: AsyncEngine, redis_client):
    """App with permissions + report_generator installed & activated."""
    reset_engine_for_testing(engine)
    app = create_app()

    registry = reset_registry_for_testing()
    registry.initialize(engine, app)
    init_permission_registry(active_plugin_names={"permissions", "report_generator"})

    container = get_container()
    await container.initialize()

    registry.register_plugin(PermissionsPlugin())
    registry.register_plugin(ReportGeneratorPlugin())
    reset_plugin_service_for_testing(registry)

    _sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
    async with _sf() as session:
        await registry.install(session, "permissions")
        await registry.activate(session, "permissions")
        await registry.install(session, "report_generator")
        await registry.activate(session, "report_generator")
        await session.commit()

    yield app
    await close_engine()


@pytest_asyncio.fixture
async def docs_client(docs_app) -> AsyncGenerator[AsyncClient, None]:
    transport = ASGITransport(app=docs_app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


def _invoice_payload(**overrides) -> dict:
    """Valid EN16931 invoice payload (2 × 100.00 net, 19% VAT)."""
    payload = {
        "invoice_number": "RE-2026-001",
        "issue_date": "2026-08-29",
        "type_code": "380",
        "currency": "EUR",
        "due_date": "2026-09-28",
        "delivery_date": "2026-08-29",
        "buyer_name": "Käufer AG",
        "buyer_reference": "PO-2026-42",
        "buyer_address": {
            "street": "Käuferstr. 1",
            "postal_code": "20095",
            "city": "Hamburg",
            "country": "DE",
        },
        "seller_name": "Verkäufer GmbH",
        "seller_vat_id": "DE123456789",
        "seller_address": {
            "street": "Verkäuferweg 2",
            "postal_code": "10115",
            "city": "Berlin",
            "country": "DE",
        },
        "payment_means_code": "58",
        "payment_terms_text": "Zahlbar innerhalb von 30 Tagen",
        "note": "Vielen Dank für Ihren Auftrag",
        "line_items": [
            {
                "name": "Beratung",
                "quantity": 2,
                "unit": "HUR",
                "unit_net_price": 100.00,
                "vat_rate": 19,
            },
        ],
        "profile": "en16931",
    }
    payload.update(overrides)
    return payload


# ─── Unit: validation ────────────────────────────────────────────────────────


class TestEInvoiceUnitValidation:
    def test_validate_ok(self):
        from app.plugins.builtins.report_generator.einvoice import validate_einvoice_data

        validate_einvoice_data(_invoice_payload())  # no raise

    def test_missing_seller_tax_ids_raises_with_bt(self):
        from app.plugins.builtins.report_generator.einvoice import (
            EInvoiceValidationError,
            validate_einvoice_data,
        )

        payload = _invoice_payload()
        del payload["seller_vat_id"]
        with pytest.raises(EInvoiceValidationError) as exc_info:
            validate_einvoice_data(payload)
        assert any("BT-31" in m or "BT-32" in m for m in exc_info.value.missing)

    def test_missing_buyer_name_raises(self):
        from app.plugins.builtins.report_generator.einvoice import (
            EInvoiceValidationError,
            validate_einvoice_data,
        )

        payload = _invoice_payload(buyer_name="")
        with pytest.raises(EInvoiceValidationError):
            validate_einvoice_data(payload)

    def test_missing_lines_raises(self):
        from app.plugins.builtins.report_generator.einvoice import (
            EInvoiceValidationError,
            validate_einvoice_data,
        )

        with pytest.raises(EInvoiceValidationError):
            validate_einvoice_data(_invoice_payload(line_items=[]))

    def test_bad_issue_date_raises(self):
        from app.plugins.builtins.report_generator.einvoice import (
            EInvoiceValidationError,
            validate_einvoice_data,
        )

        with pytest.raises(EInvoiceValidationError):
            validate_einvoice_data(_invoice_payload(issue_date="29.08.2026"))

    def test_line_item_requires_name(self):
        from app.plugins.builtins.report_generator.einvoice import (
            EInvoiceValidationError,
            validate_einvoice_data,
        )

        payload = _invoice_payload(
            line_items=[{"name": "", "quantity": 1, "unit_net_price": 10, "vat_rate": 19}]
        )
        with pytest.raises(EInvoiceValidationError):
            validate_einvoice_data(payload)


# ─── Unit: XML rendering ─────────────────────────────────────────────────────


class TestEInvoiceUnitRender:
    def test_render_xml_structure(self):
        from app.plugins.builtins.report_generator.einvoice import render_einvoice_xml

        xml = render_einvoice_xml(_invoice_payload())

        # well-formed XML with correct root
        root = ET.fromstring(xml)
        assert root.tag.endswith("CrossIndustryInvoice")

        assert "urn:cen.eu:en16931:2017" in xml  # EN16931 guideline
        assert "RE-2026-001" in xml  # BT-1
        assert "20260829" in xml  # BT-2 in format 102
        assert "380" in xml  # BT-3
        assert "Verkäufer GmbH" in xml  # BT-11
        assert "Käufer AG" in xml  # BT-10
        assert "DE123456789" in xml  # BT-31
        assert 'schemeID="VA"' in xml  # VAT registration scheme
        # computed sums: 200.00 net + 38.00 VAT = 238.00 grand
        assert "200.00" in xml
        assert "38.00" in xml
        assert "238.00" in xml
        assert "EUR" in xml  # BT-5

    def test_render_xrechnung_profile(self):
        from app.plugins.builtins.report_generator.einvoice import render_einvoice_xml

        xml = render_einvoice_xml(_invoice_payload(profile="xrechnung"))
        assert "urn:xoev-de:kosit:standard:xrechnung_3.0" in xml
        assert "urn:cen.eu:en16931:2017" in xml  # compliant base

    def test_render_escapes_xml_special_chars(self):
        from app.plugins.builtins.report_generator.einvoice import render_einvoice_xml

        payload = _invoice_payload(seller_name="A&B <GmbH>")
        xml = render_einvoice_xml(payload)
        assert "A&amp;B &lt;GmbH&gt;" in xml
        # raw unescaped angle bracket must not appear in the name element
        assert "<GmbH>" not in xml

    def test_render_line_net_amount_supplied_used(self):
        from app.plugins.builtins.report_generator.einvoice import render_einvoice_xml

        payload = _invoice_payload(
            line_items=[
                {
                    "name": "Pauschale",
                    "quantity": 1,
                    "unit_net_price": 99.99,
                    "vat_rate": 7,
                    "line_net_amount": 90.00,
                }
            ]
        )
        xml = render_einvoice_xml(payload)
        assert "90.00" in xml  # supplied amount wins over qty*price
        # tax: 90 * 7% = 6.30, grand 96.30
        assert "6.30" in xml
        assert "96.30" in xml

    def test_render_delivery_date_defaults_to_issue_date(self):
        from app.plugins.builtins.report_generator.einvoice import render_einvoice_xml

        payload = _invoice_payload()
        del payload["delivery_date"]
        xml = render_einvoice_xml(payload)
        assert "20260829" in xml


# ─── Unit: contract resolution (sales module docking point) ─────────────────


class TestEInvoiceContractResolution:
    async def test_resolve_via_contract_hook(self, db_session):
        """The future sales module provides einvoice_data() — the resolver
        picks it up through the contract registry."""
        seed = await seed_tenant_and_users(db_session)
        from app.plugins.builtins.report_generator import einvoice

        invoice_data = _invoice_payload()
        fake_contract = MagicMock()
        fake_contract.einvoice_data = AsyncMock(return_value=invoice_data)

        fake_plugin_registry = MagicMock()
        fake_plugin_registry.list_discovered.return_value = ["sales"]
        fake_contract_registry = MagicMock()
        fake_contract_registry.get_contract.return_value = fake_contract

        with (
            patch("app.plugins.registry.get_registry", return_value=fake_plugin_registry),
            patch(
                "app.plugins.builtins.contracts.get_contract_registry",
                return_value=fake_contract_registry,
            ),
        ):
            data = await einvoice.resolve_einvoice_data(
                db_session,
                seed["tenant_a"].id,
                "invoice",
                uuid_mod.uuid4(),
            )
        assert data is not None
        assert data["invoice_number"] == "RE-2026-001"

    async def test_resolve_without_hook_returns_none(self, db_session):
        """No module contributes einvoice_data → None (→ 404 for caller)."""
        seed = await seed_tenant_and_users(db_session)
        from app.plugins.builtins.report_generator import einvoice

        class NoHookContract:
            contract_name = "legacy"

        fake_plugin_registry = MagicMock()
        fake_plugin_registry.list_discovered.return_value = ["legacy"]
        fake_contract_registry = MagicMock()
        fake_contract_registry.get_contract.return_value = NoHookContract()

        with (
            patch("app.plugins.registry.get_registry", return_value=fake_plugin_registry),
            patch(
                "app.plugins.builtins.contracts.get_contract_registry",
                return_value=fake_contract_registry,
            ),
        ):
            data = await einvoice.resolve_einvoice_data(
                db_session,
                seed["tenant_a"].id,
                "invoice",
                uuid_mod.uuid4(),
            )
        assert data is None


# ─── API: e-invoice endpoints ────────────────────────────────────────────────


@pytest.mark.asyncio
class TestEInvoiceAPI:
    async def test_render_returns_xml_200(
        self, docs_client: AsyncClient, db_session
    ):
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        resp = await docs_client.post(
            f"{DOC_BASE}/einvoice/render",
            json=_invoice_payload(),
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 200, f"{resp.status_code} {resp.text}"
        assert resp.headers["content-type"].startswith("application/xml")
        assert b"CrossIndustryInvoice" in resp.content
        assert b"RE-2026-001" in resp.content
        assert "attachment" in resp.headers.get("content-disposition", "")

    async def test_render_missing_fields_422(
        self, docs_client: AsyncClient, db_session
    ):
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        payload = _invoice_payload()
        del payload["seller_vat_id"]
        del payload["buyer_name"]
        resp = await docs_client.post(
            f"{DOC_BASE}/einvoice/render",
            json=payload,
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 422
        body = resp.json()
        assert body["detail"]["code"] == "invalid_einvoice"
        missing = json.dumps(body["detail"].get("missing", []))
        assert "BT-10" in missing  # buyer_name
        assert "BT-31" in missing  # vat/tax requirement

    async def test_validate_endpoint_ok(
        self, docs_client: AsyncClient, db_session
    ):
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        resp = await docs_client.post(
            f"{DOC_BASE}/einvoice/validate",
            json=_invoice_payload(),
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 200
        assert resp.json()["valid"] is True

    async def test_validate_endpoint_reports_missing(
        self, docs_client: AsyncClient, db_session
    ):
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        payload = _invoice_payload()
        del payload["seller_vat_id"]
        resp = await docs_client.post(
            f"{DOC_BASE}/einvoice/validate",
            json=payload,
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 422
        assert resp.json()["detail"]["code"] == "invalid_einvoice"

    async def test_render_for_without_sales_module_404(
        self, docs_client: AsyncClient, db_session
    ):
        """render-for needs a module einvoice_data() — none exists yet → 404."""
        seed = await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        resp = await docs_client.post(
            f"{DOC_BASE}/einvoice/render-for",
            json={
                "entity_type": "invoice",
                "entity_id": str(seed["company_a"].id),
            },
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 404
        assert resp.json()["detail"]["code"] == "no_data_source"

    async def test_viewer_cannot_render_403(
        self, docs_client: AsyncClient, db_session
    ):
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "viewer@tenanta.com")
        resp = await docs_client.post(
            f"{DOC_BASE}/einvoice/render",
            json=_invoice_payload(),
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 403


# ─── API: AI block suggestion (L4) ────────────────────────────────────────────


_SUGGEST_VALID_BLOCKS = {
    "blocks": [
        {
            "id": "b1",
            "type": "text",
            "config": {
                "content": "Sehr geehrte Damen und Herren,\n\nhiermit offerieren wir {{name}} folgende Leistungen:"
            },
        },
        {
            "id": "b2",
            "type": "shape",
            "config": {"shape": "line", "width": "100%", "height": 2, "color": "#111827"},
        },
    ],
    "notes": "Anrede mit Firmenname + Trennlinie",
}


@pytest.mark.asyncio
class TestDocumentSuggest:
    async def test_suggest_returns_validated_blocks(
        self, docs_client: AsyncClient, db_session
    ):
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        with patch(
            "app.plugins.builtins.report_generator.documents.llm_complete",
            AsyncMock(return_value={"content": json.dumps(_SUGGEST_VALID_BLOCKS)}),
        ) as mock_llm:
            resp = await docs_client.post(
                f"{DOC_BASE}/documents/suggest",
                json={
                    "prompt": "Erstelle eine Angebotsvorlage mit Anrede und Trennlinie",
                    "entity_type": "company",
                },
                headers=ORIGIN_HEADER,
            )
        assert resp.status_code == 200, f"{resp.status_code} {resp.text}"
        body = resp.json()
        assert len(body["blocks"]) == 2
        assert body["blocks"][0]["type"] == "text"
        assert "{{name}}" in body["blocks"][0]["config"]["content"]
        assert body["blocks"][1]["type"] == "shape"
        assert body["notes"] != ""
        # LLM call wired through the central client (cost tracking params)
        assert mock_llm.await_count == 1
        kwargs = mock_llm.await_args.kwargs
        assert kwargs["model"] == "gpt-4o-mini"
        assert kwargs["messages"][0]["role"] == "system"

    async def test_suggest_filters_invalid_blocks(
        self, docs_client: AsyncClient, db_session
    ):
        """Invalid AI blocks (unknown type, empty text) are filtered, valid
        ones survive even without an id (id gets assigned server-side)."""
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        ai_response = {
            "blocks": [
                {"type": "text", "config": {"content": "gültiger Block"}},
                {"type": "no_such_block_type", "config": {}},
                {"type": "text", "config": {}},  # empty content → invalid
            ],
            "notes": "",
        }
        with patch(
            "app.plugins.builtins.report_generator.documents.llm_complete",
            AsyncMock(return_value={"content": json.dumps(ai_response)}),
        ):
            resp = await docs_client.post(
                f"{DOC_BASE}/documents/suggest",
                json={"prompt": "Angebot mit Betreffzeile"},
                headers=ORIGIN_HEADER,
            )
        assert resp.status_code == 200
        body = resp.json()
        assert len(body["blocks"]) == 1
        assert body["blocks"][0]["type"] == "text"
        assert body["blocks"][0]["id"]  # server-assigned id

    async def test_suggest_handles_code_fenced_json(
        self, docs_client: AsyncClient, db_session
    ):
        """LLMs love ```json fences — the endpoint must strip them."""
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        fenced = "```json\n" + json.dumps(_SUGGEST_VALID_BLOCKS) + "\n```"
        with patch(
            "app.plugins.builtins.report_generator.documents.llm_complete",
            AsyncMock(return_value={"content": fenced}),
        ):
            resp = await docs_client.post(
                f"{DOC_BASE}/documents/suggest",
                json={"prompt": "Angebot erstellen"},
                headers=ORIGIN_HEADER,
            )
        assert resp.status_code == 200
        assert len(resp.json()["blocks"]) == 2

    async def test_suggest_llm_failure_502(
        self, docs_client: AsyncClient, db_session
    ):
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        with patch(
            "app.plugins.builtins.report_generator.documents.llm_complete",
            AsyncMock(side_effect=RuntimeError("provider down")),
        ):
            resp = await docs_client.post(
                f"{DOC_BASE}/documents/suggest",
                json={"prompt": "Angebot erstellen"},
                headers=ORIGIN_HEADER,
            )
        assert resp.status_code == 502
        assert resp.json()["detail"]["code"] == "ai_unavailable"

    async def test_suggest_non_json_response_502(
        self, docs_client: AsyncClient, db_session
    ):
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "admin@tenanta.com")
        with patch(
            "app.plugins.builtins.report_generator.documents.llm_complete",
            AsyncMock(return_value={"content": "Tut mir leid, ich verstehe nicht."}),
        ):
            resp = await docs_client.post(
                f"{DOC_BASE}/documents/suggest",
                json={"prompt": "Angebot erstellen"},
                headers=ORIGIN_HEADER,
            )
        assert resp.status_code == 502
        assert resp.json()["detail"]["code"] == "invalid_ai_response"

    async def test_suggest_viewer_forbidden_403(
        self, docs_client: AsyncClient, db_session
    ):
        await seed_tenant_and_users(db_session)
        await login_client(docs_client, "viewer@tenanta.com")
        resp = await docs_client.post(
            f"{DOC_BASE}/documents/suggest",
            json={"prompt": "Angebot erstellen"},
            headers=ORIGIN_HEADER,
        )
        assert resp.status_code == 403