Reisacher-Forderung/tests/test_documents.py

import io
import json
import tempfile
import unittest
from email import policy
from email.parser import BytesParser
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock, patch

from app.config import settings
from app.documents import collect, recipients
from app.docuware_export import export
from app.docuware_search import LiveAdapter, MockAdapter
from app.file_storage import save_upload
from tests import test_forderung_history as history_fixture


class DocumentTests(unittest.TestCase):
    setUp = history_fixture.HistoryTests.setUp

    def prepare(self):
        temp = tempfile.TemporaryDirectory()
        self.addCleanup(temp.cleanup)
        self.root = Path(temp.name)
        self.db.connection.executescript(
            "CREATE TABLE dbo.Fahrzeuge (Client_DB TEXT, Fahrzeug_Nr TEXT, Fahrzeug_VIN TEXT); INSERT INTO dbo.Fahrzeuge VALUES ('1','V1','VIN1');"
        )
        config = patch.multiple(
            settings, DOCUWARE_MODE="mock", DOCUWARE_MOCK_PATH="data/docuware-sample", LOCAL_STORAGE=str(self.root)
        )
        config.start()
        self.addCleanup(config.stop)
        self.base = "/app/forderungen/documents/1_R1"

    def test_filters_local_history_and_customer_change(self):
        self.prepare()
        for number in ["R1", "R2", "R3", "H1", "H2", "X1"]:
            save_upload(dict(Client_DB="1", Beleg_Nr=number), number + ".txt", io.BytesIO(b"local"), 100)
        row = dict(Kunde_Nr="D1", Fahrzeug_Nr="V1")
        for scope, expected in [
            ("invoice", {"R1"}),
            ("open", {"R1", "R2"}),
            ("debtor", {"R1", "R2", "R3", "H1", "H2"}),
            ("vehicle", {"R1", "R3", "H1", "X1"}),
        ]:
            docs, _ = collect(self.db, self.pkey, row, scope)
            self.assertEqual({d["number"] for d in docs if d["source"] == "Lokal"}, expected)
        response = self.client.get(self.base + "?scope=vehicle&detail=true")
        self.assertEqual(response.status_code, 200)
        self.assertIn("Fahrzeugdokument", response.text)
        self.assertIn("data-document-preview", response.text)
        self.assertIn("X1.txt", response.text)

    def test_empty_connection_failure_and_targeted_pdf(self):
        self.prepare()
        adapter = MockAdapter()
        docs = adapter.search("1", document_no="R1")
        self.assertEqual(len(docs), 2)
        self.assertEqual(adapter.search("2", document_no="R1"), [])
        self.assertEqual(adapter.search("1", document_no="absent"), [])
        response = self.client.get(self.base)
        from app.documents import identity

        token = identity("dw", "sample", "2")
        response = self.client.get(self.base + "/content/" + token)
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.content, Path("data/docuware-sample/sample.pdf").read_bytes())
        self.assertIn("inline", response.headers["content-disposition"])
        save_upload(self.pkey, "available.txt", io.BytesIO(b"ok"), 100)
        with patch("app.documents.get_adapter") as factory:
            factory.return_value.search.side_effect = ConnectionError()
            with self.assertLogs("app.documents", level="ERROR"):
                response = self.client.get(self.base)
        self.assertIn("Keine Verbindung zu DocuWare", response.text)
        self.assertIn("available.txt", response.text)

    def test_unauthorized_tokens_scopes_and_local_paths(self):
        self.prepare()
        for suffix in ["", "/content/forged", "/recipients/forged", "/draft/forged?recipient=self"]:
            self.assertEqual(self.client.get("/app/forderungen/documents/1_R2" + suffix).status_code, 404)
        self.assertEqual(self.client.get(self.base + "/content/forged").status_code, 404)
        self.assertEqual(self.client.get(self.base + "?scope=invalid").status_code, 422)
        save_upload(dict(Client_DB="2", Beleg_Nr="R1"), "private.pdf", io.BytesIO(b"secret"), 100)
        from app.documents import identity

        token = identity("local", "2_R1/private.pdf")
        self.assertEqual(self.client.get(self.base + "/content/" + token).status_code, 404)

    def test_draft_exact_attachment_and_recipient_validation(self):
        self.prepare()
        from app.documents import identity

        token = identity("dw", "sample", "1")
        targets = [
            dict(id="self", label="An mich", email="me@example.com"),
            dict(id="insurance", label="Versicherung", email=""),
        ]
        with patch("app.documents.recipients", return_value=targets) as lookup:
            response = self.client.get(self.base + "/draft/" + token + "?recipient=self")
            self.assertEqual(response.status_code, 200)
            self.assertEqual(lookup.call_args.args[1], "1_R1")
            message = BytesParser(policy=policy.default).parsebytes(response.content)
            self.assertEqual(message["To"], "me@example.com")
            self.assertEqual(message["X-Unsent"], "1")
            attachments = list(message.iter_attachments())
            self.assertEqual(len(attachments), 1)
            self.assertEqual(attachments[0].get_filename(), "Testdokument-1.pdf")
            self.assertEqual(
                attachments[0].get_payload(decode=True), Path("data/docuware-sample/sample.pdf").read_bytes()
            )
            for recipient in ["insurance", "forged", "me@evil.com"]:
                self.assertEqual(
                    self.client.get(self.base + "/draft/" + token + "?recipient=" + recipient).status_code, 422
                )
            html = self.client.get(self.base + "/recipients/" + token).text
            self.assertIn("disabled", html)

    def test_empty_metadata_dedup_sort_and_upload_refresh(self):
        self.prepare()
        from app.documents import identity

        first = dict(archive="a", id="1", number="R1", date="2026-01-01", name="z.pdf", kind="", editor="")
        second = dict(first, id="2", date="2026-02-01", name="b.pdf")
        with patch("app.documents.get_adapter") as factory:
            factory.return_value.search.return_value = [first, second, first]
            docs, errors = collect(self.db, self.pkey, dict(Kunde_Nr="D1", Fahrzeug_Nr="V1"), "invoice")
        self.assertEqual([d["id"] for d in docs], ["2", "1"])
        self.assertEqual(errors, [])
        with patch("app.documents.get_adapter") as factory:
            factory.return_value.search.return_value = []
            response = self.client.get(self.base)
        self.assertIn("Keine Dokumente für Beleg R1 gefunden", response.text)
        save_upload(self.pkey, "local.pdf", io.BytesIO(b"%PDF-local"), 100)
        token = identity("local", "1_R1/local.pdf")
        response = self.client.get(self.base + "/content/" + token)
        self.assertEqual(response.content, b"%PDF-local")
        self.assertIn("inline", response.headers["content-disposition"])
        panel = (
            __import__("app.jinja_templates", fromlist=["templates"])
            .templates.env.get_template("forderungen/details/files_panel.html")
            .render(forderungen_details=[self.pkey], files_oob=True)
        )
        self.assertIn('hx-swap-oob="outerHTML"', panel)
        self.assertIn('data-document-key="1_R1"', panel)
        self.assertIn('data-document-detail="true"', panel)

    def test_missing_file_and_mock_path_escape(self):
        self.prepare()
        manifest = dict(
            version=1,
            documents=[dict(client_db="1", archive="a", id="1", number="R1", file="../outside.pdf", name="bad.pdf")],
        )
        (self.root / "manifest.json").write_text(json.dumps(manifest))
        with patch.object(settings, "DOCUWARE_MOCK_PATH", str(self.root)):
            with self.assertRaises(FileNotFoundError):
                MockAdapter().download("1", manifest["documents"][0])
            manifest["unavailable"] = True
            (self.root / "manifest.json").write_text(json.dumps(manifest))
            with self.assertRaises(ConnectionError):
                MockAdapter().search("1", document_no="R1")


class AdapterTests(unittest.TestCase):
    def test_live_exact_search_missing_fields_download_and_full_iteration(self):
        document = SimpleNamespace(id=1, title="Test", fields=[SimpleNamespace(name="Belegnummer", value="R1")])
        other = SimpleNamespace(id=2, title="Other", fields=[SimpleNamespace(name="Belegnummer", value="R10")])
        cabinet = Mock(id="archive")
        cabinet.search_dialog.return_value.search.return_value = iter(
            [SimpleNamespace(document=d) for d in [document, other, document]]
        )
        cabinet.get_document.return_value.download.return_value = (b"%PDF-test", "application/pdf", "test.pdf")
        with patch("app.docuware_search.get_file_cabinet", return_value=cabinet):
            adapter = LiveAdapter()
            docs = adapter.search("1", document_no="R1")
            self.assertEqual(len(docs), 2)
            self.assertEqual(docs[0]["date"], "")
            self.assertEqual(adapter.download("1", docs[0])[0], b"%PDF-test")
            cabinet.get_document.assert_called_with("1")

    def test_export_roundtrip_dedup_and_failure_report(self):
        adapter = Mock()
        doc = dict(archive="archive", id="1", number="R1", name="x.pdf", date="", kind="", editor="")
        adapter.search.return_value = [doc, doc]
        adapter.download.return_value = (b"%PDF-test", "application/pdf", "ä.pdf")
        with tempfile.TemporaryDirectory() as folder:
            result = export(adapter, "1", [dict(document_no="R1")], folder)
            self.assertEqual(len(result["documents"]), 1)
            with patch.object(settings, "DOCUWARE_MOCK_PATH", folder):
                mock = MockAdapter()
                docs = mock.search("1", document_no="R1")
                self.assertEqual(mock.download("1", docs[0])[0], b"%PDF-test")
            with self.assertRaises(ValueError):
                export(adapter, "1", [dict(document_no="R1")], folder)
        adapter.download.side_effect = ConnectionError("sensitive-token")
        with tempfile.TemporaryDirectory() as folder:
            result = export(adapter, "1", [dict(document_no="R1")], folder)
            self.assertEqual(result["errors"][0]["stage"], "download")
            self.assertNotIn("sensitive-token", json.dumps(result))

    def test_recipients_from_open_case_and_ad_override(self):
        request = SimpleNamespace(state=SimpleNamespace(principal=SimpleNamespace(email="self@example.com")))
        db = Mock()
        db.execute.return_value.mappings.return_value.first.return_value = dict(
            Verursacher_Aktuell_Benutzer_ID="override",
            Kunde_Email="customer@example.com",
            Vers_Email="insurance@example.com",
        )
        with patch(
            "app.documents.lookup_directory_user",
            return_value=SimpleNamespace(email="colleague@example.com", disabled=False),
        ) as lookup:
            targets = recipients(request, "1_R1", db)
            lookup.assert_called_once_with("override")
            self.assertEqual(
                [t["email"] for t in targets],
                ["self@example.com", "colleague@example.com", "customer@example.com", "insurance@example.com"],
            )
        self.assertEqual(db.execute.call_args.args[1], dict(client_db="1", beleg_nr="R1"))