from __future__ import annotations from fastapi.testclient import TestClient from jarvis_finance.api.dependencies import get_db from jarvis_finance.api.main import create_app from jarvis_finance.services.budget_seed_review import ( confirm_seed_candidate, ignore_seed_candidate, list_seed_candidates, preview_seed_candidate, seed_budget_candidates_from_rows, ) 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 synthetic_workbook() -> dict[str, list[list[str]]]: return { "Archiv": [["Alt", "Wert"]], "2026": [ ["Kategorie", "Position", "Jan", "Feb", "Jahr", "Typ"], ["Wohnen", "Miete", "100.00", "100.00", "1200.00", "Fixkosten"], ["Essen", "Haushalt", "50.00", "60.00", "660.00", "variabel"], ["", "Unklar ohne Betrag", "", "", "", ""], ], } def test_phase12_schema_has_requested_budget_seed_candidates_table() -> None: conn = db() assert get_schema_version(conn) == 46 cols = {row["name"]: row["type"] for row in conn.execute("PRAGMA table_info(budget_seed_candidates)")} assert { "seed_candidate_id", "source_file_label", "source_sheet", "source_row_or_range", "candidate_type", "source_label", "proposed_category_id", "proposed_parent_label", "proposed_name", "proposed_period_type", "proposed_amount_text", "currency", "confidence", "requires_review", "status", "notes", "created_at", "updated_at", }.issubset(cols) assert cols["proposed_amount_text"].upper() == "TEXT" def test_excel_dry_run_persists_seed_candidates_without_productive_budget_plans() -> None: conn = db() before_plans = conn.execute("SELECT COUNT(*) FROM budget_plan_items").fetchone()[0] summary = seed_budget_candidates_from_rows(conn, synthetic_workbook(), source_file_label="synthetic budget workbook") after_plans = conn.execute("SELECT COUNT(*) FROM budget_plan_items").fetchone()[0] assert after_plans == before_plans assert summary["current_budget_sheet"] == "2026" assert summary["category_candidate_count"] == 2 assert summary["budget_plan_candidate_count"] == 2 assert summary["recurring_candidate_count"] == 1 assert summary["unclean_range_count"] == 1 candidates = list_seed_candidates(conn) assert {c["candidate_type"] for c in candidates} >= {"category", "budget_plan", "recurring_candidate", "unclean_range"} assert all(c["status"] in {"pending", "needs_review"} for c in candidates) assert any(c["proposed_amount_text"] == "100.00" and c["candidate_type"] == "budget_plan" for c in candidates) assert conn.execute("SELECT typeof(proposed_amount_text) FROM budget_seed_candidates WHERE proposed_amount_text IS NOT NULL LIMIT 1").fetchone()[0] == "text" def test_seed_candidate_category_budget_plan_ignore_confirm_and_audit() -> None: conn = db() seed_budget_candidates_from_rows(conn, synthetic_workbook(), source_file_label="synthetic budget workbook") candidates = list_seed_candidates(conn) category_candidate = next(c for c in candidates if c["candidate_type"] == "category" and c["proposed_name"] == "Wohnen") plan_candidate = next(c for c in candidates if c["candidate_type"] == "budget_plan" and c["proposed_name"] == "Miete") unclean_candidate = next(c for c in candidates if c["candidate_type"] == "unclean_range") category_preview = preview_seed_candidate(conn, category_candidate["seed_candidate_id"], {}) assert category_preview["action"] == "accept_category" category_result = confirm_seed_candidate(conn, category_candidate["seed_candidate_id"], category_preview["payload"]) assert category_result["status"] == "confirmed" plan_preview = preview_seed_candidate(conn, plan_candidate["seed_candidate_id"], {"proposed_period_type": "monthly"}) assert plan_preview["action"] == "accept_budget_plan" plan_result = confirm_seed_candidate(conn, plan_candidate["seed_candidate_id"], plan_preview["payload"]) assert plan_result["status"] == "confirmed" row = conn.execute("SELECT source_type, typeof(monthly_amount_chf) FROM budget_plan_items WHERE plan_item_id=?", (plan_result["entity_id"],)).fetchone() assert tuple(row) == ("excel_seed_dry_run", "text") ignored = ignore_seed_candidate(conn, unclean_candidate["seed_candidate_id"], note="synthetic ignore") assert ignored["status"] == "ignored" assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_type IN ('budget_seed_candidate','budget_category','budget_plan_item')").fetchone()[0] >= 3 def test_budget_seed_review_api_lists_previews_confirms_and_ignores() -> None: conn = db() app = create_app() def override_db(): yield conn app.dependency_overrides[get_db] = override_db client = TestClient(app) generate = client.post("/api/budget/seed-candidates/generate", json={"source_file_label": "synthetic", "workbook_rows": synthetic_workbook()}) assert generate.status_code == 200 assert generate.json()["budget_plan_candidate_count"] == 2 listed = client.get("/api/budget/seed-candidates").json() assert len(listed) >= 1 plan_id = next(c["seed_candidate_id"] for c in listed if c["candidate_type"] == "budget_plan") preview = client.post(f"/api/budget/seed-candidates/{plan_id}/preview", json={"proposed_period_type": "monthly"}) assert preview.status_code == 200 confirm = client.post(f"/api/budget/seed-candidates/{plan_id}/confirm", json=preview.json()["payload"]) assert confirm.status_code == 200 unclean_id = next(c["seed_candidate_id"] for c in listed if c["candidate_type"] == "unclean_range") ignored = client.post(f"/api/budget/seed-candidates/{unclean_id}/ignore", json={"note": "not suitable"}) assert ignored.status_code == 200 def income_workbook() -> dict[str, list[list[str]]]: rows = [["Kategorie", "Position", "Jan", "Feb", "Jahr", "Typ"]] rows += [["", "", "", "", "", ""] for _ in range(35)] rows += [ ["", "Lohn Marcel", "100.00", "", "", ""], ["", "Lohn Melanie", "110.00", "", "", ""], ["", "Gratifikation", "", "", "1200.00", ""], ["", "Sonstige Einnahmen", "25.00", "", "", ""], ] return {"2026": rows} def test_ignored_candidate_can_be_reopened_and_audited() -> None: from jarvis_finance.services.budget_seed_review import reopen_seed_candidate conn = db() seed_budget_candidates_from_rows(conn, synthetic_workbook(), source_file_label="synthetic budget workbook") candidate = next(c for c in list_seed_candidates(conn) if c["candidate_type"] == "unclean_range") ignore_seed_candidate(conn, candidate["seed_candidate_id"], note="mistake") result = reopen_seed_candidate(conn, candidate["seed_candidate_id"]) assert result["status"] == "pending" assert conn.execute("SELECT status FROM budget_seed_candidates WHERE seed_candidate_id=?", (candidate["seed_candidate_id"],)).fetchone()[0] == "pending" assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE action='status_change' AND entity_id=?", (candidate["seed_candidate_id"],)).fetchone()[0] == 1 def test_unclean_range_can_be_converted_to_category_and_budget_plan_candidates() -> None: from jarvis_finance.services.budget_seed_review import convert_unclean_range conn = db() seed_budget_candidates_from_rows(conn, synthetic_workbook(), source_file_label="synthetic budget workbook") unclean = next(c for c in list_seed_candidates(conn) if c["candidate_type"] == "unclean_range") cat = convert_unclean_range(conn, unclean["seed_candidate_id"], "category", {"proposed_name": "Sonderfall"}) plan = convert_unclean_range(conn, unclean["seed_candidate_id"], "budget_plan", {"proposed_name": "Sonderbudget", "proposed_amount_text": "42.00"}) assert cat["status"] == "created" assert plan["status"] == "created" created = list_seed_candidates(conn) assert any(c["seed_candidate_id"] == cat["entity_id"] and c["candidate_type"] == "category" for c in created) assert any(c["seed_candidate_id"] == plan["entity_id"] and c["candidate_type"] == "budget_plan" for c in created) def test_income_candidates_from_2026_sheet_are_detected() -> None: conn = db() summary = seed_budget_candidates_from_rows(conn, income_workbook(), source_file_label="income synthetic workbook") candidates = list_seed_candidates(conn) income = [c for c in candidates if c["candidate_type"] == "budget_plan" and "type:income" in (c.get("notes") or "")] assert summary["income_candidate_count"] == 4 assert len(income) == 4 assert {c["proposed_name"] for c in income} >= {"Lohn Marcel", "Lohn Melanie", "Gratifikation", "Sonstige Einnahmen"} assert any(c["requires_review"] for c in income if c["proposed_name"] == "Sonstige Einnahmen") def test_clear_categories_and_budget_plans_can_be_taken_over_idempotently() -> None: from jarvis_finance.services.budget_seed_review import confirm_budget_plan_takeover, confirm_clear_category_takeover, preview_budget_plan_takeover, preview_clear_category_takeover conn = db() seed_budget_candidates_from_rows(conn, synthetic_workbook(), source_file_label="synthetic budget workbook") cat_preview = preview_clear_category_takeover(conn) assert cat_preview["count"] == 2 cat_result = confirm_clear_category_takeover(conn) assert cat_result["confirmed_count"] == 2 plan_preview = preview_budget_plan_takeover(conn) assert plan_preview["count"] >= 2 result1 = confirm_budget_plan_takeover(conn) before = conn.execute("SELECT COUNT(*) FROM budget_plan_items").fetchone()[0] result2 = confirm_budget_plan_takeover(conn) after = conn.execute("SELECT COUNT(*) FROM budget_plan_items").fetchone()[0] assert result1["confirmed_count"] >= 2 assert result2["skipped_count"] >= 2 assert after == before def test_budget_overview_shows_plan_values_without_actual_transactions() -> None: from jarvis_finance.services.budget_overview import get_budget_overview from jarvis_finance.services.budget_seed_review import confirm_budget_plan_takeover conn = db() seed_budget_candidates_from_rows(conn, synthetic_workbook(), source_file_label="synthetic budget workbook") confirm_budget_plan_takeover(conn) overview = get_budget_overview(conn, "2026-01") assert overview["planned_budget_chf"] != "0" assert overview["actual_expense_chf"] == "0.00" assert overview["actuals_label"] == "noch keine Buchungen" assert overview["budget_status"] == "no_actuals" def test_manual_income_and_expense_update_actual_values_and_audit() -> None: from jarvis_finance.services.budget_accounts import confirm_create_budget_account from jarvis_finance.services.budget_overview import get_budget_overview from jarvis_finance.services.budget_transactions import confirm_budget_transaction conn = db() account = confirm_create_budget_account(conn, {"name": "UAT Konto", "account_type": "cash", "currency": "CHF"})["entity_id"] income_cat = conn.execute("SELECT category_id FROM budget_categories WHERE name='Einnahmen'").fetchone()[0] expense_cat = conn.execute("SELECT category_id FROM budget_categories WHERE name='Wohnen'").fetchone()[0] confirm_budget_transaction(conn, {"account_id": account, "transaction_type": "income", "transaction_date": "2026-01-10", "description": "UAT income", "amount_original": "100.00", "currency_original": "CHF", "category_id": income_cat}) confirm_budget_transaction(conn, {"account_id": account, "transaction_type": "expense", "transaction_date": "2026-01-11", "description": "UAT expense", "amount_original": "20.00", "currency_original": "CHF", "category_id": expense_cat}) overview = get_budget_overview(conn, "2026-01") assert overview["actual_income_chf"] == "100.00" assert overview["actual_expense_chf"] == "20.00" assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_type='budget_transaction'").fetchone()[0] == 2