from __future__ import annotations import sqlite3 import pytest from jarvis_finance.services.budget_accounts import confirm_create_budget_account from jarvis_finance.services.budget_categories import confirm_create_category from jarvis_finance.services.budget_imports import seed_credit_card_candidates_from_rows from jarvis_finance.services.budget_planning import ( confirm_annual_budget_plan, get_annual_budget_assistant, get_budget_planning_matrix, preview_annual_budget_plan, preview_budget_planning_excel_template, ) from jarvis_finance.services.budget_recurring import calculate_annual_and_reserve from jarvis_finance.services.budget_transactions import confirm_budget_transaction from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations, get_schema_version def db(): conn = connect_memory() apply_migrations(conn) return conn def setup(conn): account = confirm_create_budget_account(conn, {"name": "Haushalt", "account_type": "checking", "currency": "CHF"})["entity_id"] expense = confirm_create_category(conn, {"name": "Wohnen", "category_type": "expense"})["entity_id"] income = confirm_create_category(conn, {"name": "Lohn", "category_type": "income"})["entity_id"] return account, expense, income def book(conn, account: str, category: str, payee: str, amount: str, tx_type: str = "expense", date: str = "2026-05-12"): return confirm_budget_transaction(conn, {"account_id": account, "transaction_type": tx_type, "transaction_date": date, "description": payee, "payee": payee, "amount_original": amount, "currency_original": "CHF", "category_id": category, "source_type": "manual"}) def test_migration_49_is_additive_and_supports_immutable_budget_versions() -> None: conn = db() assert get_schema_version(conn) == 53 recurring_columns = {row["name"] for row in conn.execute("PRAGMA table_info(budget_recurring_payments)")} plan_columns = {row["name"] for row in conn.execute("PRAGMA table_info(budget_plan_items)")} assert {"planning_cadence", "planning_type", "due_months_json", "data_version"} <= recurring_columns assert {"planning_cadence", "item_type", "calculation_basis", "manual_override"} <= plan_columns assert conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='budget_plan_versions'").fetchone() def test_quarterly_and_yearly_payments_are_never_multiplied_by_twelve() -> None: from decimal import Decimal assert calculate_annual_and_reserve(Decimal("1298.75"), "quarterly") == (Decimal("5195.00"), Decimal("432.9166666666666666666666667")) assert calculate_annual_and_reserve(Decimal("120.00"), "yearly") == (Decimal("120.00"), Decimal("10.00")) def test_xls_control_calculation_corrects_quarterly_mortgage_and_monthly_rest() -> None: preview = preview_budget_planning_excel_template(db(), {}) assert preview["productive_mutation"] is False assert preview["confirm_available"] is False assert preview["controls"] == { "annual_expenses_chf": "114569.39", "annual_income_chf": "148775.40", "annual_surplus_chf": "34206.01", "monthly_surplus_chf": "2850.50", "mortgage_payment_chf": "1298.75", "mortgage_annual_chf": "5195.00", "mortgage_monthly_reserve_chf": "432.92", "source_monthly_rest_chf": "1984.67", } assert {item["code"] for item in preview["conflicts"]} == {"quarterly_as_monthly", "monthly_rest_wrong"} def test_plan_actual_and_forecast_are_separate_and_confirm_creates_immutable_version() -> None: conn = db() account, expense, income = setup(conn) positions = [ {"name": "Lohn", "position_type": "income", "category_id": income, "payment_amount_chf": "10000.00", "cadence": "monthly", "due_months": [], "certainty": "safe"}, {"name": "Hypothek", "position_type": "fixed_cost", "category_id": expense, "payment_amount_chf": "1298.75", "cadence": "quarterly", "due_months": [3, 6, 9, 12], "certainty": "safe", "manual_override": True}, {"name": "Ferien", "position_type": "one_time_seasonal", "category_id": expense, "payment_amount_chf": "5000.00", "cadence": "one_time", "due_months": [8], "certainty": "variable"}, {"name": "ETF", "position_type": "savings_investment", "payment_amount_chf": "500.00", "cadence": "monthly", "due_months": [], "certainty": "safe"}, ] preview = preview_annual_budget_plan(conn, {"year": "2026", "positions": positions}) assert preview["review"] == {"planned_income_chf": "120000.00", "planned_expense_chf": "16195.00", "planned_surplus_chf": "103805.00"} result = confirm_annual_budget_plan(conn, preview["payload"]) assert result["version_number"] == 1 again = confirm_annual_budget_plan(conn, preview["payload"]) assert again["entity_id"] == result["entity_id"] with pytest.raises(sqlite3.IntegrityError): conn.execute("UPDATE budget_plan_versions SET version_number=2 WHERE version_id=?", (result["entity_id"],)) version_item = conn.execute( "SELECT version_item_id FROM budget_plan_version_items WHERE version_id=? LIMIT 1", (result["entity_id"],), ).fetchone() with pytest.raises(sqlite3.IntegrityError): conn.execute( "UPDATE budget_plan_version_items SET name='changed' WHERE version_item_id=?", (version_item["version_item_id"],), ) with pytest.raises(sqlite3.IntegrityError): conn.execute( "DELETE FROM budget_plan_version_items WHERE version_item_id=?", (version_item["version_item_id"],), ) with pytest.raises(sqlite3.IntegrityError): conn.execute( """INSERT INTO budget_plan_version_items( version_item_id,version_id,position_type,name,payment_amount_text,cadence, due_months_json,annual_amount_text,monthly_reserve_text,calculation_basis, certainty,manual_override,created_at) VALUES ('extra',?,'fixed_cost','extra','1.00','yearly','[]','1.00','0.08', 'not allowed','safe',0,'2026-01-01')""", (result["entity_id"],), ) with pytest.raises(sqlite3.IntegrityError): conn.execute("UPDATE audit_log SET action='changed' WHERE audit_id=?", (result["audit_id"],)) book(conn, account, income, "Arbeitgeber", "10000.00", tx_type="income", date="2026-01-25") book(conn, account, expense, "Hypothek", "1298.75", date="2026-03-31") assistant = get_annual_budget_assistant(conn, year="2026", current_month="2026-03") assert assistant["plan"]["planned_income_chf"] == "120000.00" assert assistant["actual"]["income_chf"] == "10000.00" assert assistant["forecast"]["income_chf"] is None assert assistant["forecast"]["expense_chf"] is None matrix = get_budget_planning_matrix(conn, year="2026", current_month="2026-03") housing = next(row for row in matrix["rows"] if row["category_id"] == expense) assert housing["forecast_display_chf"] is None assert housing["budget_year_chf"] == "10195.00" assert housing["forecast_basis"].startswith("Noch nicht verlässlich") assert len(assistant["summary_kpis"]) == 3 assert assistant["planning_steps"] == [] assert "monthly_close" not in assistant assert "closing" not in assistant assert assistant["optimization_hints"] == [] assert assistant["month"]["month"] == "2026-03" assert assistant["sections"] == {} def test_recurring_hints_do_not_block_the_household_year_rest() -> None: conn = db() account, expense, _income = setup(conn) for _ in range(3): book(conn, account, expense, "Apple", "20.00") for _ in range(4): book(conn, account, expense, "Netflix", "22.90", date="2026-05-09") conn.execute( "INSERT INTO platforms(platform_id,name,platform_type,created_at) " "VALUES ('synthetic-card-platform','Synthetic','bank','2026-01-01')" ) conn.execute( """INSERT INTO accounts( account_id,platform_id,account_name,account_type,currency,portfolio_bucket,created_at) VALUES ('synthetic-card','synthetic-card-platform','Synthetic card','credit_card', 'CHF','liability','2026-01-01')""" ) conn.execute( """INSERT INTO budget_accounts( budget_account_id,linked_account_id,name,account_type,currency,created_at,updated_at) VALUES ('synthetic-card','synthetic-card','Synthetic card','credit_card','CHF', '2026-01-01','2026-01-01')""" ) conn.execute( """INSERT INTO household_account_source_mappings( mapping_id,contract_version,source_type,source_reference_hash,budget_account_id, canonical_account_id,reference_hint,is_active,created_at,updated_at) VALUES ('synthetic-card-map','household_import_v1','visa_credit_card',?, 'synthetic-card','synthetic-card','synthetic',1,'2026-01-01','2026-01-01')""", ("1" * 64,), ) conn.commit() book(conn, "synthetic-card", expense, "Card purchases", "4348.40", date="2026-07-10") seed_credit_card_candidates_from_rows(conn, [{"Datum": "2026-07-20", "Beschreibung": "VISECA Zahlung", "Betrag": "-3238.20"}], source_file_label="card.csv", migros_covered=False) candidate = conn.execute("SELECT transaction_candidate_id FROM budget_transaction_candidates WHERE lower(description) LIKE '%viseca%' LIMIT 1").fetchone() conn.execute("UPDATE budget_transaction_candidates SET classification='credit_card_payment',status='needs_review',merchant='VISECA' WHERE transaction_candidate_id=?", (candidate["transaction_candidate_id"],)) conn.commit() assistant = get_annual_budget_assistant(conn, year="2026", current_month="2026-05") assert assistant["coverage"]["gaps"] == [] assert assistant["coverage"]["status"] == "partial" assert assistant["coverage"]["free_investable_available"] is False assert assistant["forecast"]["free_plannable_chf"] is None assert assistant["forecast"]["expense_chf"] is None assert assistant["forecast"]["is_estimate"] is False def test_uncategorized_confirmed_expense_is_reported_as_a_coverage_gap() -> None: conn = db() account, expense, _income = setup(conn) for month in range(1, 4): book(conn, account, expense, "Categorized", "50.00", date=f"2026-{month:02d}-20") confirm_budget_transaction( conn, { "account_id": account, "transaction_type": "expense", "transaction_date": "2026-04-10", "description": "Uncategorized", "payee": "Uncategorized", "amount_original": "100.00", "currency_original": "CHF", "source_type": "manual", }, ) assistant = get_annual_budget_assistant(conn, year="2026", current_month="2026-04") assert assistant["forecast"]["reliable"] is False assert assistant["forecast"]["expense_chf"] is None assert assistant["coverage"]["status"] == "partial" assert assistant["coverage"]["quality_label"] == "teilweise" assert assistant["coverage"]["category_coverage_percent"] == "60.00" assert assistant["coverage"]["gaps"] == [ { "code": "uncategorized_confirmed_expenses", "message": "Bestätigte Ausgaben sind noch keiner kanonischen Kategorie zugeordnet.", "amount_chf": "100.00", } ] assert assistant["data_quality"]["material_gaps"][0]["code"] == "uncategorized_confirmed_expenses" def test_material_chf_import_gap_returns_understandable_uncertainty() -> None: conn = db() account, expense, _income = setup(conn) conn.execute( """INSERT INTO budget_transactions( budget_transaction_id,account_id,category_id,transaction_date,transaction_type, description,amount_original,currency_original,amount_chf,fx_status,status,source_type, created_at,updated_at) VALUES ('missing-fx',?,?, '2026-05-12','expense','Fremdwährung ohne CHF', '500.00','USD',NULL,'missing','confirmed','import_candidate',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)""", (account, expense), ) conn.commit() assistant = get_annual_budget_assistant(conn, year="2026", current_month="2026-05") assert assistant["forecast"]["reliable"] is False assert assistant["forecast"]["free_plannable_chf"] is None assert assistant["data_quality"]["label"] == "Noch nicht verlässlich berechenbar" assert assistant["data_quality"]["material_gaps"]