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


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

        self.db.connection.executescript("""
            ALTER TABLE dbo.Benutzer_Rolle ADD Benutzer_Email TEXT;
            UPDATE dbo.Benutzer_Rolle SET Benutzer_Email=' me@example.com ' WHERE Benutzer_ID='tester';
            INSERT INTO dbo.Benutzer_Rolle (Benutzer_ID, Benutzer_Email) VALUES ('tester', 'me@example.com');
            INSERT INTO dbo.Benutzer_Rolle (Benutzer_ID, Benutzer_Email) VALUES ('other', 'wrong-user@example.com');
            CREATE TABLE dbo.Forderungen_Liste (Client_DB TEXT, Beleg_Nr TEXT, Verursacher_Email TEXT);
            INSERT INTO dbo.Forderungen_Liste VALUES ('1', 'R1', 'assignee@example.com');
            INSERT INTO dbo.Forderungen_Liste VALUES ('2', 'R1', 'wrong-tenant@example.com');
        """)
        execute = self.db.execute

        def details_and_queries(statement, values=None):
            if 'AS [Kunde_offen_gesamt]' in str(statement):
                result = Mock()
                result.mappings.return_value.first.return_value = dict(
                    Verursacher_Aktuell_Benutzer_ID='assignee', Kunde_Email='', Vers_Email='')
                return result
            return execute(statement, values)

        token = identity('dw', 'sample', '1')
        with patch.object(self.db, 'execute', side_effect=details_and_queries), patch('app.documents.lookup_directory_user') as lookup:
            response = self.client.get(self.base + '/recipients/' + token)
            self.assertEqual(response.status_code, 200)
            self.assertIn('An mich: me@example.com', response.text)
            self.assertIn('Verursacher: assignee@example.com', response.text)
            for target, expected in [('self', 'me@example.com'), ('colleague', 'assignee@example.com')]:
                response = self.client.get(self.base + '/draft/' + token, params={'recipient': target})
                self.assertEqual(response.status_code, 200)
                message = BytesParser(policy=policy.default).parsebytes(response.content)
                self.assertEqual(message['To'], expected)
                self.assertEqual(len(list(message.iter_attachments())), 1)
            lookup.assert_not_called()


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(username="tester", email="self@example.com")))
        db = Mock()
        db.execute.return_value.mappings.return_value.first.side_effect = [dict(
            Verursacher_Aktuell_Benutzer_ID="override",
            Kunde_Email="customer@example.com",
            Vers_Email="insurance@example.com",
        ), None]
        db.execute.return_value.mappings.return_value.fetchall.return_value = []
        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"))

    def test_missing_or_invalid_database_emails_keep_valid_session_fallback(self):
        request = SimpleNamespace(state=SimpleNamespace(principal=SimpleNamespace(username='tester', email='session@example.com')))
        for database_emails in [[], [{'Benutzer_Email': None}], [{'Benutzer_Email': 'invalid'}],
                                [{'Benutzer_Email': 'invalid'}, {'Benutzer_Email': 'db@example.com'}]]:
            with self.subTest(emails=database_emails):
                db = Mock()
                db.execute.return_value.mappings.return_value.first.side_effect = [dict(Kunde_Email='', Vers_Email=''), None]
                db.execute.return_value.mappings.return_value.fetchall.return_value = database_emails
                targets = recipients(request, '1_R1', db)
                expected = 'db@example.com' if len(database_emails) == 2 else 'session@example.com'
                self.assertEqual(targets[0]['email'], expected)
                self.assertEqual(targets[1]['email'], '')