from __future__ import annotations from decimal import Decimal from pathlib import Path import pytest from jarvis_finance.config.settings import load_settings from jarvis_finance.dashboard import data as dashboard_data from jarvis_finance.dashboard.demo_data import load_synthetic_demo_data from jarvis_finance.dashboard.pages import ( page_00_command_center, page_01_portfolio, page_04_crypto, page_05_wallets, page_06_ledger, page_09_alerts, page_10_audit, page_11_settings_data_quality, page_12_crypto_manage, ) from jarvis_finance.storage.database import connect_memory from jarvis_finance.crypto.assets import create_crypto_asset from jarvis_finance.crypto.holdings import create_initial_holding_snapshot from jarvis_finance.crypto.wallets import create_wallet from jarvis_finance.market.providers import PriceQuote, store_crypto_price from jarvis_finance.storage.migrations import apply_migrations 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 {"text_input", "text_area", "date_input"}: return "" 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 == "expander": return self if name == "columns": count = args[0] if args else 1 if isinstance(count, int): return [self for _ in range(count)] return [self for _ in count] return None return recorder def __enter__(self): return self def __exit__(self, *args): return False def setup_demo_conn(): conn = connect_memory() apply_migrations(conn) load_synthetic_demo_data(conn) return conn def setup_multi_wallet_crypto_conn(): conn = connect_memory() apply_migrations(conn) w1 = create_wallet(conn, wallet_name="MetaMask Demo", wallet_type="Software Wallet", platform_provider="MetaMask", last_verified_at="2026-01-31T10:00:00Z") w2 = create_wallet(conn, wallet_name="SwissBorg Demo", wallet_type="Exchange", platform_provider="SwissBorg", last_verified_at="2026-01-31T11:00:00Z") eth = create_crypto_asset(conn, coin_name="Ethereum", symbol="ETH", coingecko_id="ethereum") btc = create_crypto_asset(conn, coin_name="Bitcoin", symbol="BTC", coingecko_id="bitcoin") create_initial_holding_snapshot(conn, asset_id=eth, wallet_id=w1, quantity=Decimal("2"), verification_status="verified", last_verified_at="2026-01-31T10:00:00Z", note="synthetic eth split") create_initial_holding_snapshot(conn, asset_id=eth, wallet_id=w2, quantity=Decimal("2"), verification_status="verified", last_verified_at="2026-01-31T11:00:00Z", note="synthetic eth split") create_initial_holding_snapshot(conn, asset_id=btc, wallet_id=w1, quantity=Decimal("0.00000001"), verification_status="stale", last_verified_at="2026-01-01T00:00:00Z", note="synthetic btc tiny") store_crypto_price(conn, asset_id=eth, quote=PriceQuote("ethereum", "CHF", Decimal("2500.00"), provider="synthetic-cache", provider_timestamp="2026-02-01T00:00:00Z")) store_crypto_price(conn, asset_id=btc, quote=PriceQuote("bitcoin", "CHF", Decimal("40000.00"), provider="synthetic-cache", provider_timestamp="2026-02-01T00:00:00Z")) return conn def test_crypto_summary_cards_and_coin_aggregation_across_wallets() -> None: conn = setup_multi_wallet_crypto_conn() summary = dashboard_data.get_crypto_summary_cards(conn) assert summary["coin_count"] == "2" assert summary["wallets_with_holdings"] == "2" assert summary["holding_count"] == "3" assert Decimal(summary["total_value_chf"]) == Decimal("10000.00") + Decimal("0.0004") rows = dashboard_data.get_crypto_coin_summary(conn) eth = next(row for row in rows if row["symbol"] == "ETH") assert eth["coin"] == "Ethereum" assert eth["gesamtmenge"] == "4" assert eth["wallets"] == "2" assert eth["kurs_chf"] == "CHF 2'500.00" assert eth["gesamtwert_chf"] == "CHF 10'000.00" assert eth["anteil"] assert not ({"asset_id", "wallet_id", "dedup_key", "candidate_count", "provider_symbol_status"} & set(eth)) def test_coin_wallet_drilldown_and_wallet_summary_are_user_friendly() -> None: conn = setup_multi_wallet_crypto_conn() eth_row = next(row for row in dashboard_data.get_crypto_coin_summary(conn) if row["symbol"] == "ETH") detail = dashboard_data.get_crypto_coin_wallet_details(conn, eth_row["_asset_id"]) assert [row["wallet"] for row in detail] == ["MetaMask Demo", "SwissBorg Demo"] assert {row["menge"] for row in detail} == {"2"} assert all(row["wert_chf"] == "CHF 5'000.00" for row in detail) assert all(row["anteil"] == "50.00%" for row in detail) assert not ({"asset_id", "wallet_id", "dedup_key", "raw_json"} & set(detail[0])) wallets = dashboard_data.get_wallet_overview_cards(conn) assert wallets["wallet_count"] == "2" assert Decimal(wallets["total_value_chf"]) >= Decimal("10000") wallet_rows = dashboard_data.get_wallet_user_overview(conn) assert wallet_rows[0]["wallet_name"] assert "total_value_chf" in wallet_rows[0] assert "coin_count" in wallet_rows[0] assert "wallet_id" not in wallet_rows[0] def test_user_mode_hides_technical_fields_and_admin_mode_can_show_them() -> None: conn = setup_multi_wallet_crypto_conn() user_rows = dashboard_data.get_crypto_coin_summary(conn, admin_mode=False) admin_rows = dashboard_data.get_crypto_coin_summary(conn, admin_mode=True) assert "asset_id" not in user_rows[0] assert "_asset_id" in user_rows[0] # internal for drilldown, not rendered by user tables assert "asset_id" in admin_rows[0] assert "coingecko_id" in admin_rows[0] def test_command_center_summary_includes_user_overview_fields() -> None: conn = setup_multi_wallet_crypto_conn() summary = dashboard_data.get_command_center_summary(conn) assert "total_portfolio_chf" in summary assert "crypto_total_chf" in summary assert "equity_etf_total_chf" in summary assert "cash_total_chf" in summary assert "critical_alert_count" in summary assert "last_price_update_at" in summary def test_dashboard_modules_import_and_page_render_smoke() -> None: conn = setup_demo_conn() st = FakeStreamlit() for module in [ page_00_command_center, page_01_portfolio, page_04_crypto, page_05_wallets, page_06_ledger, page_09_alerts, page_10_audit, page_11_settings_data_quality, page_12_crypto_manage, ]: module.render(st, conn) assert any(call[0] in {"title", "header", "subheader"} for call in st.calls) def test_dashboard_reader_functions_return_expected_dummy_data() -> None: conn = setup_demo_conn() summary = dashboard_data.get_command_center_summary(conn) assert summary["base_currency"] == "CHF" assert Decimal(summary["cash_total_chf"]) > Decimal("0") assert Decimal(summary["crypto_total_chf"]) > Decimal("0") assert dashboard_data.get_platform_overview(conn) assert dashboard_data.get_ledger_transactions(conn) assert dashboard_data.get_audit_events(conn) def test_crypto_dashboard_reads_cached_db_prices_without_api_calls(monkeypatch: pytest.MonkeyPatch) -> None: def explode(*args, **kwargs): raise AssertionError("dashboard must not call CoinGecko/API provider on render") monkeypatch.setattr("jarvis_finance.market.providers.CoinGeckoClient.get_crypto_price", explode) conn = setup_demo_conn() st = FakeStreamlit() page_04_crypto.render(st, conn) rows = dashboard_data.get_crypto_overview(conn) assert rows assert any(row["latest_price_chf"] for row in rows) def test_dashboard_preserves_decimal_crypto_quantity_format() -> None: conn = setup_demo_conn() wallet_rows = dashboard_data.get_crypto_by_wallet(conn) quantities = {row["quantity"] for row in wallet_rows} assert "0.0899" in quantities for row in wallet_rows: assert "E" not in row["quantity"].upper() def test_alert_lifecycle_data_loaded_readably() -> None: conn = setup_demo_conn() alerts = dashboard_data.get_alerts(conn) assert alerts assert {"status", "priority", "occurrence_count", "last_seen_at", "fingerprint", "dedup_key"} <= set(alerts[0]) def test_dashboard_runtime_path_inside_repo_remains_blocked(tmp_path: Path) -> None: repo = tmp_path / "repo" repo.mkdir() (repo / "pyproject.toml").write_text("[project]\nname='demo'\n") with pytest.raises(ValueError, match="Runtime directory must be outside"): load_settings(repo_root=repo, environ={"JARVIS_FINANCE_RUNTIME_DIR": str(repo / "runtime")})