Reisacher-Forderung/tests/test_forderung_history.py

"""Exercise production SQL predicates on SQLite, adapting only SQL Server TOP syntax.

This does not replace verification against the deployed SQL Server schema.
"""

import re
import sqlite3
import unittest
from datetime import date, datetime
from html.parser import HTMLParser
from unittest.mock import patch
from urllib.parse import parse_qs, urlsplit

from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.auth import AuthenticationMiddleware
from app.config import settings
from app.db import get_session
from app.forderung_history import history_context
from app.oauth_token import create_access_token
from app.permissions import Entitlement
from app.routes import router


class Record(dict):
    def __getattr__(self, name):
        try:
            return self[name]
        except KeyError:
            raise AttributeError(name) from None


class Result:
    def __init__(self, cursor):
        self.rows = []
        if cursor.description:
            for values in cursor.fetchall():
                row = Record(zip((c[0] for c in cursor.description), values))
                if row.get("Timestamp"):
                    row["Timestamp"] = datetime.fromisoformat(row["Timestamp"])
                if row.get("Beleg_Datum"):
                    row["Beleg_Datum"] = date.fromisoformat(row["Beleg_Datum"])
                self.rows.append(row)

    def fetchall(self):
        return self.rows

    def first(self):
        return next(iter(self.rows), None)

    def mappings(self):
        return self


class FixtureDB:
    def __init__(self):
        self.connection = sqlite3.connect(":memory:", check_same_thread=False)
        self.connection.execute("ATTACH DATABASE ':memory:' AS dbo")
        self.calls = []
        self.commits = 0

    def __enter__(self):
        return self

    def __exit__(self, *args):
        pass

    def execute(self, query, values=None):
        sql = str(query)
        self.calls.append((sql, dict(values or {})))
        top = re.search(r"SELECT TOP (\d+)", sql, re.I)
        if top:
            sql = re.sub(r"SELECT TOP \d+", "SELECT", sql, count=1, flags=re.I).rstrip() + " LIMIT " + top[1]
        sql = re.sub(r"\s+WITH \(UPDLOCK, HOLDLOCK\)", "", sql)
        return Result(self.connection.execute(sql, values or {}))

    def commit(self):
        self.commits += 1
        self.connection.commit()

    def rollback(self):
        self.connection.rollback()


def seed(db):
    db.connection.executescript("""
        CREATE TABLE dbo.Forderungen_Kopf (
            Client_DB TEXT, Beleg_Nr TEXT, Kunde_Nr TEXT, Fahrzeug_Nr TEXT, Beleg_Offen REAL,
            Hauptbetrieb_ID TEXT, Standort_ID TEXT, Abteilung TEXT, Kunde_Text TEXT, Beleg_Datum TEXT
        );
        CREATE TABLE dbo.Forderungen_Historie (Client_DB TEXT, Beleg_Nr TEXT, Kunde_Nr TEXT, Fahrzeug_Nr TEXT);
        CREATE TABLE dbo.Forderungen_Buchungen (
            Client_DB TEXT, Kunde_Nr TEXT, Fahrzeug_Nr TEXT, Beleg_Nr TEXT, Bezug_Beleg_Nr TEXT,
            Beleg_Datum TEXT, Beleg_Beschreibung TEXT, Betrag REAL, Benutzer_ID TEXT, Bezahlt TEXT, Beleg_Art TEXT
        );
        CREATE TABLE dbo.Forderungen_Kommentar (
            Client_DB TEXT, Beleg_Nr TEXT, Timestamp TEXT, Benutzer_ID TEXT, Kommentar TEXT,
            Aktuell_Fibu INTEGER DEFAULT 0, Aktuell_Abteilung INTEGER DEFAULT 0
        );
        CREATE TABLE dbo.Benutzer_Rolle (Benutzer_ID TEXT, Name TEXT, Rolle_Name TEXT, Rolle_ID TEXT);
        INSERT INTO dbo.Benutzer_Rolle VALUES ('tester', 'Tester', 'Mitarbeiter', 'MA');
        CREATE VIEW dbo.Forderungen_Kommentar_Benutzer AS
            SELECT *, 'Test User' AS Name, 'MA' AS Rolle_ID, 'Mitarbeiter' AS Rolle_Name
            FROM Forderungen_Kommentar;
        CREATE TABLE dbo.Benutzer_Rechte (
            Benutzer_ID TEXT, Hauptbetrieb_ID TEXT, Standort_ID TEXT, Abteilung TEXT
        );
        INSERT INTO dbo.Benutzer_Rechte VALUES ('tester', '1', '10', 'Service');
    """)
    for tenant, number, customer, vehicle, balance, branch in [
        ("1", "R1", "D1", "V1", 100, "10"),
        ("1", "R2", "D1", "V2", 20, "20"),
        ("1", "R3", "D1", "V1", 0, "10"),
        ("1", "X1", "D2", "V1", 90, "10"),
        ("2", "R1", "D1", "V1", 100, "10"),
        ("1", "NV", "D3", None, 50, "10"),
    ]:
        db.connection.execute(
            "INSERT INTO dbo.Forderungen_Kopf VALUES (?, ?, ?, ?, ?, '1', ?, 'Service', 'Test debtor', '2026-01-01')",
            (tenant, number, customer, vehicle, balance, branch),
        )
    for number, vehicle in [("H1", "V1"), ("H2", "V2"), ("R3", "V1")]:
        db.connection.execute("INSERT INTO dbo.Forderungen_Historie VALUES ('1', ?, 'D1', ?)", (number, vehicle))
    for tenant, customer, vehicle, number, reference in [
        ("1", "D1", "V1", "R1", "R1"),
        ("1", "D1", None, "P1", "R1"),
        ("1", "D1", "V2", "R2", None),
        ("1", "D1", "V1", "R3", None),
        ("1", "D1", "V1", "H1", None),
        ("1", "D1", "V2", "H2", None),
        ("1", "D1", "V1", "LOOSE", None),
        ("1", "D2", "V1", "X1", "R1"),
        ("2", "D1", "V1", "R1", "R1"),
    ]:
        db.connection.execute(
            "INSERT INTO dbo.Forderungen_Buchungen VALUES (?, ?, ?, ?, ?, '2026-01-02', ?, 10, 'tester', 'N', 'RE')",
            (tenant, customer, vehicle, number, reference, "Booking " + number),
        )
    for tenant, number in [("1", "R1"), ("1", "R2"), ("1", "R3"), ("1", "H1"), ("1", "H2"), ("1", "X1"), ("2", "R1")]:
        db.connection.execute(
            "INSERT INTO dbo.Forderungen_Kommentar (Client_DB, Beleg_Nr, Timestamp, Benutzer_ID, Kommentar) "
            "VALUES (?, ?, '2026-01-02T12:00:00', 'tester', ?)",
            (tenant, number, f"Comment {tenant}/{number}"),
        )
    db.connection.commit()


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 HistoryTests(unittest.TestCase):
    def setUp(self):
        self.db = FixtureDB()
        seed(self.db)
        self.addCleanup(self.db.connection.close)
        self.application = FastAPI()
        self.application.include_router(router)
        self.application.dependency_overrides[get_session] = lambda: self.db
        self.application.add_middleware(AuthenticationMiddleware)
        self.entitlements = (Entitlement("1", "10", "Service", "Mitarbeiter", 100),)
        for target, kwargs in [
            ("app.auth.SessionLocal", {"return_value": self.db}),
            ("app.auth.load_entitlements", {"return_value": self.entitlements}),
        ]:
            mocked = patch(target, **kwargs)
            mocked.start()
            self.addCleanup(mocked.stop)
        config = patch.multiple(settings, COOKIE_SECURE=False, TRUSTED_PROXY_IPS=[])
        config.start()
        self.addCleanup(config.stop)
        self.client = TestClient(self.application)
        self.addCleanup(self.client.close)
        self.token = create_access_token(
            username="tester", name="Tester", email="", role="Mitarbeiter", auth_source="ldap", csrf="csrf"
        )
        self.client.cookies.set(settings.SESSION_COOKIE_NAME, self.token)
        self.pkey = {"Client_DB": "1", "Beleg_Nr": "R1"}

    def test_all_scopes_apply_to_real_sql_without_duplicate_bookings(self):
        expected_comments = {
            "invoice": {"R1"},
            "open": {"R1", "R2"},
            "vehicle": {"R1", "R3", "H1"},
            "debtor": {"R1", "R2", "R3", "H1", "H2"},
        }
        expected_bookings = {
            "invoice": {"R1", "P1"},
            "open": {"R1", "P1", "R2"},
            "vehicle": {"R1", "P1", "R3", "H1", "LOOSE"},
            "debtor": {"R1", "P1", "R2", "R3", "H1", "H2", "LOOSE"},
        }
        for section, expectations in [("kommentare", expected_comments), ("buchungen", expected_bookings)]:
            for scope, expected in expectations.items():
                with self.subTest(section=section, scope=scope):
                    rows = history_context(self.db, self.pkey, section, scope)["forderungen_" + section]
                    self.assertEqual({r.Beleg_Nr for r in rows}, expected)
                    self.assertEqual(len(rows), len(expected))
                    if section == "kommentare":
                        self.assertTrue(all(r.Kommentar.startswith("Comment 1/") for r in rows))
                    else:
                        self.assertTrue(all(r.Client_DB == "1" and r.Kunde_Nr == "D1" for r in rows))

    def test_routes_render_isolated_fragments_and_default_to_invoice(self):
        for section, target in [("kommentare", "comments-history"), ("buchungen", "bookings-history")]:
            for scope in ["invoice", "open", "vehicle", "debtor"]:
                with self.subTest(section=section, scope=scope):
                    response = self.client.get(f"/app/forderungen/details/1_R1/{section}?scope={scope}")
                    self.assertEqual(response.status_code, 200)
                    tags = Tags(response.text).tags
                    self.assertIn(("div", next(a for t, a in tags if a.get("id") == target)), tags)
                    buttons = [a for t, a in tags if t == "button"]
                    self.assertEqual(len(buttons), 4)
                    self.assertEqual(sum(b["aria-pressed"] == "true" for b in buttons), 1)
                    self.assertTrue(all(b["hx-target"] == "#" + target for b in buttons))
                    selected = next(b for b in buttons if b["aria-pressed"] == "true")
                    self.assertTrue(selected["hx-get"].endswith("scope=" + scope))
                    self.assertNotIn("<html", response.text)
                    self.assertNotIn('id="formular"', response.text)
                    self.assertNotIn('name="Kommentar"', response.text)
            response = self.client.get(f"/app/forderungen/details/1_R1/{section}")
            self.assertIn('aria-pressed="true"', response.text)
            self.assertNotIn("Booking R2" if section == "buchungen" else "Comment 1/R2", response.text)

    def test_expansion_permits_other_branch_but_direct_access_remains_denied(self):
        expanded = self.client.get("/app/forderungen/details/1_R1/kommentare?scope=debtor")
        self.assertIn("Comment 1/R2", expanded.text)
        for section in ["kommentare", "buchungen"]:
            self.db.calls.clear()
            response = self.client.get(f"/app/forderungen/details/1_R2/{section}?scope=debtor")
            self.assertEqual(response.status_code, 404)
            self.assertEqual(len(self.db.calls), 1)

    def test_invalid_scope_does_not_query_history(self):
        for section in ["kommentare", "buchungen"]:
            self.db.calls.clear()
            response = self.client.get(f"/app/forderungen/details/1_R1/{section}?scope=invalid")
            self.assertEqual(response.status_code, 422)
            self.assertFalse(self.db.calls)

    def test_missing_vehicle_and_empty_results_leave_filters_available(self):
        for section in ["kommentare", "buchungen"]:
            for scope in ["invoice", "vehicle"]:
                response = self.client.get(f"/app/forderungen/details/1_NV/{section}?scope={scope}")
                self.assertEqual(response.status_code, 200)
                buttons = [a for t, a in Tags(response.text).tags if t == "button"]
                vehicle = next(b for b in buttons if b["hx-get"].endswith("scope=vehicle"))
                self.assertIn("disabled", vehicle)
                self.assertIn("Keine Kommentare" if section == "kommentare" else "Keine Buchungsbelege", response.text)
                if scope == "vehicle":
                    self.assertIn("kein Fahrzeug hinterlegt", response.text)

    def test_comment_post_preserves_scope_and_immutable_invoice_identity(self):
        for scope in ["invoice", "open", "vehicle", "debtor"]:
            response = self.client.post(
                "/app/forderungen/details/1_R1",
                data={
                    "source": "comments",
                    "_csrf": "csrf",
                    "comment_scope": scope,
                    "Kommentar": "New comment " + scope,
                    "Client_DB": "2",
                    "Beleg_Nr": "X1",
                    "Benutzer_ID": "intruder",
                },
            )
            self.assertEqual(response.status_code, 200)
            self.assertIn("New comment " + scope, response.text)
            self.assertIn(f'name="comment_scope" value="{scope}"', response.text)
            self.assertNotIn('id="bookings-history"', response.text)
            self.assertNotIn('id="comment-form"', response.text)
        rows = self.db.connection.execute(
            "SELECT Client_DB, Beleg_Nr, Benutzer_ID FROM dbo.Forderungen_Kommentar WHERE Kommentar LIKE 'New comment%'"
        ).fetchall()
        self.assertEqual(rows, [("1", "R1", "tester")] * 4)
        self.assertEqual(self.db.commits, 4)

    def test_post_rejects_invalid_csrf_scope_and_unauthorized_invoice(self):
        for suffix, extra, expected in [
            ("1_R1", {"_csrf": "wrong"}, 403),
            ("1_R1", {"comment_scope": "invalid"}, 422),
            ("1_R2", {}, 404),
        ]:
            data = {"source": "comments", "_csrf": "csrf", "Kommentar": "Rejected"}
            data.update(extra)
            response = self.client.post("/app/forderungen/details/" + suffix, data=data)
            self.assertEqual(response.status_code, expected)
        self.assertEqual(self.db.commits, 0)

    def test_booking_limit_is_enforced_and_disclosed(self):
        self.db.connection.executemany(
            "INSERT INTO dbo.Forderungen_Buchungen VALUES ('1', 'D1', 'V1', ?, 'R1', '2026-02-01', 'Bulk', 1, 'tester', 'N', 'RE')",
            [(f"B{i:04}",) for i in range(1001)],
        )
        rows = history_context(self.db, self.pkey, "buchungen")["forderungen_buchungen"]
        self.assertEqual(len(rows), 1000)
        self.assertEqual(rows[0].Beleg_Nr, "P1")
        response = self.client.get("/app/forderungen/details/1_R1/buchungen")
        self.assertIn("ersten 1.000 Buchungsbelege", response.text)

    def test_comment_order_and_escaping(self):
        self.db.connection.execute(
            "INSERT INTO dbo.Forderungen_Kommentar (Client_DB, Beleg_Nr, Timestamp, Benutzer_ID, Kommentar) "
            "VALUES ('1', 'R1', '2025-01-01T12:00:00', 'tester', '<script>old</script>')"
        )
        response = self.client.get("/app/forderungen/details/1_R1/kommentare?scope=debtor")
        self.assertIn("&lt;script&gt;old&lt;/script&gt;", response.text)
        self.assertLess(response.text.index("&lt;script&gt;old"), response.text.index("Comment 1/R1"))
        self.assertIn("Rechnung H1", response.text)

    def test_full_detail_page_keeps_filters_and_comment_form_when_lists_are_empty(self):
        context = history_context(self.db, self.pkey, "kommentare")
        context.update(history_context(self.db, self.pkey, "buchungen"))
        invoice = dict(context["forderungen_details"][0])
        invoice.update(
            Kunde_offen_gesamt=120,
            Mwst_Betrag=0,
            Mwst_Betrag_berechnet=19,
            Fahrzeug_Erstzulassung=None,
            Fahrzeug_Kaufdatum=None,
            Wiedervorlage=date(2026, 11, 1),
        )
        context.update(
            forderungen_details=[invoice],
            forderungen_kommentare=[],
            forderungen_buchungen=[],
            files=[],
            auftraege_positionen=[],
            versicherungen=[],
            summary=False,
        )
        with patch("app.routes.get_template_context", return_value=context):
            response = self.client.get("/app/forderungen/details/1_R1")
        self.assertEqual(response.status_code, 200)
        tags = Tags(response.text).tags
        selected = [a for t, a in tags if t == "button" and a.get("aria-pressed") == "true"]
        self.assertEqual(len(selected), 2)
        self.assertTrue(all(a["hx-get"].endswith("scope=invoice") for a in selected))
        form = next(a for t, a in tags if a.get("id") == "comment-form")
        self.assertEqual(urlsplit(form["hx-post"]).path, "/app/forderungen/details/1_R1")
        self.assertEqual(parse_qs(urlsplit(form["hx-post"]).query)["return_to"], ["/app/forderungen/liste?focus=1_R1"])
        self.assertEqual(form["hx-target"], "#comments-history")
        self.assertIn("Keine Kommentare", response.text)
        self.assertIn("Keine Buchungsbelege", response.text)


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