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, segment_sql
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


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(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 == "Beleg_Tage_offen"
                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 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'"
        )
        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("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()
            return execute(query, values)

        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"]]:
                    raw = parse_qs(urlsplit(item["url"]).query)["dashboard_segment"][0]
                    self.assertTrue(parse_segment(raw)["only_invoices"])
                    response = self.client.get(item["url"])
                    self.assertIn('id="list-count">' + str(item["count"]), response.text)
                    self.assertIn("Nur Rechnungen", 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.assertIn('name="dashboard_segment"', response.text)
                        self.assertIn("Dashboardfilter entfernen", response.text)
                        self.assertIn("Aktive Filter (1)", response.text)
                        from app.jinja_templates import number_format

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

    def test_follow_up_boundaries_and_null_group(self):
        cases = [
            ("2026-11-01", "overdue"),
            ("2026-11-02", "soon"),
            ("2026-11-05 23:30:00", "soon"),
            ("2026-11-06", "not_due"),
            ("2026-11-16 23:30:00", "not_due"),
            ("2026-11-17", "deferred"),
        ]
        for value, expected in cases:
            with self.subTest(date=value):
                self.db.connection.execute(
                    "UPDATE dbo.Forderungen_Details SET Wiedervorlage=? WHERE Beleg_Nr='R1'", (value,)
                )
                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"], expected)
                response = self.client.get(group["segments"][0]["url"])
                self.assertIn('id="list-count">1', response.text)
        ids = [item["id"] for item in data["stack_items"]]
        self.assertEqual(ids, ["overdue", "soon", "not_due", "deferred", "undated"])
        self.assertNotIn("Wiedervorlage", data["views"])

    def test_follow_up_falls_back_to_document_due_date(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='2026-11-17' 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"], "deferred")
        self.assertIn('id="list-count">1', self.client.get(group["segments"][0]["url"]).text)

    def test_follow_up_year_boundary_and_berlin_calendar_day(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")
        selection = parse_segment(parse_qs(urlsplit(group["segments"][0]["url"]).query)["dashboard_segment"][0])
        self.assertEqual(selection["follow_up"], {"from": "2026-12-29", "to": "2027-01-01"})
        self.assertEqual(group["segments"][0]["id"], "soon")

    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"]:
                        selection = parse_segment(parse_qs(urlsplit(segment["url"]).query)["dashboard_segment"][0])
                        self.assertIn("follow_up", selection)
                        self.assertNotIn("department", selection)
                        response = self.client.get(segment["url"])
                        self.assertEqual(response.status_code, 200)
                        self.assertIn('id="list-count">' + str(segment["count"]), 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">2', response.text)
        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_different_user_ids_and_clients_remain_separate(self):
        groups = [group for group in self.data()["views"]["Verursacher"] if group["label"].startswith("Alice (")]
        self.assertEqual(len(groups), 2)
        self.assertNotEqual(groups[0]["label"], groups[1]["label"])
        self.assertNotEqual(groups[0]["url"], groups[1]["url"])
        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">5', partial.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"]
        selection = parse_qs(urlsplit(url).query)["dashboard_segment"][0]
        returned = list_return_url(url, "1_R1")
        self.assertEqual(parse_qs(urlsplit(returned).query)["dashboard_segment"], [selection])
        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] = :ds_client", 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()