Reisacher-Forderung/tests/test_forderung_list.py

"""List response contracts with an isolated query result (no SQL Server needed)."""

import unittest
from collections import namedtuple
from datetime import date
from html.parser import HTMLParser
from types import SimpleNamespace

from fastapi import FastAPI
from fastapi.testclient import TestClient

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


def invoice(number=0, **changes):
    data = dict(
        Client_DB="1",
        Hauptbetrieb_Name="AHR",
        Standort_ID="10",
        Standort_Name="ULM",
        Abteilung="SC",
        Verursacher_Text="dki",
        Rechnung_Gutschrift="Rechnung",
        Fahrzeug_Nr="V1",
        Kunde_Text="Musterkunde",
        Beleg_Nr=f"R{number}",
        Beleg_Datum=date(2026, 10, 1),
        Beleg_Faellig_Datum=date(2026, 10, 15),
        Staffel_Text="> 12 Wochen",
        Mahnstufe="1",
        Bearbeitet="N",
        Abwarten="",
        Wiedervorlage=date(2026, 11, 1),
        Wiedervorlage_Stufe="0",
        Beleg_Betrag=20,
        Beleg_Offen=10,
        Kunde_offen_gesamt=50,
        Kunde_gesamt_Bezug="",
        Besitzer_Kunde_Text="",
        Kunde_Email="kunde@example.com",
        Beleg_Tage_offen=28,
        Kommentar_Fibu="Fibu Kommentar",
        Kommentar_Abteilung="Abteilung Kommentar",
        Antrag_Nr="",
        Antrag_Art="",
        Antrag_Datum=None,
        Antrag_Benutzer_ID="",
        Antrag_Status="",
    )
    data.update(changes)
    return namedtuple("Invoice", data)(**data)


class QueryResult:
    def __init__(self, rows):
        self.rows = rows

    def keys(self):
        return invoice()._fields

    def fetchall(self):
        return self.rows


class ListDB:
    def __init__(self):
        self.rows = []
        self.calls = []

    def execute(self, query, values):
        self.calls.append((str(query), values))
        return QueryResult(self.rows)


class Tags(HTMLParser):
    def __init__(self, html):
        super().__init__()
        self.tags = []
        self.feed(html)

    def handle_starttag(self, tag, attrs):
        self.tags.append((tag, dict(attrs)))


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

        @self.application.middleware("http")
        async def fixture_principal(request, call_next):
            request.state.principal = SimpleNamespace(username="tester", name="Tester", role="Buchhaltung", csrf="csrf")
            return await call_next(request)

        self.client = TestClient(self.application)
        self.addCleanup(self.client.close)

    def test_empty_first_page_retains_all_filters_and_zero_summary(self):
        response = self.client.get("/app/forderungen/liste")
        self.assertEqual(response.status_code, 200)
        self.assertIn("Keine Forderungen", response.text)
        self.assertIn("0,00 €", response.text)
        tags = Tags(response.text).tags
        fields = [a for t, a in tags if t in ("input", "select") and a.get("id")]
        self.assertEqual(len(fields), 16)
        self.assertIn('id="list-count">0', response.text)

    def test_empty_results_preserve_selected_filter_and_query_parameters(self):
        response = self.client.get("/app/forderungen/liste?Filiale=10&Kunde=Nichtgefunden")
        self.assertEqual(response.status_code, 200)
        self.assertIn('<option value="10" selected>10</option>', response.text)
        self.assertIn('value="Nichtgefunden"', response.text)
        sql, params = self.db.calls[-1]
        self.assertEqual(params["auth_username"], "tester")
        self.assertEqual(params["filter_filiale"], "%10%")
        self.assertEqual(params["filter_kunde"], "%Nichtgefunden%")
        self.assertIn("F.[Standort_ID] LIKE :filter_filiale", sql)

    def test_summary_uses_all_matches_and_paging_preserves_filter_form(self):
        self.db.rows = [invoice(n) for n in range(103)]
        response = self.client.get("/app/forderungen/liste?Filiale=10")
        self.assertIn('id="list-count">103', response.text)
        self.assertIn("1.030,00 €", response.text)
        tags = Tags(response.text).tags
        rows = [a for t, a in tags if t == "tr" and a.get("class") == "gc-invoice-row"]
        self.assertEqual(len(rows), 100)
        loader = next(a for t, a in tags if a.get("class") == "gc-list-load")
        self.assertEqual(loader["hx-include"], "#liste-filter")
        self.assertIn('"page": 1', loader["hx-vals"])

    def test_last_page_is_a_fragment_and_exhausted_page_is_empty_html(self):
        self.db.rows = [invoice(n) for n in range(103)]
        response = self.client.get("/app/forderungen/liste?page=1&Filiale=10")
        self.assertEqual(response.status_code, 200)
        self.assertNotIn("<html", response.text)
        self.assertNotIn("gc-list-load", response.text)
        self.assertEqual(response.text.count('class="gc-invoice-row"'), 3)
        self.assertIn("R100", response.text)
        exhausted = self.client.get("/app/forderungen/liste?page=2&Filiale=10")
        self.assertEqual(exhausted.status_code, 200)
        self.assertEqual(exhausted.text, "")
        self.assertTrue(exhausted.headers["content-type"].startswith("text/html"))

    def test_comments_are_escaped_and_unique_across_tenants(self):
        self.db.rows = [invoice(1, Kommentar_Fibu="<script>unsafe</script>"), invoice(1, Client_DB="2")]
        response = self.client.get("/app/forderungen/liste")
        self.assertIn("&lt;script&gt;unsafe&lt;/script&gt;", response.text)
        tags = Tags(response.text).tags
        controls = [a["aria-controls"] for t, a in tags if "data-gc-comments" in a]
        self.assertEqual(len(set(controls)), 2)
        for target in controls:
            self.assertTrue(any(t == "tr" and a.get("id") == target and "hidden" in a for t, a in tags))

    def test_application_information_precedes_department_comment(self):
        self.db.rows = [
            invoice(
                1,
                Antrag_Nr="A42",
                Antrag_Art="Antrag",
                Antrag_Datum=date(2026, 10, 1),
                Antrag_Benutzer_ID="dki",
                Antrag_Status="Offen",
            ),
            invoice(2, Kommentar_Fibu="", Kommentar_Abteilung=""),
        ]
        response = self.client.get("/app/forderungen/liste")
        self.assertIn("Antrag A42", response.text)
        self.assertNotIn("Abteilung Kommentar", response.text)
        self.assertEqual(response.text.count("data-gc-comments"), 1)

    def test_return_focus_loads_through_target_and_continues_at_next_block(self):
        self.db.rows = [invoice(n) for n in range(305)]
        response = self.client.get("/app/forderungen/liste?Filiale=10&focus=1_R220")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.text.count('class="gc-invoice-row"'), 300)
        self.assertIn('data-primary-key="1_R220"', response.text)
        self.assertIn('"page": 3', response.text)
        self.assertIn('id="list-count">305', response.text)
        self.assertEqual(response.headers["cache-control"], "no-store")
        sql, parameters = self.db.calls[-1]
        self.assertNotIn("focus", parameters)
        self.assertNotIn("focus", sql.lower())

    def test_focus_is_found_at_new_position_and_within_exact_tenant(self):
        self.db.rows = [invoice(220, Client_DB="2")] + [invoice(n) for n in range(150)] + [invoice(220)]
        response = self.client.get("/app/forderungen/liste?Filiale=10&focus=1_R220")
        self.assertEqual(response.text.count('class="gc-invoice-row"'), 152)
        self.assertNotIn("gc-return-status", response.text)
        self.assertIn('data-focus="1_R220"', response.text)

    def test_missing_focus_preserves_filters_and_explains_missing_row(self):
        response = self.client.get("/app/forderungen/liste?Filiale=10&focus=1_R404")
        self.assertIn("nicht mehr enthalten", response.text)
        self.assertIn('<option value="10" selected>10</option>', response.text)
        self.assertIn("Keine Forderungen", response.text)

    def test_detail_links_keep_filters_including_on_loaded_fragments(self):
        from urllib.parse import parse_qs, urlsplit

        self.db.rows = [invoice(n) for n in range(103)]
        for url, key in [
            ("/app/forderungen/liste?Filiale=10&Kunde=Muster", "1_R0"),
            ("/app/forderungen/liste?page=1&Filiale=10&Kunde=Muster", "1_R100"),
        ]:
            response = self.client.get(url)
            link = next(a for t, a in Tags(response.text).tags if "data-gc-detail-link" in a)
            return_to = parse_qs(urlsplit(link["href"]).query)["return_to"][0]
            self.assertEqual(
                parse_qs(urlsplit(return_to).query), {"Filiale": ["10"], "Kunde": ["Muster"], "focus": [key]}
            )


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