Reisacher-Forderung/tests/test_navigation.py

import unittest
from datetime import date, datetime
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
from urllib.parse import parse_qs, urlencode, urlsplit
from zoneinfo import ZoneInfo

from fastapi import FastAPI
from fastapi.testclient import TestClient

import tests  # Initialize temporary logging before importing the application.
from app.config import settings
from app.db import get_session
from app.ldap import DirectoryUser
from app.main import app
from app.navigation import LIST_PATH, list_return_url, normalize_page_url
from app.permissions import Entitlement
from app.routes import router


class NavigationUrlTests(unittest.TestCase):
    def test_return_url_preserves_filters_and_replaces_transport_and_focus(self):
        value = LIST_PATH + "?Filiale=10&Kunde=A%26B&page=3&limit=100&scope=debtor&focus=2_OTHER"
        result = list_return_url(value, "1_R1")
        self.assertEqual(parse_qs(urlsplit(result).query), {"Filiale": ["10"], "Kunde": ["A&B"], "focus": ["1_R1"]})

    def test_invalid_return_destinations_fall_back_to_filtered_page_type(self):
        for value in [
            "https://evil.test/",
            "//evil.test",
            "/\\evil",
            "/%2f%2fevil",
            "/select",
            "/app/forderungen/liste\n",
        ]:
            with self.subTest(value=value):
                self.assertEqual(list_return_url(value, "1_R1"), LIST_PATH + "?focus=1_R1")

    def test_fragment_recovery_uses_only_matching_invoice_context(self):
        return_to = LIST_PATH + "?Filiale=10&focus=1_R1"
        current = "https://external-proxy.test/app/forderungen/details/1_R1?" + urlencode({"return_to": return_to})
        for fragment in ["kommentare", "buchungen"]:
            result = normalize_page_url(
                "/app/forderungen/details/1_R1/" + fragment + "?scope=debtor", current_url=current
            )
            self.assertEqual(result, "/app/forderungen/details/1_R1?" + urlencode({"return_to": return_to}))
        self.assertEqual(
            normalize_page_url("/app/forderungen/details/2_R1/kommentare", current_url=current),
            "/app/forderungen/details/2_R1",
        )

    def test_external_current_url_never_becomes_redirect_origin(self):
        result = normalize_page_url(
            "/app/forderungen/details/1_R1/buchungen",
            current_url="https://evil.test/app/forderungen/details/1_R1?return_to=https://evil.test",
        )
        self.assertTrue(result.startswith("/app/forderungen/details/1_R1?"))
        self.assertNotIn("evil", result)


class LoginRecoveryTests(unittest.TestCase):
    def setUp(self):
        config = patch.multiple(settings, DEBUG_AUTH=False, TRUSTED_PROXY_IPS=[], COOKIE_SECURE=False)
        config.start()
        self.addCleanup(config.stop)
        session = patch("app.auth.SessionLocal", return_value=MagicMock())
        session.start()
        self.addCleanup(session.stop)
        self.client = TestClient(app)
        self.addCleanup(self.client.close)

    def login_target(self, response):
        self.assertEqual(response.status_code, 200)
        login = response.headers["HX-Redirect"]
        self.assertTrue(login.startswith("/login?next="))
        return parse_qs(urlsplit(login).query)["next"][0]

    def test_expired_list_paging_resumes_full_filtered_page(self):
        response = self.client.get(
            LIST_PATH + "?page=3&limit=100&Filiale=10",
            headers={
                "HX-Request": "true",
                "HX-Current-URL": "http://testserver" + LIST_PATH + "?Filiale=10&focus=1_R120",
            },
            follow_redirects=False,
        )
        self.assertEqual(self.login_target(response), LIST_PATH + "?Filiale=10&focus=1_R120")

    def test_fragment_and_save_expiry_resume_detail_without_replaying_post(self):
        return_to = LIST_PATH + "?Filiale=10&Kunde=Muster&focus=1_R1"
        current = "http://testserver/app/forderungen/details/1_R1?" + urlencode({"return_to": return_to})
        headers = {"HX-Request": "true", "HX-Current-URL": current}
        for suffix in ["/kommentare?scope=debtor", "/buchungen?scope=open", ""]:
            with self.subTest(suffix=suffix):
                if suffix:
                    response = self.client.get("/app/forderungen/details/1_R1" + suffix, headers=headers)
                else:
                    response = self.client.post(
                        "/app/forderungen/details/1_R1",
                        headers=headers,
                        data={"source": "details", "return_to": return_to},
                    )
                self.assertEqual(self.login_target(response), current.removeprefix("http://testserver"))

    def test_login_post_normalizes_fragment_next_to_full_get(self):
        self.client.get("/login")
        csrf = self.client.cookies[settings.LOGIN_CSRF_COOKIE_NAME]
        user = DirectoryUser("tester", "CN=tester", "Tester", "")
        rights = (Entitlement("1", "10", "SC", "Mitarbeiter", 100),)
        with (
            patch("app.routes.ldap_authenticate", return_value=user),
            patch("app.routes.load_entitlements", return_value=rights),
        ):
            response = self.client.post(
                "/login",
                data={
                    "username": "tester",
                    "password": "test",
                    "_csrf": csrf,
                    "next": "/app/forderungen/details/1_R1/kommentare?scope=debtor",
                },
                follow_redirects=False,
            )
        self.assertEqual(response.status_code, 303)
        self.assertEqual(response.headers["location"], "/app/forderungen/details/1_R1")

    def test_native_page_request_uses_full_page_redirect(self):
        response = self.client.get(LIST_PATH + "?Filiale=10", follow_redirects=False)
        self.assertEqual(response.status_code, 303)
        self.assertEqual(parse_qs(urlsplit(response.headers["location"]).query)["next"], [LIST_PATH + "?Filiale=10"])


class DetailReturnTests(unittest.TestCase):
    def setUp(self):
        self.db = MagicMock()
        self.db.execute.return_value.first.return_value = (1,)
        application = FastAPI()
        application.include_router(router)
        application.dependency_overrides[get_session] = lambda: self.db

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

        self.client = TestClient(application, raise_server_exceptions=False)
        self.addCleanup(self.client.close)
        self.row = dict(
            Client_DB="1",
            Beleg_Nr="R1",
            Beleg_Text="Rechnung R1",
            Beleg_Offen=10,
            Kunde_offen_gesamt=20,
            Beleg_Datum=date(2026, 10, 1),
            Fahrzeug_Erstzulassung=None,
            Fahrzeug_Kaufdatum=None,
            Wiedervorlage=date(2026, 11, 1),
            Mwst_Betrag=0,
            Mwst_Betrag_berechnet=1.9,
        )
        context = dict(
            forderungen_details=[self.row],
            forderungen_kommentare=[],
            forderungen_buchungen=[],
            auftraege_positionen=[],
            versicherungen=[],
            files=[],
            detail_url="/app/forderungen/details/1_R1",
            kommentare_scope="invoice",
            buchungen_scope="invoice",
            vehicle_missing=False,
            summary=False,
        )
        fixture = patch("app.routes.get_template_context", side_effect=lambda *args: dict(context))
        fixture.start()
        self.addCleanup(fixture.stop)
        self.return_url = LIST_PATH + "?Filiale=10&Kunde=Muster&focus=1_R1"

    def test_detail_customer_total_uses_database_column_alias(self):
        from contextlib import closing
        import sqlite3
        from app.jinja_templates import templates
        sql=templates.env.get_template('forderungen/details/queries/forderungen_details.sql').render()
        projection=next(line.strip().removeprefix(',').strip() for line in sql.splitlines() if 'AS [Kunde_offen_gesamt]' in line)
        with closing(sqlite3.connect(':memory:')) as connection:
            result=connection.execute('SELECT '+projection+' FROM (SELECT 12345.67 AS [Kunde_Gesamt_Offen]) AS [F]')
            self.row.pop('Kunde_offen_gesamt')
            self.row.update(dict(zip([column[0] for column in result.description],result.fetchone())))
        response=self.client.get('/app/forderungen/details/1_R1')
        self.assertEqual(response.status_code,200)
        self.assertIn('12.345,67',response.text)

    def test_detail_displays_missing_amounts_without_claiming_zero(self):
        self.row.update(Beleg_Offen=None,Kunde_offen_gesamt=None,Mwst_Betrag=None,Mwst_Betrag_berechnet=None)
        response=self.client.get('/app/forderungen/details/1_R1')
        self.assertEqual(response.status_code,200)
        self.assertIn('—',response.text)

    def test_detail_and_saved_form_keep_explicit_return_context(self):
        response = self.client.get("/app/forderungen/details/1_R1?" + urlencode({"return_to": self.return_url}))
        self.assertEqual(response.status_code, 200)
        self.assertIn("Filiale=10&Kunde=Muster&focus=1_R1", response.text)
        response = self.client.post(
            "/app/forderungen/details/1_R1",
            data={
                "source": "details",
                "return_to": self.return_url,
            },
            headers={"HX-Request": "true", "Referer": "http://testserver/app/forderungen/details/1_R1"},
        )
        self.assertEqual(response.status_code, 200)
        self.assertIn(
            'name="return_to" value="/app/forderungen/liste?Filiale=10&Kunde=Muster&focus=1_R1"', response.text
        )
        self.assertEqual(self.db.commit.call_count, 1)

    def test_save_and_back_redirects_after_commit_to_full_list(self):
        response = self.client.post(
            "/app/forderungen/details/1_R1",
            data={
                "source": "details",
                "return_to": self.return_url,
                "redirect": "list",
            },
            headers={"HX-Request": "true"},
        )
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.headers["HX-Redirect"], self.return_url)
        self.db.commit.assert_called_once()
        native = self.client.post(
            "/app/forderungen/details/1_R1",
            data={
                "source": "details",
                "return_to": self.return_url,
                "redirect": "list",
            },
            follow_redirects=False,
        )
        self.assertEqual(native.status_code, 303)
        self.assertEqual(native.headers["location"], self.return_url)

    def test_rendered_save_and_back_supplies_all_sql_parameters(self):
        import json
        from html.parser import HTMLParser

        class DetailForm(HTMLParser):
            active = False

            def __init__(self):
                super().__init__()
                self.values = {}
                self.save_back = None
                self.textarea = None

            def handle_starttag(self, tag, attrs):
                attrs = dict(attrs)
                if tag == "form":
                    self.active = attrs.get("id") == "formular"
                if not self.active:
                    return
                if tag == "input" and attrs.get("name"):
                    if attrs.get("type") != "checkbox" or "checked" in attrs:
                        self.values[attrs["name"]] = attrs.get("value", "on")
                if tag == "button" and "hx-vals" in attrs:
                    self.save_back = attrs
                if tag == "textarea" and attrs.get("name"):
                    self.textarea = attrs["name"]
                    self.values[self.textarea] = ''

            def handle_data(self, data):
                if self.textarea:
                    self.values[self.textarea] += data

            def handle_endtag(self, tag):
                if tag == "textarea":
                    self.textarea = None
                if tag == "form":
                    self.active = False

        self.row['Versicherung_Schadennummer'] = 'SCH-123,45'
        self.row['Schaden_Details'] = 'Stoßfänger beschädigt <vorne> & rechts.\nLackierung nötig, 2 Teile.'
        response = self.client.get('/app/forderungen/details/1_R1', params={'return_to': self.return_url})
        self.assertEqual(response.status_code, 200)
        form = DetailForm()
        form.feed(response.text)
        self.assertIsNotNone(form.save_back)
        self.assertNotIn('"', form.save_back)
        submitted = {**form.values, **json.loads(form.save_back['hx-vals'])}
        submitted['Wiedervorlage'] = '2026-11-18'

        def check_bindings(statement, values):
            # Exercise SQLAlchemy's actual required-parameter validation; a plain
            # MagicMock accepted the missing Schadennummer parameter unnoticed.
            statement.compile().construct_params(values)
            return MagicMock(first=lambda: (1,))

        self.db.execute.side_effect = check_bindings
        response = self.client.post('/app/forderungen/details/1_R1', data=submitted, headers={'HX-Request': 'true'})
        self.assertEqual(response.status_code, 200)
        self.db.commit.assert_called_once()
        self.assertEqual(response.headers['HX-Redirect'], self.return_url)
        update = next(call for call in self.db.execute.call_args_list if str(call.args[0]).lstrip().startswith('UPDATE'))
        self.assertEqual(update.args[1]['Schadennummer'], 'SCH-123,45')
        self.assertEqual(update.args[1]['Schaden_Details'], self.row['Schaden_Details'])
        self.assertIn('[Schaden_Details] = :Schaden_Details', str(update.args[0]))
        self.assertEqual(update.args[1]['Wiedervorlage'], date(2026, 11, 18))

    def test_form_saves_stamp_authenticated_user_and_server_time(self):
        for redirect in ('', 'list'):
            with self.subTest(redirect=redirect):
                self.db.reset_mock()
                before = datetime.now(ZoneInfo('Europe/Berlin')).replace(tzinfo=None)
                response = self.client.post('/app/forderungen/details/1_R1', data={
                    'source': 'details', 'redirect': redirect, 'return_to': self.return_url,
                    'Schaden_Details': 'Neue Schadenbeschreibung',
                    'Bearbeitung_Benutzer_ID': 'forged-user',
                    'Bearbeitung_Timestamp': '2000-01-01T00:00:00',
                }, headers={'HX-Request': 'true'})
                after = datetime.now(ZoneInfo('Europe/Berlin')).replace(tzinfo=None)
                self.assertEqual(response.status_code, 200)
                update = next(call for call in self.db.execute.call_args_list
                              if str(call.args[0]).lstrip().startswith('UPDATE'))
                statement, values = update.args
                statement.compile().construct_params(values)
                self.assertIn('[Bearbeitung_Benutzer_ID] = :Bearbeitung_Benutzer_ID', str(statement))
                self.assertIn('[Bearbeitung_Timestamp] = :Bearbeitung_Timestamp', str(statement))
                self.assertEqual(values['Bearbeitung_Benutzer_ID'], 'tester')
                self.assertEqual(values['Schaden_Details'], 'Neue Schadenbeschreibung')
                self.assertIs(type(values['Bearbeitung_Timestamp']), datetime)
                self.assertIsNone(values['Bearbeitung_Timestamp'].tzinfo)
                self.assertLessEqual(before, values['Bearbeitung_Timestamp'])
                self.assertLessEqual(values['Bearbeitung_Timestamp'], after)
                self.db.commit.assert_called_once()
                if redirect:
                    self.assertEqual(response.headers['HX-Redirect'], self.return_url)

    def test_failed_save_does_not_return_and_external_target_is_ignored(self):
        self.db.commit.side_effect = RuntimeError("test database unavailable")
        response = self.client.post(
            "/app/forderungen/details/1_R1",
            data={
                "source": "details",
                "return_to": self.return_url,
                "redirect": "list",
            },
            headers={"HX-Request": "true"},
        )
        self.assertEqual(response.status_code, 500)
        self.assertNotIn("HX-Redirect", response.headers)
        self.db.commit.side_effect = None
        response = self.client.post(
            "/app/forderungen/details/1_R1",
            data={
                "source": "details",
                "redirect": "https://evil.test",
            },
            headers={"HX-Request": "true"},
        )
        self.assertEqual(response.headers["HX-Redirect"], LIST_PATH + "?focus=1_R1")