Reisacher-Forderung/tests/test_forderung_notice.py

import unittest
from datetime import date
from decimal import Decimal
from unittest.mock import MagicMock, patch

from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.testclient import TestClient

from app.db import get_session
from app.permissions import Principal
from app.routes import router


class ForderungNoticeTests(unittest.TestCase):
    def setUp(self):
        self.db = MagicMock()
        self.application = FastAPI()
        self.application.include_router(router)
        self.application.dependency_overrides[get_session] = lambda: self.db

        @self.application.middleware("http")
        async def authenticated(request: Request, call_next):
            request.state.principal = Principal("tester", "Tester", "", "Mitarbeiter", "ldap", "csrf", ())
            return await call_next(request)

        self.client = TestClient(self.application)
        self.addCleanup(self.client.close)
        self.detail_context_patch = patch("app.routes.get_template_context")
        self.detail_context = self.detail_context_patch.start()
        self.addCleanup(self.detail_context_patch.stop)
        self.current = {
            "Beleg_Nr": "WGG24501320",
            "Beleg_Datum": date(2024, 5, 13),
            "Kunde_Text": "Beispielkunde <script>alert(1)</script>",
            "Beleg_Betrag": Decimal("1234.50"),
            "Beleg_Offen": Decimal("250.00"),
        }

    def result(self, row=None, access=False):
        result = MagicMock()
        result.first.return_value = (1,) if access else None
        result.mappings.return_value.first.return_value = row
        return result

    def get_notice(self, htmx=False):
        return self.client.get("/app/forderungen/details/1_WGG24501320", headers={"HX-Request": "true"} if htmx else {})

    def assert_notice(self, response):
        self.assertEqual(response.status_code, 200)
        self.assertIn('href="/app/forderungen/liste?focus=1_WGG24501320">Zurück zur Liste', response.text)
        self.assertNotIn('id="formular"', response.text)
        self.assertNotIn('name="Kommentar"', response.text)
        self.detail_context.assert_not_called()

    def test_open_forderung_displays_only_summary_and_skips_history(self):
        self.db.execute.side_effect = [self.result(), self.result(self.current)]
        response = self.get_notice()
        self.assert_notice(response)
        self.assertIn("Forderung WGG24501320 vom 13.05.2024 ist noch offen.", response.text)
        self.assertIn("1.234,50 €", response.text)
        self.assertIn("250,00 €", response.text)
        self.assertIn("&lt;script&gt;", response.text)
        self.assertNotIn("<script>alert(1)</script>", response.text)
        self.assertEqual(self.db.execute.call_count, 2)

    def test_history_displays_dates_and_queries_exact_tenant_key(self):
        history = {"Beleg_Nr": "WGG24501320", "Beleg_Datum": date(2024, 5, 13), "Auszifferung_Datum": date(2024, 6, 2)}
        self.db.execute.side_effect = [self.result(), self.result(), self.result(history)]
        response = self.get_notice(htmx=True)
        self.assert_notice(response)
        self.assertIn("wurde am 02.06.2024 beglichen.", response.text)
        query, params = self.db.execute.call_args.args
        self.assertIn("dbo.Forderungen_Historie", str(query))
        self.assertIn("ORDER BY Auszifferung_Datum DESC", str(query))
        self.assertEqual(params, {"client_db": "1", "beleg_nr": "WGG24501320"})
        self.assertNotIn("WGG24501320", str(query))

    def test_missing_entry_displays_short_message(self):
        self.db.execute.side_effect = [self.result(), self.result(), self.result()]
        response = self.get_notice()
        self.assert_notice(response)
        self.assertIn("Eintrag nicht gefunden.", response.text)

    def test_null_fields_and_zero_amount_render_correctly(self):
        row = dict(self.current, Beleg_Datum=None, Kunde_Text=None, Beleg_Betrag=None, Beleg_Offen=Decimal("0"))
        self.db.execute.side_effect = [self.result(), self.result(row)]
        response = self.get_notice()
        self.assert_notice(response)
        self.assertEqual(response.text.count("nicht hinterlegt"), 3)
        self.assertIn("0,00 €", response.text)

    def test_authorized_user_retains_existing_detail_flow(self):
        self.db.execute.return_value = self.result(access=True)
        self.detail_context.return_value = {"existing": "context"}
        with patch("app.routes.templates.TemplateResponse", return_value=HTMLResponse("Full details")) as render:
            response = self.get_notice()
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.text, "Full details")
        self.detail_context.assert_called_once()
        self.assertEqual(render.call_args.args[1], "forderungen/details/details.html")
        self.db.execute.assert_called_once()

    def test_database_failure_is_not_reported_as_missing(self):
        self.db.execute.side_effect = [self.result(), RuntimeError("Database unavailable")]
        with TestClient(self.application, raise_server_exceptions=False) as client:
            response = client.get("/app/forderungen/details/1_WGG24501320")
        self.assertEqual(response.status_code, 500)
        self.assertNotIn("Eintrag nicht gefunden", response.text)

    def test_write_file_and_export_routes_remain_denied(self):
        self.db.execute.return_value = self.result()
        cases = [
            ("post", "/app/forderungen/details/1_WGG24501320", {"data": {"source": "comments", "Kommentar": "test"}}),
            ("get", "/app/forderungen/files/1_WGG24501320", {}),
            ("get", "/files/?name=1/WGG24501320/invoice.pdf", {}),
            ("get", "/app/forderungen/export/1_WGG24501320/", {}),
        ]
        for method, url, kwargs in cases:
            with self.subTest(url=url):
                self.db.execute.reset_mock()
                response = getattr(self.client, method)(url, **kwargs)
                self.assertEqual(response.status_code, 404)
                self.db.execute.assert_called_once()
        self.db.commit.assert_not_called()


if __name__ == "__main__":
    unittest.main()