from __future__ import annotations from datetime import date from decimal import Decimal from pathlib import Path import sqlite3 import threading import pytest from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development from jarvis_finance.services.performance_scope import set_performance_scope_classification from jarvis_finance.services.raiffeisen_manual_snapshot import ( confirm_raiffeisen_manual_snapshot, preview_raiffeisen_manual_snapshot, ) from jarvis_finance.storage.database import connect, connect_memory from jarvis_finance.storage.migrations import apply_migrations NOW = "2026-08-20T12:00:00+00:00" def database(): conn = connect_memory() apply_migrations(conn) conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('bank','Raiffeisen','bank',?)", (NOW,), ) for account_id, name in ( ("private", "Privatkonto ••••5632"), ("savings", "Sparkonto ••••5031"), ): conn.execute( """INSERT INTO accounts( account_id,platform_id,account_name,account_type,currency,performance_included, is_active,created_at,balance_mode,portfolio_bucket ) VALUES(?, 'bank', ?, 'cash','CHF',0,1,?,'snapshot','cash')""", (account_id, name, NOW), ) for snapshot_id, account_id, value in ( ("old-private", "private", "100.00"), ("old-savings", "savings", "10.00"), ): conn.execute( """INSERT INTO cash_account_snapshots( snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, amount_chf,source,created_at,created_by,semantic_identity ) VALUES(?,?, 'manual_balance','2026-08-20',?,'CHF',?,'test',?,'test',?)""", (snapshot_id, account_id, value, value, NOW, snapshot_id), ) conn.commit() return conn def source_facts(): return { "snapshot_date": date(2026, 8, 21), "private_account_value_chf": Decimal("90.00"), "savings_account_value_chf": Decimal("20.00"), "membership_value_chf": Decimal("5.00"), } def test_preview_is_read_only_and_keeps_targets_separate(): conn = database() before = conn.total_changes preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) assert conn.total_changes == before assert preview["bank_cash_after_chf"] == "110.00" assert preview["separate_membership_asset_after_chf"] == "5.00" assert preview["known_wealth_before_chf"] == "110.00" assert preview["expected_known_wealth_after_chf"] == "115.00" assert preview["expected_total_wealth_change_chf"] == "5.00" assert preview["creates_transactions"] is False assert [row["account_label"] for row in preview["affected_accounts"]] == [ "Bankkonto ••••5632", "Bankkonto ••••5031", "Raiffeisen Genossenschaftsanteil", ] assert preview["affected_accounts"][2]["previous_status"] == "not_created" def test_preview_uses_confirmed_cash_movements_after_latest_snapshot(): conn = database() conn.execute( """INSERT INTO transactions( transaction_id,transaction_type,account_id,trade_date,net_amount_original, currency_original,fx_rate_to_chf,net_amount_chf,source_type,is_confirmed, quality_status,created_at,is_voided ) VALUES('movement','cash_movement','private','2026-08-21','5','CHF','1','5', 'household_csv',1,'ok',?,0)""", (NOW,), ) conn.commit() preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) private = preview["affected_accounts"][0] assert private["previous_value_chf"] == "105.00" assert preview["bank_cash_after_chf"] == "110.00" assert preview["known_wealth_before_chf"] == "115.00" assert preview["expected_total_wealth_change_chf"] == "0.00" assert preview["expected_known_wealth_after_chf"] == "115.00" def test_confirm_is_append_only_audited_and_idempotent(): conn = database() preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) with pytest.raises(ValueError, match="confirmation_token_mismatch"): confirm_raiffeisen_manual_snapshot( conn, **source_facts(), preview_id=f"raiffeisen-preview-{'0' * 32}", confirmation_id=f"raiffeisen-confirm-{'0' * 32}", input_fingerprint=preview["input_fingerprint"], ) request = { **source_facts(), "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "input_fingerprint": preview["input_fingerprint"], } result = confirm_raiffeisen_manual_snapshot(conn, **request) replay = confirm_raiffeisen_manual_snapshot(conn, **request) assert result["status"] == "confirmed" assert result["created_snapshot_count"] == 3 assert result["created_transaction_count"] == 0 assert result["known_wealth_after_chf"] == "115.00" assert replay["status"] == "already_applied" assert conn.execute("SELECT COUNT(*) FROM transactions").fetchone()[0] == 0 assert conn.execute("SELECT COUNT(*) FROM cash_account_snapshots WHERE source='manual_screenshot_snapshot'").fetchone()[0] == 2 assert conn.execute("SELECT COUNT(*) FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot'").fetchone()[0] == 1 membership = conn.execute( "SELECT account_type,portfolio_bucket FROM accounts WHERE lower(account_name) LIKE '%genossenschaft%'" ).fetchone() assert dict(membership) == {"account_type": "other_asset", "portfolio_bucket": "other"} assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_id=?", (preview["confirmation_id"],)).fetchone()[0] == 1 with pytest.raises(ValueError, match="confirmation_id_reused_with_different_input"): confirm_raiffeisen_manual_snapshot( conn, **{**request, "membership_value_chf": Decimal("6.00")}, ) with pytest.raises(sqlite3.IntegrityError, match="manual cash snapshots are immutable"): conn.execute( "UPDATE cash_account_snapshots SET amount_chf='999.00' WHERE source='manual_screenshot_snapshot'" ) conn.rollback() with pytest.raises(sqlite3.IntegrityError, match="manual asset snapshots cannot be deleted"): conn.execute( "DELETE FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot'" ) conn.rollback() with pytest.raises(sqlite3.IntegrityError, match="manual cash snapshots cannot be replaced"): conn.execute( """INSERT OR REPLACE INTO cash_account_snapshots( snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, amount_chf,source,note,created_at,created_by,audit_id,semantic_identity ) SELECT snapshot_id,account_id,snapshot_type,balance_date,'999.00',currency, '999.00',source,note,created_at,created_by,audit_id,semantic_identity FROM cash_account_snapshots WHERE source='manual_screenshot_snapshot' LIMIT 1""" ) conn.rollback() with pytest.raises(sqlite3.IntegrityError, match="manual asset snapshots cannot be replaced"): conn.execute( """INSERT OR REPLACE INTO account_value_snapshots( snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type, quality_status,notes,created_at,updated_at,valuation_at,source_reference,is_active ) SELECT snapshot_id,account_id,valuation_date,'999.00',currency,source_type, quality_status,notes,created_at,updated_at,valuation_at,source_reference,is_active FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot' LIMIT 1""" ) conn.rollback() with pytest.raises(sqlite3.IntegrityError, match="manual snapshot confirmations cannot be replaced"): conn.execute( """INSERT OR REPLACE INTO manual_snapshot_confirmations( confirmation_id,preview_id,input_fingerprint,payload_hash,snapshot_date,source_kind, known_wealth_after_chf,bank_cash_after_chf,separate_membership_asset_after_chf, created_snapshot_count,created_at,audit_id ) SELECT confirmation_id,preview_id,input_fingerprint,payload_hash,snapshot_date,source_kind, known_wealth_after_chf,bank_cash_after_chf,separate_membership_asset_after_chf, created_snapshot_count,created_at,audit_id FROM manual_snapshot_confirmations LIMIT 1""" ) conn.rollback() def test_confirm_fails_closed_when_baseline_changed(): conn = database() preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) conn.execute( """INSERT INTO cash_account_snapshots( snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency,amount_chf, source,created_at,created_by,semantic_identity ) VALUES('changed','private','manual_balance','2026-08-21','95','CHF','95','test',?,'test','changed')""", (NOW,), ) conn.commit() with pytest.raises(ValueError, match="baseline_changed"): confirm_raiffeisen_manual_snapshot( conn, **source_facts(), preview_id=preview["preview_id"], confirmation_id=preview["confirmation_id"], input_fingerprint=preview["input_fingerprint"], ) def test_confirm_fails_closed_when_cash_movement_changes_preview_projection(): conn = database() preview = preview_raiffeisen_manual_snapshot(conn, **source_facts()) conn.execute( """INSERT INTO transactions( transaction_id,transaction_type,account_id,trade_date,currency_original, net_amount_chf,fx_status,source_type,is_confirmed,quality_status,created_at,updated_at ) VALUES('movement-after-preview','cash','private','2026-08-21','CHF', '3.00','ok','test_manual_adjustment',1,'ok',?,?)""", (NOW, NOW), ) conn.commit() with pytest.raises(ValueError, match="baseline_changed"): confirm_raiffeisen_manual_snapshot( conn, **source_facts(), preview_id=preview["preview_id"], confirmation_id=preview["confirmation_id"], input_fingerprint=preview["input_fingerprint"], ) def test_acceptance_sums_include_unchanged_bank_cash_and_keep_component_correction_after_anchor(): conn = database() conn.execute( "UPDATE cash_account_snapshots SET amount_original='29042.53',amount_chf='29042.53' WHERE account_id='private'" ) conn.execute( "UPDATE cash_account_snapshots SET amount_original='19.84',amount_chf='19.84' WHERE account_id='savings'" ) conn.execute( """INSERT INTO accounts( account_id,platform_id,account_name,account_type,currency,performance_included, is_active,created_at,balance_mode,portfolio_bucket ) VALUES('unchanged-bank','bank','Unchanged bank account','cash','CHF',0,1,?,'snapshot','cash')""", (NOW,), ) conn.execute( """INSERT INTO cash_account_snapshots( snapshot_id,account_id,snapshot_type,balance_date,amount_original,currency, amount_chf,source,created_at,created_by,semantic_identity ) VALUES('unchanged-bank-value','unchanged-bank','manual_balance','2026-08-20', '74900.12','CHF','74900.12','test',?,'test','unchanged-bank-value')""", (NOW,), ) conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,created_at) VALUES('pf','PostFinance','broker',?)", (NOW,), ) conn.execute( """INSERT INTO accounts( account_id,platform_id,account_name,account_type,currency,performance_included, is_active,created_at,balance_mode,portfolio_bucket ) VALUES('pf-depot','pf','PostFinance Depot','brokerage','CHF',0,1,?,'snapshot','equity')""", (NOW,), ) set_performance_scope_classification( conn, account_id="pf-depot", included=True, classification_role="postfinance_etrading_depot", source="test", note="acceptance anchor", classified_at=NOW, ) conn.execute( """INSERT INTO account_value_snapshots( snapshot_id,account_id,valuation_date,total_value_chf,currency,source_type, quality_status,created_at,valuation_at,is_active ) VALUES('pf-anchor','pf-depot','2026-08-26','523788.47','CHF', 'postfinance_official_import','ok',?,'2026-08-26T12:00:00+00:00',1)""", (NOW,), ) conn.commit() facts = { "snapshot_date": date(2026, 8, 27), "private_account_value_chf": Decimal("29059.44"), "savings_account_value_chf": Decimal("19.84"), "membership_value_chf": Decimal("200.00"), } preview = preview_raiffeisen_manual_snapshot(conn, **facts) assert preview["bank_cash_after_chf"] == "103979.40" assert preview["separate_membership_asset_after_chf"] == "200.00" assert preview["known_wealth_before_chf"] == "627750.96" assert preview["expected_total_wealth_change_chf"] == "216.91" assert preview["expected_known_wealth_after_chf"] == "627967.87" confirmed = confirm_raiffeisen_manual_snapshot( conn, **facts, preview_id=preview["preview_id"], confirmation_id=preview["confirmation_id"], input_fingerprint=preview["input_fingerprint"], ) assert confirmed["bank_cash_after_chf"] == "103979.40" assert confirmed["known_wealth_after_chf"] == "627967.87" model = build_modelled_wealth_development(conn, as_of="2026-08-27", period="all") assert model["last_confirmed_anchor_date"] == "2026-08-26" def test_concurrent_identical_confirm_is_one_write_and_one_truthful_replay(tmp_path: Path): source = database() db_path = tmp_path / "concurrent.sqlite3" target = connect(db_path) source.backup(target) source.close() preview = preview_raiffeisen_manual_snapshot(target, **source_facts()) target.close() request = { **source_facts(), "preview_id": preview["preview_id"], "confirmation_id": preview["confirmation_id"], "input_fingerprint": preview["input_fingerprint"], } barrier = threading.Barrier(2) statuses: list[str] = [] failures: list[Exception] = [] def worker() -> None: conn = connect(db_path) conn.execute("PRAGMA busy_timeout=5000") barrier.wait() try: statuses.append(confirm_raiffeisen_manual_snapshot(conn, **request)["status"]) except Exception as exc: # pragma: no cover - asserted empty below failures.append(exc) finally: conn.close() first = threading.Thread(target=worker) second = threading.Thread(target=worker) first.start() second.start() first.join() second.join() assert failures == [] assert sorted(statuses) == ["already_applied", "confirmed"] conn = connect(db_path) assert conn.execute("SELECT COUNT(*) FROM manual_snapshot_confirmations").fetchone()[0] == 1 assert conn.execute( "SELECT COUNT(*) FROM cash_account_snapshots WHERE source='manual_screenshot_snapshot'" ).fetchone()[0] == 2 assert conn.execute( "SELECT COUNT(*) FROM account_value_snapshots WHERE source_type='manual_screenshot_snapshot'" ).fetchone()[0] == 1 conn.close()