Reisacher-Forderung/tests/test_dashboard.py

"""Dashboard and exact drill-down contracts on the shared production SQL projection."""

import json
import re
import unittest
from collections import namedtuple
from datetime import datetime
from decimal import Decimal
from unittest.mock import patch
from urllib.parse import parse_qs, urlencode, urlsplit

from app.dashboard_segment import pack_segment, parse_segment
from app.forderung_dashboard import BERLIN, build_dashboard, color
from app.navigation import list_return_url, normalize_page_url
from tests import test_verursacher as fixture
from tests.test_forderung_list import invoice
from app.query_filter import decode_filter_values


class DashboardTests(unittest.TestCase):
    def setUp(self):
        self.fixture = fixture.VerursacherTests()
        self.fixture.setUp()
        self.addCleanup(self.fixture.doCleanups)
        self.client, self.db = self.fixture.client, self.fixture.db
        from app.jinja_templates import templates

        sql = templates.env.get_template("forderungen/liste/queries/forderungen_liste.sql").render()
        needed = set(invoice()._fields) | set(re.findall(r"(?:\[F\]|F)\.\[([^]]+)\]", sql))
        existing = {row[1] for row in self.db.connection.execute("PRAGMA dbo.table_info(Forderungen_Kopf)")}
        for name in needed - existing:
            kind = (
                "INTEGER"
                if name in {"Beleg_Tage_offen", "Wiedervorlage_Staffel_Nr"}
                else "REAL" if name in {"Beleg_Betrag", "Kunde_offen_gesamt"} else "TEXT"
            )
            self.db.connection.execute(f"ALTER TABLE dbo.Forderungen_Kopf ADD [{name}] {kind}")
        self.db.connection.executescript("""
            ALTER TABLE dbo.Forderungen_Details ADD Rechtsanwalt TEXT;
            ALTER TABLE dbo.Forderungen_Details ADD Reklamation TEXT;
            ALTER TABLE dbo.Forderungen_Details ADD Mahnen_aussetzen TEXT;
            CREATE TABLE dbo.Forderungen_Kommentar_letzte (Client_DB TEXT, Beleg_Nr TEXT, Kommentar_Fibu TEXT, Kommentar_Abteilung TEXT, Begründung TEXT);
            CREATE TABLE dbo.Garantie2 (Client_DB TEXT, Beleg_Nr TEXT, Antrag_Art TEXT, Antrag_Nr TEXT, Antrag_Datum TEXT, Antrag_Benutzer_ID TEXT, Antrag_Status TEXT, Letzte_Beantragung TEXT);
            UPDATE dbo.Forderungen_Kopf SET Standort_Name='ULM', Hauptbetrieb_Name='AHR',
                Staffel_Text='> 12 Wochen', Beleg_Tage_offen=100, Rechnung_Gutschrift='Rechnung',
                Beleg_Faellig_Datum='2026-11-02', Beleg_Betrag=Beleg_Offen, Kunde_offen_gesamt=100;
            UPDATE dbo.Forderungen_Kopf SET Hauptbetrieb_ID='2', Hauptbetrieb_Name='AAM' WHERE Client_DB='2';
            UPDATE dbo.Forderungen_Kopf SET Staffel_Text='< 4 Wochen', Beleg_Tage_offen=10 WHERE Beleg_Nr='R1';
            INSERT INTO dbo.Forderungen_Details (Client_DB,Beleg_Nr,Verursacher_Benutzer_ID,Wiedervorlage) VALUES
                ('1','R1','alice','2026-11-01'), ('1','R3',NULL,'2026-11-02 13:30:00'),
                ('1','X1','bob','2026-11-09 13:30:00'), ('1','NV',NULL,'2026-11-10');
            UPDATE dbo.Benutzer_Rolle SET Name='Alice' WHERE Benutzer_ID='bob';
            INSERT INTO Garantie2 (Client_DB,Beleg_Nr,Antrag_Nr) VALUES ('1','R1','A'),('1','R1','B');
            INSERT INTO dbo.Benutzer_Rechte VALUES ('tester','1','10','Service');
            INSERT INTO dbo.Benutzer_Rechte VALUES ('tester','1','10','');
        """)
        columns = [row[1] for row in self.db.connection.execute("PRAGMA dbo.table_info(Forderungen_Kopf)")]
        select = ",".join(
            "'CR'" if name == "Beleg_Nr" else "-40" if name == "Beleg_Offen" else "[" + name + "]" for name in columns
        )
        self.db.connection.execute(
            "INSERT INTO dbo.Forderungen_Kopf SELECT "
            + select
            + " FROM dbo.Forderungen_Kopf WHERE Client_DB='1' AND Beleg_Nr='NV'"
        )
        self.db.connection.execute(
            "UPDATE dbo.Forderungen_Kopf SET Rechnung_Gutschrift='Gutschrift', Abteilung='', Staffel_Text=NULL, Verursacher_Benutzer_ID=NULL, Verursacher_Text=NULL, Beleg_Faellig_Datum=NULL WHERE Beleg_Nr='CR'"
        )
        # Model source-owned classifications independently of the dashboard clock.
        self.db.connection.executescript("""
            UPDATE dbo.Forderungen_Kopf SET Wiedervorlage_Staffel_Nr=2, Wiedervorlage_Staffel_Text='In 3 Tagen fällig';
            UPDATE dbo.Forderungen_Kopf SET Wiedervorlage_Staffel_Nr=1, Wiedervorlage_Staffel_Text='Überfällig' WHERE Beleg_Nr='R1';
            UPDATE dbo.Forderungen_Kopf SET Wiedervorlage_Staffel_Nr=3, Wiedervorlage_Staffel_Text='Noch nicht fällig' WHERE Beleg_Nr IN ('X1','NV');
            UPDATE dbo.Forderungen_Kopf SET Wiedervorlage_Staffel_Nr=NULL, Wiedervorlage_Staffel_Text=NULL WHERE Beleg_Nr='CR';
        """)
        # The production query now reads a DWH view. Model that view in SQLite;
        # keep the detail overlays live so update/authorization tests still run real SQL.
        overlays = {
            'Verursacher_Benutzer_ID': 'COALESCE(D.Verursacher_Benutzer_ID, F.Verursacher_Benutzer_ID)',
            'Verursacher_Text': 'COALESCE(B.Name, D.Verursacher_Benutzer_ID, F.Verursacher_Text)',
            'Wiedervorlage': 'COALESCE(D.Wiedervorlage, F.Beleg_Faellig_Datum)',
            'Bearbeitet': "COALESCE(D.Bearbeitet, 'N')",
            'Abwarten': "CASE WHEN D.Reklamation='J' THEN 'Rekl.' WHEN D.Versicherung='J' THEN 'Vers.' "
                        "WHEN D.Rechtsanwalt='J' THEN 'RA' WHEN D.Mahnen_aussetzen='J' THEN 'N' ELSE '' END",
        }
        projection = ', '.join(f'{overlays.get(name, "F.[" + name + "]")} AS [{name}]' for name in columns)
        self.db.connection.execute(
            'CREATE VIEW dbo.Forderungen_Liste AS SELECT ' + projection +
            ' FROM Forderungen_Kopf F LEFT JOIN Forderungen_Details D ON F.Client_DB=D.Client_DB AND F.Beleg_Nr=D.Beleg_Nr'
            ' LEFT JOIN Benutzer_Rolle B ON D.Verursacher_Benutzer_ID=B.Benutzer_ID'
            ' LEFT JOIN Garantie2 G ON F.Client_DB=G.Client_DB AND F.Beleg_Nr=G.Beleg_Nr'
        )
        self.db.connection.commit()
        execute = self.db.execute

        def dialect(query, values=None):
            query = str(query)
            for column in ["Kommentar_Fibu", "Kommentar_Abteilung"]:
                query = query.replace(
                    f"left(isnull([FK].[{column}], ''), 50)", f"substr(ifnull([FK].[{column}], ''), 1, 50)"
                )
            query = (
                query.replace("isnull(", "ifnull(")
                .replace("nvarchar(max)", "TEXT")
                .replace("dateadd(day, 4, getdate())", "'2026-11-06'")
                .replace("getdate()", "'2026-11-02'")
            )
            if query.strip().startswith("SELECT F.* FROM"):
                self.db.calls.append((query, values))
                cursor = self.db.connection.execute(query, values or {})
                columns = [item[0] for item in cursor.description]
                record = namedtuple("DashboardInvoice", columns)
                rows = []
                for raw in cursor.fetchall():
                    row = dict(zip(columns, raw))
                    for name in ["Beleg_Datum", "Beleg_Faellig_Datum", "Wiedervorlage"]:
                        if row.get(name):
                            row[name] = datetime.fromisoformat(row[name])
                    rows.append(record(**row))

                class SQLRows:
                    def keys(self):
                        return columns

                    def fetchall(self):
                        return rows

                return SQLRows()
            result = execute(query, values)
            if ' AS value, MIN(' in query:
                result.rows = [tuple(row.values()) for row in result.rows]
            return result

        adapter = patch.object(self.db, "execute", side_effect=dialect)
        adapter.start()
        self.addCleanup(adapter.stop)
        self.now = datetime(2026, 11, 2, 10, tzinfo=BERLIN)

    def data(self):
        return build_dashboard(self.db, "tester", self.now)

    def test_totals_rights_duplicate_roles_and_duplicate_join_rows(self):
        data = self.data()
        self.assertEqual(data["count"], 5)
        self.assertEqual(data["amount"], "200.00")
        self.assertEqual(data["amount_display"], "200,00")
        for groups in data["views"].values():
            self.assertEqual(sum(group["count"] for group in groups), 5)
            self.assertEqual(sum(Decimal(group["amount"]) for group in groups), Decimal(200))
        sql, values = self.db.calls[-1]
        self.assertIn("WHERE EXISTS", sql)
        self.assertEqual(values["auth_username"], "tester")
        self.assertTrue(all("AAM" not in group["label"] for group in data["views"]["Standort"]))

    def test_only_invoices_filters_all_aggregates_and_drilldowns(self):
        data = build_dashboard(self.db, "tester", self.now, only_invoices=True)
        self.assertEqual(data["count"], 4)
        self.assertEqual(data["amount"], "240.00")
        self.assertEqual(self.data()["count"], 5)
        for groups in data["views"].values():
            self.assertEqual(sum(group["count"] for group in groups), 4)
            for group in groups:
                for item in [group, *group["segments"]]:
                    params = parse_qs(urlsplit(item["url"]).query)
                    self.assertEqual(params['RechnungGutschrift'], ['Rechnung'])
                    self.assertNotIn('dashboard_segment', params)
                    response = self.client.get(item["url"])
                    self.assertIn('id="list-count">' + str(item["count"]), response.text)
                    self.assertIn('name="RechnungGutschrift" value="Rechnung"', response.text)
        self.db.connection.execute(
            "UPDATE dbo.Forderungen_Kopf SET Rechnung_Gutschrift='Rechnungskorrektur' WHERE Beleg_Nr='CR'"
        )
        self.db.connection.execute(
            "UPDATE dbo.Forderungen_Kopf SET Beleg_Offen=-100 WHERE Client_DB='1' AND Beleg_Nr='R1'"
        )
        negative = build_dashboard(self.db, "tester", self.now, only_invoices=True)
        self.assertEqual(negative["count"], 4)
        self.assertEqual(negative["amount"], "40.00")

    def test_invoice_checkbox_url_state_and_strict_segment_flag(self):
        response = self.client.get("/app/forderungen/dashboard?page=Verursacher&metric=anzahl&only_invoices=true")
        self.assertEqual(response.status_code, 200)
        self.assertIn('name="only_invoices" value="true" checked', response.text)
        self.assertIn('name="page" value="Verursacher"', response.text)
        self.assertIn('name="metric" value="anzahl"', response.text)
        self.assertIn("240,00", response.text)
        unchecked = self.client.get("/app/forderungen/dashboard")
        self.assertNotIn('name="only_invoices" value="true" checked', unchecked.text)
        self.assertIn("200,00", unchecked.text)
        self.assertIn('data-dashboard-view="Standort" data-dashboard-metric=', unchecked.text)
        for invalid in ["true", 1, None]:
            with self.assertRaises(ValueError):
                parse_segment(pack_segment({"view": "Altersstaffel", "key": None, "only_invoices": invalid}))

    def test_all_segments_round_trip_to_exact_list_count_and_amount(self):
        for view, groups in self.data()["views"].items():
            for group in groups:
                for segment in group["segments"]:
                    with self.subTest(view=view, label=group["label"], department=segment["department"]):
                        response = self.client.get(segment["url"])
                        self.assertEqual(response.status_code, 200)
                        count = int(re.search(r'id="list-count">(\d+)', response.text)[1])
                        self.assertEqual(count, segment["count"])
                        self.assertNotIn('name="dashboard_segment"', response.text)
                        self.assertNotIn("Dashboardfilter entfernen", response.text)
                        self.assertIn("Aktive Filter (", response.text)
                        from app.jinja_templates import number_format

                        self.assertIn(number_format(Decimal(segment["amount"])) + " €", response.text)

    def test_follow_up_source_staffels_and_21_day_boundary(self):
        cases = [
            ("2026-11-01", "1", "Überfällig"),
            ("2026-11-02", "2", "In 3 Tagen fällig"),
            ("2026-11-05 23:30:00", "2", "In 3 Tagen fällig"),
            ("2026-11-06", "3", "Noch nicht fällig"),
            ("2026-11-17", "3", "Noch nicht fällig"),
            ("2026-11-22 23:30:00", "3", "Noch nicht fällig"),
            ("2026-11-23", "4", "zurückgestellt"),
        ]
        for value, expected, label in cases:
            with self.subTest(date=value):
                self.db.connection.execute(
                    "UPDATE dbo.Forderungen_Details SET Wiedervorlage=? WHERE Beleg_Nr='R1'", (value,)
                )
                self.db.connection.execute(
                    "UPDATE dbo.Forderungen_Kopf SET Wiedervorlage_Staffel_Nr=?, Wiedervorlage_Staffel_Text=? WHERE Beleg_Nr='R1'",
                    (expected, label),
                )
                data = build_dashboard(self.db, "tester", self.now, stack="wiedervorlage")
                group = next(group for group in data["views"]["Altersstaffel"] if group["label"] == "< 4 Wochen")
                segment = group['segments'][0]
                self.assertEqual(segment["id"], expected)
                self.assertEqual(segment["label"], label)
                self.assertEqual(parse_qs(urlsplit(segment['url']).query)['WVStaffel'], [expected])
                response = self.client.get(segment["url"])
                self.assertIn('id="list-count">1', response.text)
                self.assertIn(f'aria-label="{label} entfernen"', response.text)
        ids = [item["id"] for item in data["stack_items"]]
        self.assertEqual(ids, ["2", "3", "4", None])
        self.assertNotIn("Wiedervorlage", data["views"])

    def test_follow_up_uses_source_staffel_even_when_dates_disagree(self):
        self.db.connection.execute("UPDATE dbo.Forderungen_Details SET Wiedervorlage=NULL WHERE Beleg_Nr='R1'")
        self.db.connection.execute(
            "UPDATE dbo.Forderungen_Kopf SET Beleg_Faellig_Datum='2099-01-01', "
            "Wiedervorlage_Staffel_Nr=10, Wiedervorlage_Staffel_Text='Bezeichnung aus Datenquelle' "
            "WHERE Client_DB='1' AND Beleg_Nr='R1'"
        )
        data = build_dashboard(self.db, "tester", self.now, stack="wiedervorlage")
        group = next(group for group in data["views"]["Altersstaffel"] if group["label"] == "< 4 Wochen")
        self.assertEqual(group["segments"][0]["id"], "10")
        self.assertEqual(group["segments"][0]["label"], "Bezeichnung aus Datenquelle")
        self.assertEqual([item['id'] for item in data['stack_items']], ['2', '3', '10', None])
        self.assertIn('id="list-count">1', self.client.get(group["segments"][0]["url"]).text)

    def test_follow_up_source_staffels_are_independent_of_dashboard_clock(self):
        self.db.connection.execute("UPDATE dbo.Forderungen_Details SET Wiedervorlage='2027-01-01' WHERE Beleg_Nr='R1'")
        now = datetime.fromisoformat("2026-12-28T23:30:00+00:00")
        groups = build_dashboard(self.db, "tester", now, stack="wiedervorlage")["views"]["Altersstaffel"]
        group = next(group for group in groups if group["label"] == "< 4 Wochen")
        params = parse_qs(urlsplit(group['segments'][0]['url']).query)
        self.assertEqual(decode_filter_values(params['Staffel'][0]), ['< 4 Wochen'])
        self.assertEqual(params['WVStaffel'], ['1'])
        self.assertNotIn('WVStaffel', parse_qs(urlsplit(group['url']).query))
        self.assertNotIn('WiedervorlageVon', params)
        self.assertNotIn('WiedervorlageBis', params)
        self.assertEqual(group["segments"][0]["id"], "1")

    def test_follow_up_stack_roundtrips_for_all_rubrics_and_invoice_scope(self):
        for only in [False, True]:
            data = build_dashboard(self.db, "tester", self.now, only_invoices=only, stack="wiedervorlage")
            for groups in data["views"].values():
                self.assertEqual(sum(group["count"] for group in groups), data["count"])
                self.assertEqual(sum(Decimal(group["amount"]) for group in groups), Decimal(data["amount"]))
                for group in groups:
                    for segment in group["segments"]:
                        params = parse_qs(urlsplit(segment['url']).query)
                        self.assertNotIn('dashboard_segment', params)
                        self.assertNotIn('WiedervorlageVon', params)
                        self.assertNotIn('WiedervorlageBis', params)
                        self.assertNotIn('Abteilung', params)
                        self.assertEqual(params['WVStaffel'], [segment['id'] or '__empty__'])
                        self.assertNotIn('WVStaffel', parse_qs(urlsplit(group['url']).query))
                        response = self.client.get(segment["url"])
                        self.assertEqual(response.status_code, 200)
                        self.assertIn('id="list-count">' + str(segment["count"]), response.text)
                        from app.jinja_templates import number_format
                        self.assertIn(number_format(Decimal(segment['amount'])) + ' €', response.text)

    def test_stack_selection_and_legacy_follow_up_links(self):
        response = self.client.get("/app/forderungen/dashboard?stack=wiedervorlage&metric=anzahl&only_invoices=true")
        self.assertEqual(response.status_code, 200)
        self.assertNotIn('data-dashboard-view="Wiedervorlage"', response.text)
        self.assertIn('value="wiedervorlage" selected', response.text)
        self.assertIn('name="stack" value="wiedervorlage"', response.text)
        self.assertEqual(self.client.get("/app/forderungen/dashboard?stack=bad").status_code, 422)
        old = self.client.get(
            "/app/forderungen/dashboard?page=Wiedervorlage&metric=anzahl&only_invoices=true", follow_redirects=False
        )
        self.assertEqual(old.status_code, 303)
        params = parse_qs(urlsplit(old.headers["location"]).query)
        self.assertEqual(
            params, {"page": ["Standort"], "metric": ["anzahl"], "stack": ["wiedervorlage"], "only_invoices": ["true"]}
        )
        legacy = pack_segment({"view": "Wiedervorlage", "key": {"from": "2026-11-02", "to": "2026-11-09"}})
        response = self.client.get("/app/forderungen/liste", params={"dashboard_segment": legacy})
        self.assertIn('id="list-count">5', response.text)
        self.assertEqual(urlsplit(str(response.url)).query, '')
        for bad in [True, "tomorrow", {"from": "2026-11-02", "to": "2026-11-01"}]:
            with self.assertRaises(ValueError):
                parse_segment(pack_segment({"view": "Altersstaffel", "key": None, "follow_up": bad}))

    def test_same_names_share_normal_filter_but_clients_remain_separate(self):
        groups = [group for group in self.data()["views"]["Verursacher"] if group["label"] == 'Alice']
        self.assertEqual(len(groups), 1)
        self.assertEqual(groups[0]['count'], 2)
        self.assertEqual(parse_qs(urlsplit(groups[0]['url']).query), {'Verursacher': ['Alice']})
        self.db.connection.execute("INSERT INTO dbo.Benutzer_Rechte VALUES ('tester','2','10','Service')")
        groups = self.data()["views"]["Standort"]
        self.assertEqual(len(groups), 2)
        for group in groups:
            self.assertEqual(self.client.get(group["url"]).status_code, 200)
        self.assertEqual({group["label"] for group in groups}, {"ULM (AHR)", "ULM (AAM)"})

    def test_decimal_sum_and_zero_amount_segments_still_have_links(self):
        self.db.connection.execute(
            "UPDATE dbo.Forderungen_Kopf SET Beleg_Offen=0.1 WHERE Client_DB='1' AND Beleg_Nr='R1'"
        )
        self.db.connection.execute(
            "UPDATE dbo.Forderungen_Kopf SET Beleg_Offen=0.2 WHERE Client_DB='1' AND Beleg_Nr='X1'"
        )
        data = self.data()
        self.assertEqual(data["amount"], "10.30")
        zero = build_dashboard(self.db, "tester", self.now, stack="wiedervorlage")
        self.assertTrue(
            any(
                segment["amount"] == "0.00" and segment["url"]
                for group in zero["views"]["Verursacher"]
                for segment in group["segments"]
            )
        )
        self.assertEqual(color("Service"), color("Service"))

    def test_exact_department_filter_does_not_match_prefixes(self):
        self.db.connection.execute("UPDATE dbo.Forderungen_Kopf SET Abteilung='Service2' WHERE Beleg_Nr='CR'")
        self.db.connection.execute("INSERT INTO dbo.Benutzer_Rechte VALUES ('tester','1','10','Service2')")
        group = self.data()["views"]["Standort"][0]
        segment = next(segment for segment in group["segments"] if segment["department"] == "Service")
        exact = self.client.get(segment["url"])
        partial = self.client.get("/app/forderungen/liste?Abteilung=Service")
        self.assertIn('id="list-count">4', exact.text)
        self.assertIn('id="list-count">4', partial.text)

    def test_dashboard_filters_can_be_changed_without_hidden_constraints(self):
        from app.navigation import list_url
        group = self.data()['views']['Standort'][0]
        segment = next(item for item in group['segments'] if item['department'] == 'Service')
        params = {name: values[0] for name, values in parse_qs(urlsplit(segment['url']).query).items()}
        self.assertEqual(params, {'Mandant': '1', 'Filiale': '10', 'Abteilung': 'Service'})
        response = self.client.get(segment['url'])
        self.assertIn('id="list-count">4', response.text)
        self.assertIn('data-remove="Service"', response.text)
        params.pop('Abteilung')
        response = self.client.get(list_url(params))
        self.assertIn('id="list-count">5', response.text)
        self.assertNotIn('name="dashboard_segment"', response.text)
        params['Staffel'] = '%3C%204%20Wochen'
        response = self.client.get(list_url(params))
        self.assertIn('id="list-count">1', response.text)

    def test_old_dashboard_link_redirects_to_visible_filters(self):
        legacy = pack_segment({'view': 'Standort', 'key': {'client': '1', 'site': '10'}, 'department': 'Service'})
        response = self.client.get('/app/forderungen/liste', params={'dashboard_segment': legacy}, follow_redirects=False)
        self.assertEqual(response.status_code, 307)
        self.assertEqual(response.headers['location'], '/app/forderungen/liste?Mandant=1&Filiale=10&Abteilung=Service')
        response = self.client.get(response.headers['location'])
        self.assertIn('id="list-count">4', response.text)
        self.assertIn('data-remove="Service"', response.text)
        self.assertNotIn('dashboard_segment', response.text)
        self.assertNotIn('Dashboardfilter', response.text)

    def test_old_person_id_is_translated_to_normal_name_filter(self):
        legacy = pack_segment({'view': 'Verursacher', 'key': {'id': 'alice'}, 'only_invoices': True})
        response = self.client.get('/app/forderungen/liste', params={'dashboard_segment': legacy})
        params = parse_qs(urlsplit(str(response.url)).query)
        self.assertEqual(params, {'Verursacher': ['Alice'], 'RechnungGutschrift': ['Rechnung']})
        self.assertIn('id="list-count">2', response.text)
        self.assertIn('data-remove="Alice"', response.text)
        self.assertNotIn('dashboard_segment', response.text)

    def test_explicit_filters_override_old_link_and_login_recovers_visible_filters(self):
        legacy = pack_segment({'view': 'Standort', 'key': {'client': '1', 'site': '10'}, 'department': 'Service'})
        response = self.client.get('/app/forderungen/liste', params={'dashboard_segment': legacy, 'Filiale': '20'})
        self.assertEqual(parse_qs(urlsplit(str(response.url)).query)['Filiale'], ['20'])
        self.assertIn('id="list-count">0', response.text)
        self.client.cookies.clear()
        response = self.client.get('/app/forderungen/liste', params={'dashboard_segment': legacy}, headers={'HX-Request': 'true'})
        target = parse_qs(urlsplit(response.headers['HX-Redirect']).query)['next'][0]
        self.assertEqual(target, '/app/forderungen/liste?Mandant=1&Filiale=10&Abteilung=Service')

    def test_new_links_preserve_names_with_commas_and_spaces(self):
        self.db.connection.execute("UPDATE dbo.Benutzer_Rolle SET Name='Link, Florian' WHERE Benutzer_ID='alice'")
        group = next(item for item in self.data()['views']['Verursacher'] if item['label'] == 'Link, Florian')
        self.assertIn('Verursacher=Link%252C%2520Florian', group['url'])
        response = self.client.get(group['url'])
        self.assertIn('id="list-count">1', response.text)
        self.assertIn('data-remove="Link, Florian"', response.text)

    def test_forged_segment_cannot_expand_invoice_access(self):
        raw = pack_segment({"view": "Standort", "key": {"client": "2", "site": "10"}})
        response = self.client.get("/app/forderungen/liste", params={"dashboard_segment": raw})
        self.assertEqual(response.status_code, 200)
        self.assertIn('id="list-count">0', response.text)
        self.assertIn("0,00 €", response.text)

    def test_segment_preserved_through_detail_return_login_and_normal_filters(self):
        url = self.data()["views"]["Standort"][0]["url"]
        filters = parse_qs(urlsplit(url).query)
        returned = list_return_url(url, "1_R1")
        self.assertEqual(parse_qs(urlsplit(returned).query), {**filters, 'focus': ['1_R1']})
        detail = "/app/forderungen/details/1_R1?" + urlencode({"return_to": returned})
        recovered = normalize_page_url(
            "/app/forderungen/details/1_R1/kommentare", current_url="https://proxy.test" + detail
        )
        self.assertEqual(recovered, detail)
        response = self.client.get(url + "&Kunde=Test&page=1&limit=1")
        self.assertEqual(response.status_code, 200)
        self.assertIn('hx-include="#liste-filter"', response.text)
        sql, values = self.db.calls[-1]
        self.assertIn("F.[Client_DB] = :filter_mandant", sql)
        self.assertEqual(values["filter_kunde"], "%Test%")
        self.client.cookies.clear()
        response = self.client.get(url, headers={"HX-Request": "true"})
        target = parse_qs(urlsplit(response.headers["HX-Redirect"]).query)["next"][0]
        self.assertEqual(target, url)

    def test_bad_segments_are_rejected_before_data_query(self):
        invalid = [
            "not-json",
            "{}",
            '{"view":"Standort","key":{}}',
            '{"view":"Wiedervorlage","key":{"from":"x","to":null}}',
            '{"view":"Altersstaffel","key":null,"sql":"1=1"}',
            '{"view":"Verursacher","key":{"id":null}}',
            '{"view":"Standort","key":{"client":true,"site":"10"}}',
        ]
        for value in invalid:
            with self.subTest(value=value):
                self.db.calls.clear()
                response = self.client.get("/app/forderungen/liste", params={"dashboard_segment": value})
                self.assertEqual(response.status_code, 400)
                self.assertFalse(self.db.calls)

    def test_empty_error_and_invalid_dashboard_states(self):
        response = self.client.get("/app/forderungen/dashboard?page=Standort&metric=anzahl")
        self.assertEqual(response.status_code, 200)
        self.assertIn("Forderungen | Dashboard", response.text)
        self.assertIn('data-dashboard-metric="anzahl"', response.text)
        self.assertEqual(self.client.get("/app/forderungen/dashboard?page=bad").status_code, 422)
        self.db.connection.execute("DELETE FROM dbo.Benutzer_Rechte WHERE Benutzer_ID='tester'")
        data = self.data()
        self.assertEqual(data["count"], 0)
        self.assertEqual(data["amount"], "0.00")
        with (
            patch("app.forderung_dashboard.read_invoices", side_effect=RuntimeError("offline")),
            self.assertLogs("app.routes", level="ERROR"),
        ):
            response = self.client.get("/app/forderungen/dashboard")
        self.assertEqual(response.status_code, 503)
        self.assertIn("nicht geladen", response.text)
        self.assertNotIn("Anzahl der Belege", response.text)


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