Reisacher-Forderung/tests/test_navigation.py

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

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_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")