from __future__ import annotations import base64 from io import BytesIO from pathlib import Path from zipfile import ZIP_DEFLATED, ZipFile import pytest from fastapi.testclient import TestClient from pydantic import ValidationError from pypdf import PdfReader, PdfWriter from jarvis_finance.api.schemas.postfinance import PostFinanceImportRequest from jarvis_finance.api.dependencies import get_db from jarvis_finance.api.main import MAX_POSTFINANCE_REQUEST_BYTES, create_app from jarvis_finance.api.routers.postfinance import _safe_error_message from jarvis_finance.imports import postfinance_documents as documents from jarvis_finance.imports.postfinance_documents import ( MAX_FILES, MAX_PDF_BYTES, MAX_PDF_PAGES, MAX_UNCOMPRESSED_BYTES, MAX_ZIP_BYTES, parse_postfinance_bundle, ) from jarvis_finance.services import postfinance_service as service from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations DEPOT = "account_pf_depot_synthetic" CASH = "account_pf_cash_synthetic" def _pdf(lines: list[str]) -> bytes: def escaped(value: str) -> str: return value.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)") commands = ["BT", "/F1 8 Tf", "42 805 Td", "10 TL"] for line in lines: commands.extend((f"({escaped(line)}) Tj", "T*")) commands.append("ET") stream = "\n".join(commands).encode("latin-1") objects = [ b"<< /Type /Catalog /Pages 2 0 R >>", b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>", b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 595 842] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>", b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>", b"<< /Length %d >>\nstream\n" % len(stream) + stream + b"\nendstream", ] output = bytearray(b"%PDF-1.4\n") offsets = [0] for index, obj in enumerate(objects, start=1): offsets.append(len(output)) output.extend(f"{index} 0 obj\n".encode()) output.extend(obj) output.extend(b"\nendobj\n") xref = len(output) output.extend(f"xref\n0 {len(objects) + 1}\n".encode()) output.extend(b"0000000000 65535 f \n") for offset in offsets[1:]: output.extend(f"{offset:010d} 00000 n \n".encode()) output.extend( f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n".encode() ) return bytes(output) def _overview(day: str = "27/07/2026") -> bytes: lines = [ f"Portfolio Stand {day} 09:14:30", "Offene Aufträge 0", "EUR 1.0000 100.00 x x 100.00 x", "USD 1.0000 100.00 x x 100.00 x", "Gesamt CHF 200.00 2'200.00 2'400.00", "Produkt Anzahl Einstandskurs Einstandswert Differenz Kurs x Wert CHF Anteil %", "Aktien", ] for index in range(11): lines.append(f"Synthetic Asset {index + 1:02d} 1 50 50 - 100 CHF x 100 4.17") lines.extend(("Zwischensumme Aktien in CHF 0 0 1'100.00", "ETFs")) for index in range(11, 22): lines.append(f"Synthetic Asset {index + 1:02d} 1 50 50 - 100 CHF x 100 4.17") lines.append("Zwischensumme ETFs in CHF 0 0 1'100.00") return _pdf(lines) def _statement(end: str = "27.07.2026") -> bytes: return _pdf( [ "PostFinance Kontoauszug", f"Kontoauszug vom 01.07.2026 bis {end}", "Kontonummer: 000000", "Keine weiteren Buchungen im synthetischen Testzeitraum.", ] ) def _trade() -> bytes: return _pdf( [ "Transaktionsbeleg", "Börsentransaktion: Kauf", "Unsere Referenz: 10000001", "Kontonummer: 000000", "Ausführungsdatum: 20.07.2026", "Valutadatum: 22.07.2026", "Synthetic Asset 01 ISIN: CH0000000000", "Anzahl Preis Betrag", "1 50 CHF 50", "Betrag belastet CHF 50", ] ) def _portfolio_performance() -> bytes: return _pdf( [ "Portfolio Wertentwicklung", "von 01.07.2026 bis 27.07.2026", "Rein synthetisches Testdokument ohne Personen- oder Vermoegensdaten.", ] ) def _zip(*, statement_end: str = "27.07.2026", duplicate: bool = False, unsupported: bool = False, ambiguous_transfer: bool = False, traversal: bool = False) -> bytes: output = BytesIO() with ZipFile(output, "w") as archive: statement = _statement(statement_end) archive.writestr("Kontoauszug_000000_20260727.pdf", statement) if duplicate: archive.writestr("Kontoauszug_Kopie_000000_20260727.pdf", statement) archive.writestr("Borsenabrechnung_000000_10000001_20260720.pdf", _trade()) archive.writestr("Portfolio-Wertentwicklung_000000_20260727.pdf", _portfolio_performance()) if unsupported: archive.writestr("Unbekannt_000000_20260727.pdf", _pdf(["Unbekannter Dokumenttyp"])) if ambiguous_transfer: archive.writestr( "Transferabrechnung_000000_20000001_20260727.pdf", _pdf( [ "Zahlungsverkehr - Gutschrift", "Unsere Referenz: 20000001", "Kontonummer: 000000", "Valutadatum: 27.07.2026", "Gutgeschriebener Betrag CHF 10", ] ), ) if traversal: archive.writestr("../Kontoauszug_000000_20260727.pdf", _statement()) return output.getvalue() def _request(zip_raw: bytes, overview_raw: bytes, **overrides: object) -> dict[str, object]: payload: dict[str, object] = { "zip_file_name": "postfinance-documents.zip", "overview_file_name": "postfinance-account-overview.pdf", "zip_mime_type": "application/zip", "overview_mime_type": "application/pdf", "zip_size_bytes": len(zip_raw), "overview_size_bytes": len(overview_raw), "zip_content_base64": base64.b64encode(zip_raw).decode(), "overview_content_base64": base64.b64encode(overview_raw).decode(), } payload.update(overrides) return payload def _conn(): conn = connect_memory() apply_migrations(conn) conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,country,default_currency,is_active,created_at) VALUES('pf-synthetic','PostFinance','bank','CH','CHF',1,'2026-01-01')" ) conn.execute( "INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,performance_included,is_active,created_at,portfolio_bucket) VALUES(?,'pf-synthetic','PostFinance E-Trading','brokerage','CHF',0,1,'2026-01-01','postfinance')", (DEPOT,), ) conn.execute( "INSERT INTO accounts(account_id,platform_id,account_name,account_type,currency,performance_included,is_active,created_at,portfolio_bucket) VALUES(?,'pf-synthetic','PostFinance Cash','cash','CHF',0,1,'2026-01-01','cash')", (CASH,), ) for index in range(22): instrument = f"instrument_pf_synthetic_{index:02d}" label = f"Synthetic Asset {index + 1:02d}" conn.execute( "INSERT INTO instruments(instrument_id,name,ticker,isin,asset_class,currency,is_active,created_at) VALUES(?,?,?,?,?,?,1,'2026-01-01')", (instrument, label, f"S{index:02d}", f"CH{index:010d}", "equity" if index < 11 else "etf", "CHF"), ) conn.execute( """INSERT INTO positions_snapshot(position_snapshot_id,snapshot_date,account_id,platform_id, instrument_id,quantity,market_value_chf,data_quality_status,created_at) VALUES(?, '2026-06-30', ?, 'pf-synthetic', ?, '1', '90', 'complete', '2026-06-30')""", (f"baseline-synthetic-{index}", DEPOT, instrument), ) conn.commit() return conn def _encrypted_pdf(raw: bytes) -> bytes: reader = PdfReader(BytesIO(raw)) writer = PdfWriter() writer.clone_document_from_reader(reader) writer.encrypt("synthetic-password") output = BytesIO() writer.write(output) return output.getvalue() def _blank_pdf() -> bytes: writer = PdfWriter() writer.add_blank_page(width=595, height=842) output = BytesIO() writer.write(output) return output.getvalue() def _image_only_pdf() -> bytes: image = b"\x80" content = b"q 100 0 0 100 50 700 cm /Im1 Do Q" objects = [ b"<< /Type /Catalog /Pages 2 0 R >>", b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>", b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 595 842] /Resources << /XObject << /Im1 4 0 R >> >> /Contents 5 0 R >>", b"<< /Type /XObject /Subtype /Image /Width 1 /Height 1 /ColorSpace /DeviceGray /BitsPerComponent 8 /Length 1 >>\nstream\n" + image + b"\nendstream", b"<< /Length %d >>\nstream\n" % len(content) + content + b"\nendstream", ] output = bytearray(b"%PDF-1.4\n") offsets = [0] for index, obj in enumerate(objects, start=1): offsets.append(len(output)) output.extend(f"{index} 0 obj\n".encode()) output.extend(obj) output.extend(b"\nendobj\n") xref = len(output) output.extend(f"xref\n0 {len(objects) + 1}\n".encode()) output.extend(b"0000000000 65535 f \n") for offset in offsets[1:]: output.extend(f"{offset:010d} 00000 n \n".encode()) output.extend( f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n".encode() ) return bytes(output) def test_actual_synthetic_zip_and_overview_preview_confirm_and_replay_are_isolated(monkeypatch, tmp_path): conn = _conn() zip_raw, overview_raw = _zip(), _overview() request = _request(zip_raw, overview_raw) runtime = tmp_path / "runtime" monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(runtime)) before = {table: conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] for table in ("postfinance_import_batches", "postfinance_documents", "postfinance_ledger_events", "postfinance_snapshots", "audit_log")} preview = service.preview_postfinance_import(conn, request) after_preview = {table: conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] for table in before} assert after_preview == before assert not runtime.exists() assert preview["document_types"] == {"account_statement": 1, "portfolio_performance": 1, "trade_confirmation": 1} assert preview["covered_periods"] assert preview["components"][0]["count"] == 22 assert preview["components"][1]["count"] == 2 assert preview["mappings"] == {"secure": 22, "review_required": 0, "blocked": 0} assert preview["event_counts"] == {"buy": 1} assert preview["confirm_allowed"] is True first = service.confirm_postfinance_import( conn, {**request, "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "confirm": True}, ) replay_preview = service.preview_postfinance_import(conn, request) rows_before_replay = conn.total_changes second = service.confirm_postfinance_import( conn, {**request, "preview_id": replay_preview["preview_id"], "confirmation_id": replay_preview["confirmation_id"], "confirm": True}, ) assert first["idempotent"] is False assert second["idempotent"] is True assert conn.total_changes == rows_before_replay assert conn.execute("SELECT COUNT(*) FROM postfinance_import_batches").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM postfinance_ledger_events").fetchone()[0] == 1 archive = runtime / "imports" / "postfinance" / "archive" assert archive.is_dir() and not any(path.name.startswith(".") for path in archive.iterdir()) assert all(path.stat().st_mode & 0o077 == 0 for path in archive.iterdir()) def test_read_only_api_allows_preview_hides_internal_hashes_and_blocks_confirm(): conn = _conn() app = create_app(write_mode="disabled") def override_db(): yield conn app.dependency_overrides[get_db] = override_db client = TestClient(app) request = _request(_zip(), _overview()) before = conn.total_changes response = client.post("/api/postfinance/imports/preview", json=request) assert response.status_code == 200 assert conn.total_changes == before body = response.json() assert body["source_references"]["upload"].startswith("PF-") assert "bundle_sha256" not in body assert "db_revision" not in body assert DEPOT not in response.text and CASH not in response.text confirm = client.post( "/api/postfinance/imports/confirm", json={**request, "preview_id": body["preview_id"], "confirmation_id": body["confirmation_id"], "confirm": True}, ) assert confirm.status_code == 403 assert conn.total_changes == before secret_marker = "SYNTHETIC_PRIVATE_FILENAME_123" invalid = client.post( "/api/postfinance/imports/preview", json={**request, "zip_file_name": f"{secret_marker}.txt"}, ) assert invalid.status_code == 422 assert invalid.json() == {"detail": "Bitte ein unterstütztes PostFinance-ZIP auswählen."} assert secret_marker not in invalid.text assert request["zip_content_base64"] not in invalid.text oversized = client.post( "/api/postfinance/imports/preview", content=b"{}", headers={ "content-type": "application/json", "content-length": str(MAX_POSTFINANCE_REQUEST_BYTES + 1), }, ) assert oversized.status_code == 413 assert oversized.json() == { "detail": "Upload ist grösser als das sichere Verarbeitungslimit." } missing_length = client.post( "/api/postfinance/imports/preview", content=b"{}", headers={"content-type": "application/json", "content-length": ""}, ) assert missing_length.status_code == 411 assert missing_length.json() == {"detail": "Upload benötigt eine prüfbare Dateigrösse."} def test_unclassified_parser_errors_are_redacted(): marker = "SYNTHETIC_INTERNAL_PARSER_DETAIL" assert marker not in _safe_error_message(marker) @pytest.mark.parametrize( ("case", "zip_raw", "overview_raw", "message"), [ ("missing-overview", _zip(), b"", "leer"), ("corrupt-zip", b"PK\x03\x04broken", _overview(), "valid ZIP"), ("encrypted-overview", _zip(), _encrypted_pdf(_overview()), "Encrypted"), ("scan-overview", _zip(), _image_only_pdf(), "selectable text"), ("unsupported-document", _zip(unsupported=True), _overview(), "Unsupported document"), ("duplicate-document", _zip(duplicate=True), _overview(), "duplicate document"), ("ambiguous-transfer", _zip(ambiguous_transfer=True), _overview(), "pairing is incomplete"), ("path-traversal", _zip(traversal=True), _overview(), "unsafe entry"), ], ) def test_actual_synthetic_invalid_sources_fail_closed(case, zip_raw, overview_raw, message): del case conn = _conn() request = _request(zip_raw, overview_raw) before = conn.total_changes with pytest.raises(ValueError, match=message): service.preview_postfinance_import(conn, request) assert conn.total_changes == before def test_blank_pdf_without_text_and_password_pdf_are_rejected(): with pytest.raises(ValueError, match="selectable text"): parse_postfinance_bundle(_zip(), _blank_pdf()) with pytest.raises(ValueError, match="Encrypted"): parse_postfinance_bundle(_zip(), _encrypted_pdf(_overview())) def test_wrong_type_empty_size_and_path_names_are_rejected_before_parsing(): zip_raw, overview_raw = _zip(), _overview() with pytest.raises(ValidationError): PostFinanceImportRequest.model_validate(_request(zip_raw, overview_raw, zip_file_name="documents.txt")) with pytest.raises(ValidationError): PostFinanceImportRequest.model_validate(_request(zip_raw, overview_raw, zip_mime_type="text/plain")) with pytest.raises(ValueError, match="ZIP und eine separate PDF"): service._source_bytes(_request(zip_raw, overview_raw, zip_file_name="../documents.zip")) with pytest.raises(ValueError, match="leer|Dateigrösse"): service._source_bytes(_request(b"", overview_raw)) def test_file_count_individual_size_unpacked_size_and_outer_size_limits(): overview = _overview() too_many = BytesIO() with ZipFile(too_many, "w") as archive: for index in range(MAX_FILES + 1): archive.writestr(f"Kontoauszug_{index:03d}_20260727.pdf", b"%PDF-test") with pytest.raises(ValueError, match="safe limit"): parse_postfinance_bundle(too_many.getvalue(), overview) oversized_pdf = BytesIO() with ZipFile(oversized_pdf, "w", compression=ZIP_DEFLATED) as archive: archive.writestr("Kontoauszug_000000_20260727.pdf", b"%PDF-" + b"A" * MAX_PDF_BYTES) with pytest.raises(ValueError, match="unsupported entry"): parse_postfinance_bundle(oversized_pdf.getvalue(), overview) oversized_unpacked = BytesIO() repeated = b"A" * MAX_PDF_BYTES with ZipFile(oversized_unpacked, "w", compression=ZIP_DEFLATED) as archive: for index in range(MAX_UNCOMPRESSED_BYTES // MAX_PDF_BYTES + 1): archive.writestr(f"Kontoauszug_{index:03d}_20260727.pdf", repeated) with pytest.raises(ValueError, match="too large after extraction"): parse_postfinance_bundle(oversized_unpacked.getvalue(), overview) writer = PdfWriter() for _ in range(MAX_PDF_PAGES + 1): writer.add_blank_page(width=595, height=842) many_pages = BytesIO() writer.write(many_pages) page_limited = BytesIO() with ZipFile(page_limited, "w", compression=ZIP_DEFLATED) as archive: archive.writestr("Kontoauszug_000000_20260727.pdf", many_pages.getvalue()) with pytest.raises(ValueError, match="safe page limit"): parse_postfinance_bundle(page_limited.getvalue(), overview) with pytest.raises(ValidationError): PostFinanceImportRequest.model_validate( _request(_zip(), overview, zip_size_bytes=MAX_ZIP_BYTES + 1) ) def test_extracted_text_limit_and_filename_content_disagreement_are_rejected(monkeypatch): monkeypatch.setattr(documents, "MAX_PDF_EXTRACTED_CHARS", 10) with pytest.raises(ValueError, match="extracted-text limit"): parse_postfinance_bundle(_zip(), _overview()) output = BytesIO() with ZipFile(output, "w") as archive: archive.writestr( "Kontoauszug_000000_20260727.pdf", _pdf(["Unrelated selectable synthetic text without the expected marker"]), ) monkeypatch.setattr(documents, "MAX_PDF_EXTRACTED_CHARS", 2_000_000) with pytest.raises(ValueError, match="filename and document content disagree"): parse_postfinance_bundle(output.getvalue(), _overview()) def test_incomplete_period_allows_snapshot_confirm_but_keeps_performance_closed(): conn = _conn() zip_raw, overview_raw = _zip(statement_end="26.07.2026"), _overview() request = _request(zip_raw, overview_raw) preview = service.preview_postfinance_import(conn, request) assert preview["confirm_allowed"] is True assert preview["performance_available"] is False assert "portfolio_overview_end_date_mismatch" in preview["performance_blocker_codes"] assert "PostFinance E-Trading-Kontoauszug" in preview["data_gaps"][0] result = service.confirm_postfinance_import( conn, {**request, "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "confirm": True}, ) assert result["status"] == "confirmed" stored = conn.execute( "SELECT performance_coverage_complete FROM postfinance_import_batches WHERE batch_id=?", (result["batch_id"],), ).fetchone() assert stored["performance_coverage_complete"] == 0 assert conn.execute("SELECT COUNT(*) FROM postfinance_snapshots").fetchone()[0] == 1 def test_known_cash_projection_blocks_new_confirm(): conn = _conn() conn.execute( """INSERT INTO audit_log(audit_id,timestamp,source,action,entity_type,entity_id, old_values_json,new_values_json,created_by,created_at) VALUES('old-audit','2026-06-30','synthetic','confirm','batch','old-batch','{}','{}','synthetic','2026-06-30')""" ) conn.execute( """INSERT INTO postfinance_import_batches(batch_id,bundle_sha256,zip_sha256,overview_sha256, parser_id,parser_version,db_revision,archive_reference,snapshot_at,status,audit_id, confirmed_at,confirmed_by) VALUES('old-batch',printf('%064d',1),printf('%064d',2),printf('%064d',3),'old','1','rev','private','2026-06-30','confirmed','old-audit','2026-06-30','synthetic')""" ) conn.execute( """INSERT INTO postfinance_snapshots(snapshot_id,batch_id,account_id,snapshot_at,total_chf, securities_chf,cash_chf,stocks_chf,etfs_chf,component_total_chf,difference_chf, tolerance_chf,position_count,cash_count,open_orders,reconciliation_status,created_at) VALUES('old-snapshot','old-batch',?,'2026-06-30','2400','2200','200','1100','1100','2400','0','0.05',22,2,0,'matched','2026-06-30')""", (DEPOT,), ) conn.execute( """INSERT INTO cash_balances(cash_balance_id,account_id,balance_date,currency,amount_original, fx_rate_to_chf,amount_chf,source_type,quality_status,created_at) VALUES('bad-cash',?,'2026-06-30','CHF','2400','1','2400','postfinance_official_import','complete','2026-06-30')""", (CASH,), ) conn.commit() request = _request(_zip(), _overview()) before = conn.total_changes preview = service.preview_postfinance_import(conn, request) assert preview["confirm_allowed"] is False assert "known_cash_projection_requires_correction" in preview["blocker_codes"] assert conn.total_changes == before correction = service.confirm_known_cash_projection_correction(conn, confirm=True) assert correction["corrected_rows"] == 1 and correction["idempotent"] is False assert service.confirm_known_cash_projection_correction(conn, confirm=True)["idempotent"] is True corrected_preview = service.preview_postfinance_import(conn, request) assert "known_cash_projection_requires_correction" not in corrected_preview["blocker_codes"] assert conn.execute("SELECT COUNT(*) FROM cash_balances WHERE cash_balance_id='bad-cash'").fetchone()[0] == 1 def test_any_same_day_cash_projection_blocks_preview_and_confirm_without_writes(): conn = _conn() conn.execute( """INSERT INTO cash_balances(cash_balance_id,account_id,balance_date,currency,amount_original, fx_rate_to_chf,amount_chf,source_type,quality_status,created_at) VALUES('manual-cash',?,'2026-07-27','CHF','200','1','200','manual','complete','2026-07-27')""", (CASH,), ) conn.commit() request = _request(_zip(), _overview()) before = conn.total_changes preview = service.preview_postfinance_import(conn, request) assert preview["conflict"] is True assert preview["confirm_allowed"] is False with pytest.raises(ValueError, match="conflicts|gesperrt"): service.confirm_postfinance_import( conn, { **request, "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "confirm": True, }, ) assert conn.total_changes == before def test_revision_binds_account_role_and_instrument_mapping_state(): conn = _conn() accounts = service._resolve_accounts(conn) before = service._revision(conn, accounts) swapped_cash_roles = { **accounts, "efinance": accounts["etrading_cash"], "etrading_cash": accounts["efinance"], } assert service._revision(conn, swapped_cash_roles) != before conn.execute( "UPDATE accounts SET account_name='Changed Synthetic Cash Label' WHERE account_id=?", (accounts["etrading_cash"],), ) assert service._revision(conn, accounts) != before conn.execute( "UPDATE instruments SET name='Changed Synthetic Label' WHERE instrument_id='instrument_pf_synthetic_00'" ) assert service._revision(conn, accounts) != before def test_upload_contract_limits_are_explicit_and_runtime_is_outside_repository(tmp_path): zip_raw, overview_raw = _zip(), _overview() validated = PostFinanceImportRequest.model_validate(_request(zip_raw, overview_raw)) assert validated.zip_size_bytes == len(zip_raw) assert validated.overview_size_bytes == len(overview_raw) runtime = tmp_path / "quarantine" repo = Path(__file__).resolve().parents[2] assert runtime.resolve() != repo and repo not in runtime.resolve().parents