from __future__ import annotations import json from pathlib import Path import pytest from jarvis_finance.dashboard import data as dashboard_data from jarvis_finance.dashboard.pages import page_14_import_wizard, page_15_manual_review_queue, page_16_data_quality_center from jarvis_finance.imports.broker_mapping import ( DryRunInput, assert_write_enabled, confirm_instrument_mapping, create_broker_import_dry_run, create_instrument_mapping, create_platform_account_mapping, get_manual_review_queue, ) from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations, utc_now class FakeStreamlit: def __init__(self) -> None: self.calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = [] self.sidebar = self def __getattr__(self, name: str): def recorder(*args, **kwargs): self.calls.append((name, args, kwargs)) if name in {"button", "checkbox"}: return False if name in {"selectbox", "radio"}: options = args[1] if len(args) > 1 else kwargs.get("options", []) return list(options)[0] if options else None if name == "tabs": return [self for _ in (args[0] if args else [])] if name == "columns": count = args[0] if args else 1 return [self for _ in range(count)] if isinstance(count, int) else [self for _ in count] return None return recorder def __enter__(self): return self def __exit__(self, *args): return False def conn(): c = connect_memory() apply_migrations(c) return c def seed_instrument(c, *, instrument_id="inst-demo", isin="CH0000000001", ticker="DGA", exchange="SIX", currency="CHF"): c.execute( "INSERT INTO instruments(instrument_id, asset_class, name, ticker, isin, exchange, currency, created_at) VALUES (?, 'etf', 'Demo ETF', ?, ?, ?, ?, ?)", (instrument_id, ticker, isin, exchange, currency, utc_now()), ) return instrument_id def seed_platform_account(c, *, platform_id="pf", account_id="pf-cash", platform="PostFinance", account="Cash"): now = utc_now() c.execute("INSERT INTO platforms(platform_id, name, platform_type, default_currency, created_at) VALUES (?, ?, 'bank', 'CHF', ?)", (platform_id, platform, now)) c.execute("INSERT INTO accounts(account_id, platform_id, account_name, account_type, currency, created_at) VALUES (?, ?, ?, 'cash', 'CHF', ?)", (account_id, platform_id, account, now)) return platform_id, account_id def test_instrument_mapping_can_be_created_and_auto_maps_by_isin() -> None: c = conn() instrument_id = seed_instrument(c) result = create_instrument_mapping(c, source_platform="PostFinance", source_label="Demo ETF", asset_class="etf", isin="CH0000000001", currency="CHF") row = c.execute("SELECT * FROM instrument_mappings WHERE mapping_id=?", (result["mapping_id"],)).fetchone() assert row["mapping_status"] == "mapped" assert row["instrument_id"] == instrument_id def test_missing_isin_creates_needs_manual_review() -> None: c = conn() result = create_instrument_mapping(c, source_platform="PostFinance", source_label="Demo ETF", asset_class="etf", ticker="DGA", currency="CHF") assert result["mapping_status"] == "needs_manual_review" assert "missing_isin" in result["quality_flags"] def test_name_without_isin_or_ticker_is_not_automapped() -> None: c = conn() result = create_instrument_mapping(c, source_platform="True Wealth", source_label="Global Index Fund", asset_class="etf", currency="USD") assert result["mapping_status"] == "needs_manual_review" assert result["instrument_id"] is None assert "name_only_not_auto_mapped" in result["quality_flags"] def test_platform_account_mapping_for_postfinance_and_true_wealth_and_raiffeisen_cash() -> None: c = conn() pf_platform, pf_account = seed_platform_account(c, platform_id="pf", account_id="pf-depot", platform="PostFinance", account="Depot") pf = create_platform_account_mapping(c, source_platform="PostFinance", source_account_label="Depot", account_type="brokerage", currency="CHF", internal_platform_id=pf_platform, internal_account_id=pf_account) tw = create_platform_account_mapping(c, source_platform="True Wealth", source_account_label="Robo Portfolio", account_type="robo_portfolio", currency="CHF") rf = create_platform_account_mapping(c, source_platform="Raiffeisen", source_account_label="Privatkonto", account_type="brokerage", currency="CHF") assert pf["mapping_status"] == "mapped" assert tw["mapping_status"] == "needs_manual_review" assert rf["account_type"] == "cash" assert "raiffeisen_instrument_detail_missing" in rf["quality_flags"] def test_broker_import_dry_run_stores_summary_without_productive_positions() -> None: c = conn() dry_run_id = create_broker_import_dry_run(c, DryRunInput(source_platform="PostFinance", source_file_type="docx", source_filename="real-file-redacted.docx", snapshot_date_status="missing", rows_total=10, candidate_positions=3, candidate_cash_rows=2, mapped_positions=1, blocked_positions=2, warnings_count=4, quality_flags={"missing_isin": 2}, summary={"rows": [{"forbidden": "raw"}], "safe": "aggregate"})) row = c.execute("SELECT * FROM broker_import_dry_runs WHERE dry_run_id=?", (dry_run_id,)).fetchone() assert row["candidate_positions"] == 3 assert json.loads(row["summary_json"]) == {"safe": "aggregate"} assert c.execute("SELECT COUNT(*) AS n FROM positions_snapshot").fetchone()["n"] == 0 def test_import_wizard_manual_review_and_data_quality_pages_import_and_render() -> None: c = conn() create_instrument_mapping(c, source_platform="PostFinance", source_label="Unknown ETF", asset_class="etf", currency="USD") st = FakeStreamlit() for module in [page_14_import_wizard, page_15_manual_review_queue, page_16_data_quality_center]: module.render(st, c) assert any(call[0] == "title" for call in st.calls) def test_manual_review_queue_loads_open_mapping_problems() -> None: c = conn() create_instrument_mapping(c, source_platform="PostFinance", source_label="Unknown ETF", asset_class="etf") rows = get_manual_review_queue(c) assert rows assert rows[0]["item_type"] == "instrument_mapping" def test_data_quality_center_reports_missing_isin_fx_and_prices() -> None: c = conn() create_instrument_mapping(c, source_platform="True Wealth", source_label="Unknown USD ETF", asset_class="etf", currency="USD") c.execute("UPDATE instrument_mappings SET quality_flags_json=?", (json.dumps(["missing_fx", "snapshot_only", "cost_basis_uncertain"]),)) rows = dashboard_data.get_data_quality(c) checks = {row["check"]: row["count"] for row in rows} assert checks["fehlende ISIN"] == "1" assert checks["fehlende FX in Import-Reviews"] == "1" assert "fehlende Crypto-Preise" in checks def test_read_only_mode_blocks_commit_actions() -> None: with pytest.raises(PermissionError): assert_write_enabled(write_enabled=False) def test_audit_log_created_when_manual_mapping_confirmed() -> None: c = conn() instrument_id = seed_instrument(c) result = create_instrument_mapping(c, source_platform="PostFinance", source_label="Manual ETF", asset_class="etf", ticker="MAN", currency="CHF") audit_id = confirm_instrument_mapping(c, mapping_id=result["mapping_id"], instrument_id=instrument_id, note="Synthetic manual confirmation") assert audit_id assert c.execute("SELECT COUNT(*) AS n FROM audit_log WHERE entity_id=?", (result["mapping_id"],)).fetchone()["n"] == 1 def test_no_real_files_created_in_repo(tmp_path: Path) -> None: c = conn() create_broker_import_dry_run(c, DryRunInput(source_platform="Raiffeisen", source_file_type="xlsx", source_filename="redacted.xlsx", rows_total=1)) assert not any(tmp_path.rglob("*.xlsx"))