from __future__ import annotations from decimal import Decimal import pytest from fastapi import HTTPException 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 ( get_budget_planning_matrix, get_category_forecast_explanation, ) from jarvis_finance.services.budget_transactions import confirm_budget_transaction from jarvis_finance.services.prior_year_actuals import ( confirm_prior_year_actuals, get_prior_year_actuals, preview_prior_year_actuals, ) from jarvis_finance.storage.database import connect_memory from jarvis_finance.storage.migrations import apply_migrations def db(): conn = connect_memory() apply_migrations(conn) return conn def setup_category(conn, name: str = "Essen & Haushalt"): account_id = confirm_create_budget_account( conn, {"name": "Haushalt", "account_type": "checking", "currency": "CHF"}, )["entity_id"] category_id = confirm_create_category( conn, {"name": name, "category_type": "expense"}, )["entity_id"] return account_id, category_id def book_expense( conn, account_id: str, category_id: str, booked_on: str, amount: str, tx_type: str = "expense", ): return confirm_budget_transaction( conn, { "account_id": account_id, "transaction_type": tx_type, "transaction_date": booked_on, "description": "confirmed booking", "payee": "Provider", "amount_original": amount, "currency_original": "CHF", "category_id": category_id, "source_type": "manual", }, ) def mark_import_coverage(conn, start: str, end: str) -> None: conn.execute( """INSERT INTO budget_import_sessions( import_session_id,file_period_start,file_period_end,status,created_at,updated_at) VALUES (?,?,?,?,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)""", (f"coverage-{start}-{end}", start, end, "candidates_created"), ) conn.commit() def save_prior_year(conn, year: str, values: list[dict[str, str | None]]): preview = preview_prior_year_actuals(conn, {"year": year, "values": values}) return confirm_prior_year_actuals(conn, preview["payload"]) def test_prior_year_batch_keeps_blank_and_explicit_zero_distinct_and_auditable() -> None: conn = db() _account, first = setup_category(conn) second = confirm_create_category( conn, {"name": "Freizeit", "category_type": "expense"} )["entity_id"] result = save_prior_year( conn, "2025", [ {"category_id": first, "amount_chf": "1200.00"}, {"category_id": second, "amount_chf": "0.00"}, ], ) actuals = get_prior_year_actuals(conn, year="2025") by_id = {row["category_id"]: row for row in actuals["rows"]} assert result["status"] == "confirmed" assert by_id[first]["annual_actual_chf"] == "1200.00" assert by_id[first]["monthly_average_chf"] == "100.00" assert by_id[second]["annual_actual_chf"] == "0.00" assert by_id[second]["monthly_average_chf"] == "0.00" assert actuals["totals"]["annual_actual_chf"] == "1200.00" assert actuals["totals"]["monthly_average_chf"] == "100.00" assert actuals["totals"]["recorded_category_count"] == 2 assert actuals["totals"]["missing_category_count"] >= 1 assert conn.execute( "SELECT COUNT(*) FROM audit_log WHERE action='prior_year_actuals_saved'" ).fetchone()[0] == 1 save_prior_year( conn, "2025", [ {"category_id": first, "amount_chf": None}, {"category_id": second, "amount_chf": "0.00"}, ], ) actuals = get_prior_year_actuals(conn, year="2025") by_id = {row["category_id"]: row for row in actuals["rows"]} assert by_id[first]["annual_actual_chf"] is None assert by_id[first]["monthly_average_chf"] is None assert actuals["totals"]["annual_actual_chf"] == "0.00" assert actuals["totals"]["recorded_category_count"] == 1 assert actuals["totals"]["missing_category_count"] >= 1 def test_default_and_additional_past_years_work_but_future_year_is_blocked() -> None: conn = db() _account, category_id = setup_category(conn) assert get_prior_year_actuals(conn)["year"] == "2025" save_prior_year(conn, "2024", [{"category_id": category_id, "amount_chf": "240.00"}]) assert "2024" in get_prior_year_actuals(conn, year="2024")["available_years"] with pytest.raises(HTTPException, match="past year"): get_prior_year_actuals(conn, year="9999") with pytest.raises(HTTPException, match="past year"): preview_prior_year_actuals(conn, {"year": "9" * 5000, "values": []}) with pytest.raises(HTTPException, match="non-negative CHF"): preview_prior_year_actuals( conn, {"year": "2025", "values": [{"category_id": category_id, "amount_chf": "1e999999"}]}, ) def test_forecast_uses_confirmed_actuals_plus_deterministic_rest_only() -> None: conn = db() account_id, category_id = setup_category(conn) mark_import_coverage(conn, "2026-01-01", "2026-04-30") save_prior_year(conn, "2025", [{"category_id": category_id, "amount_chf": "2400.00"}]) for month, amount in enumerate(("100", "200", "300", "400"), start=1): book_expense(conn, account_id, category_id, f"2026-{month:02d}-28", amount) book_expense(conn, account_id, category_id, "2026-05-15", "50.00") book_expense(conn, account_id, category_id, "2026-05-16", "999.00", tx_type="transfer") seed_credit_card_candidates_from_rows( conn, [{"Datum": "2026-05-16", "Beschreibung": "Abo-Kandidat", "Betrag": "999.00"}], source_file_label="visa.csv", migros_covered=False, ) matrix = get_budget_planning_matrix( conn, year="2026", current_month="2026-05", comparison_year="2025", ) row = next(item for item in matrix["rows"] if item["category_id"] == category_id) average = Decimal("250.00") expected = Decimal("1050.00") + average * Decimal(row["remaining_month_parts"]) assert row["covered_complete_months"] == ["01", "02", "03", "04"] assert row["actual_year_to_date_chf"] == "1050.00" assert row["used_monthly_average_chf"] == "250.00" assert row["forecast_display_chf"] == f"{expected.quantize(Decimal('0.01')):.2f}" assert row["comparison_year_actual_chf"] == "2400.00" assert Decimal(row["deviation_chf"]) == Decimal(row["forecast_display_chf"]) - Decimal("2400.00") assert row["deviation_percent"] == f"{(Decimal(row['deviation_chf']) / Decimal('2400') * 100).quantize(Decimal('0.01')):.2f}" assert "comparison_year=2025" in row["data_explorer_url"] assert "Abo" not in row["forecast_basis"] def test_primary_totals_include_unrepresented_expenses_and_withhold_forecast() -> None: conn = db() account_id, category_id = setup_category(conn) mark_import_coverage(conn, "2026-01-01", "2026-03-31") for month in range(1, 4): book_expense(conn, account_id, category_id, f"2026-{month:02d}-20", "50.00") confirm_budget_transaction( conn, { "account_id": account_id, "transaction_type": "expense", "transaction_date": "2026-04-10", "description": "Uncategorized", "payee": "Uncategorized", "amount_original": "100.00", "currency_original": "CHF", "source_type": "manual", }, ) matrix = get_budget_planning_matrix(conn, year="2026", current_month="2026-04") assert matrix["totals"]["actual_year_to_date_chf"] == "250.00" assert matrix["totals"]["unrepresented_actual_chf"] == "100.00" assert matrix["totals"]["unrepresented_transaction_count"] == 1 assert matrix["totals"]["forecast_year_chf"] is None assert matrix["totals"]["deviation_chf"] is None assert matrix["quality_warnings"][-1]["code"] == "expenses_outside_active_categories" def test_partial_import_ranges_never_count_as_complete_forecast_months() -> None: conn = db() account_id, category_id = setup_category(conn) mark_import_coverage(conn, "2026-01-10", "2026-03-20") for month in range(1, 4): book_expense(conn, account_id, category_id, f"2026-{month:02d}-15", "100.00") matrix = get_budget_planning_matrix(conn, year="2026", current_month="2026-04") row = next(item for item in matrix["rows"] if item["category_id"] == category_id) assert row["covered_complete_months"] == [] assert row["forecast_reliable"] is False assert row["forecast_display_chf"] is None def test_forecast_is_missing_until_three_complete_months_and_never_zero_filled() -> None: conn = db() account_id, category_id = setup_category(conn) book_expense(conn, account_id, category_id, "2026-01-31", "100.00") book_expense(conn, account_id, category_id, "2026-02-15", "50.00") matrix = get_budget_planning_matrix(conn, year="2026", current_month="2026-02") row = next(item for item in matrix["rows"] if item["category_id"] == category_id) assert row["forecast_reliable"] is False assert row["forecast_display_chf"] is None assert row["deviation_chf"] is None assert row["deviation_percent"] is None assert row["evaluation"] == "Vergleichswert für 2025 noch nicht erfasst" assert matrix["totals"]["forecast_year_chf"] is None def test_total_percent_comes_from_total_sums_not_category_percentages() -> None: conn = db() account_id, first = setup_category(conn) mark_import_coverage(conn, "2026-01-01", "2026-04-30") second = confirm_create_category( conn, {"name": "Freizeit", "category_type": "expense"} )["entity_id"] comparison_values: list[dict[str, str | None]] = [ { "category_id": str(row["category_id"]), "amount_chf": ( "1000.00" if str(row["category_id"]) == first else "3000.00" if str(row["category_id"]) == second else "0.00" ), } for row in conn.execute( "SELECT category_id FROM budget_categories WHERE is_active=1 AND category_type='expense'" ).fetchall() ] save_prior_year(conn, "2025", comparison_values) for month in range(1, 5): book_expense(conn, account_id, first, f"2026-{month:02d}-28", "100.00") book_expense(conn, account_id, second, f"2026-{month:02d}-28", "200.00") book_expense(conn, account_id, first, "2026-05-31", "100.00") matrix = get_budget_planning_matrix(conn, year="2026", current_month="2026-05") totals = matrix["totals"] expected_percent = ( Decimal(totals["deviation_chf"]) / Decimal(totals["comparison_year_actual_chf"]) * Decimal("100") ).quantize(Decimal("0.01")) assert Decimal(totals["deviation_percent"]) == expected_percent def test_explorer_explanation_reconciles_exactly_with_displayed_forecast() -> None: conn = db() account_id, category_id = setup_category(conn) mark_import_coverage(conn, "2026-01-01", "2026-03-31") save_prior_year(conn, "2025", [{"category_id": category_id, "amount_chf": "1000.00"}]) for month, amount in enumerate(("33.00", "33.00", "34.00"), start=1): book_expense(conn, account_id, category_id, f"2026-{month:02d}-28", amount) book_expense(conn, account_id, category_id, "2026-04-15", "1.00") explanation = get_category_forecast_explanation( conn, category_id=category_id, year="2026", comparison_year="2025" ) reconciled = Decimal(explanation["actual_year_to_date_chf"]) + Decimal( explanation["forecast_remaining_chf"] ) assert reconciled == Decimal(explanation["forecast_year_chf"]) assert explanation["covered_complete_month_count"] == 3 assert explanation["confirmed_transaction_count"] == 4 assert explanation["excluded_transfer_count"] == 0 assert any("Recurring" in item for item in explanation["exclusions"])