Reisacher-Forderung/tests/test_document_generation.py

"""Render the supplied templates and reuse stored case documents through HTTP."""

import io
import tempfile
import unittest
from concurrent.futures import ThreadPoolExecutor
from datetime import date
from decimal import Decimal
from pathlib import Path
from types import MappingProxyType, SimpleNamespace
from unittest.mock import MagicMock, patch
from zipfile import ZipFile
from xml.etree import ElementTree

from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.config import settings
from app.db import get_session
from app.document_generation import DOCX_MEDIA_TYPE, TEMPLATE_DIR
from app.file_storage import get_or_create_file
from app.routes import router


class DocumentGenerationTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        config = patch.object(settings, 'LOCAL_STORAGE', self.temp.name)
        config.start()
        self.addCleanup(config.stop)
        self.db = MagicMock()
        self.db.execute.return_value.first.return_value = (1,)
        self.row = dict(
            Client_DB='1', Beleg_Nr='R1', Kunde_Nr='K1', Kunde_Name='Müller & <Partner>',
            Kunde_Adresse='Musterstraße 1', Kunde_PLZ='89073', Kunde_Ort='Ulm',
            Kunde_Email='kunde@example.com', Kunde_Telefon='012345',
            Beleg_Datum=date(2026, 11, 18), Beleg_Betrag=Decimal('550.48'), Beleg_Offen=Decimal('300.00'),
            Selbstbeteiligung_Betrag=Decimal('150.00'), Mwst_Betrag=Decimal('0'), Mwst_Betrag_berechnet=Decimal('104.59'),
            Versicherung_Typ='Vollkasko', Fahrzeug_Kennzeichen='UL-AB 123',
            Standort_Name='Ulm', Abteilung='SC', Mahnstufe=1,
            Verursacher_Text='Original', Verursacher_Aktuell_Text='Aktueller Verursacher',
        )
        self.db.execute.return_value.mappings.return_value.first.return_value = MappingProxyType(self.row)
        app = FastAPI()
        app.include_router(router)
        app.dependency_overrides[get_session] = lambda: self.db

        @app.middleware('http')
        async def authenticated(request, call_next):
            request.state.principal = SimpleNamespace(username='tester', name='Tester', csrf='csrf', role='Mitarbeiter')
            return await call_next(request)

        self.client = TestClient(app)
        self.addCleanup(self.client.close)
        self.url = '/app/forderungen/export/1_R1/'

    def test_each_supplied_template_renders_to_its_own_local_document(self):
        from app.document_generation import DocxTemplate

        for kind in ('Rechnung', 'Mahnung', 'Beitreibungsauftrag'):
            with self.subTest(kind=kind), patch('app.document_generation.DocxTemplate', wraps=DocxTemplate) as template:
                response = self.client.get(self.url, params={'document_type': kind})
                self.assertEqual(response.status_code, 200)
                self.assertEqual(response.headers['content-type'], DOCX_MEDIA_TYPE)
                template.assert_called_once_with(TEMPLATE_DIR / f'{kind}.docx')
                stored = self.root / '1_R1' / f'{kind}_1_R1.docx'
                self.assertEqual(stored.read_bytes(), response.content)
                with ZipFile(io.BytesIO(response.content)) as archive:
                    document = ElementTree.fromstring(archive.read('word/document.xml'))
                text = ''.join(document.itertext())
                self.assertIn('Müller & <Partner>', text)
                self.assertIn('R1', text)
                self.assertNotIn('{{', text)
                if kind != 'Beitreibungsauftrag':
                    self.assertIn('150,00', text)
                    self.assertIn('0,00', text)
                else:
                    self.assertIn('300,00', text)
                    self.assertIn('Aktueller Verursacher', text)
        self.assertEqual(self.row['Beleg_Datum'], date(2026, 11, 18))

    def test_cached_documents_are_returned_unchanged_in_both_storage_layouts(self):
        for kind in ('Rechnung', 'Mahnung', 'Beitreibungsauftrag'):
            for legacy in (False, True):
                with self.subTest(kind=kind, legacy=legacy):
                    tenant = '2' if legacy else '1'
                    folder = self.root / tenant / 'R1' if legacy else self.root / f'{tenant}_R1'
                    folder.mkdir(parents=True, exist_ok=True)
                    stored = folder / f'{kind}_{tenant}_R1.docx'
                    original = f'Already edited {kind}'.encode()
                    stored.write_bytes(original)
                    modified = stored.stat().st_mtime_ns
                    self.db.reset_mock()
                    with patch('app.document_generation.DocxTemplate') as template:
                        response = self.client.get(f'/app/forderungen/export/{tenant}_R1/', params={'document_type': kind})
                    self.assertEqual(response.content, original)
                    self.assertEqual(stored.stat().st_mtime_ns, modified)
                    template.assert_not_called()
                    # Reuse still checks access but does not fetch/render new case data.
                    self.db.execute.assert_called_once()

    def test_old_export_url_defaults_to_mahnung(self):
        response = self.client.get(self.url)
        self.assertEqual(response.status_code, 200)
        self.assertTrue((self.root / '1_R1/Mahnung_1_R1.docx').is_file())

    def test_unsupported_templates_and_unauthorized_cases_are_not_exported(self):
        response = self.client.get(self.url, params={'document_type': '../Rechnung'})
        self.assertEqual(response.status_code, 422)
        stored = self.root / '1_R1/Rechnung_1_R1.docx'
        stored.parent.mkdir()
        stored.write_bytes(b'cached')
        self.db.execute.return_value.first.return_value = None
        response = self.client.get(self.url, params={'document_type': 'Rechnung'})
        self.assertEqual(response.status_code, 404)
        self.assertEqual(stored.read_bytes(), b'cached')

    def test_missing_storage_or_template_has_a_useful_error(self):
        with patch.object(settings, 'LOCAL_STORAGE', None):
            self.assertEqual(self.client.get(self.url).status_code, 503)
        with patch('app.document_generation.TEMPLATE_DIR', self.root / 'missing-templates'):
            response = self.client.get(self.url)
        self.assertEqual(response.status_code, 503)
        self.assertIn('Mahnung.docx', response.json()['detail'])
        self.assertFalse((self.root / '1_R1/Mahnung_1_R1.docx').exists())

    def test_render_failure_does_not_publish_a_partial_document(self):
        with patch('app.document_generation.DocxTemplate') as template:
            template.return_value.render.side_effect = RuntimeError('broken template')
            with self.assertRaises(RuntimeError):
                self.client.get(self.url)
        self.assertFalse((self.root / '1_R1/Mahnung_1_R1.docx').exists())
        self.assertEqual(list((self.root / '.upload-staging').iterdir()), [])

    def test_concurrent_creation_reuses_one_result(self):
        created = []

        def create(path):
            created.append(path)
            path.write_bytes(b'rendered')

        def load(_):
            return get_or_create_file({'Client_DB': '1', 'Beleg_Nr': 'R1'}, 'Rechnung_1_R1.docx', create)

        with ThreadPoolExecutor(max_workers=3) as pool:
            paths = list(pool.map(load, range(3)))
        self.assertEqual(len(created), 1)
        self.assertEqual(len(set(paths)), 1)
        self.assertEqual(paths[0].read_bytes(), b'rendered')

    def test_document_paths_are_validated_before_creating_files(self):
        for number in ('../escape', 'a/b', 'CON'):
            with self.subTest(number=number), self.assertRaises(ValueError):
                get_or_create_file({'Client_DB': '1', 'Beleg_Nr': number}, 'Rechnung.docx', lambda path: None)

    def test_form_buttons_select_the_correct_template_and_allow_downloads(self):
        from html.parser import HTMLParser
        from urllib.parse import parse_qs, urlsplit
        from app.jinja_templates import templates

        class ExportLinks(HTMLParser):
            def __init__(self):
                super().__init__()
                self.links = []

            def handle_starttag(self, tag, attrs):
                attrs = dict(attrs)
                if tag == 'a' and '/export/' in attrs.get('href', ''):
                    self.links.append(attrs)

        html = templates.env.get_template('forderungen/details/details_formular.html').render(
            forderungen_details=[{**self.row, 'Wiedervorlage': date(2026, 11, 18)}], versicherungen=[],
            request=SimpleNamespace(state=SimpleNamespace(principal=SimpleNamespace(csrf='csrf'))),
            return_url='/app/forderungen/liste', detail_url='/app/forderungen/details/1_R1', add7days='2026-11-25',
        )
        parser = ExportLinks()
        parser.feed(html)
        self.assertEqual(len(parser.links), 3)
        for link, kind in zip(parser.links, ('Rechnung', 'Mahnung', 'Beitreibungsauftrag')):
            self.assertEqual(link['hx-boost'], 'false')
            url = urlsplit(link['href'])
            self.assertEqual(url.path, self.url)
            self.assertEqual(parse_qs(url.query), {'document_type': [kind]})