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_accounts import confirm_create_budget_account, create_budget_account_preview, list_budget_accounts from jarvis_finance.services.budget_categories import confirm_create_category, create_category_preview, list_categories from jarvis_finance.services.budget_overview import get_budget_overview, get_category_summary, get_monthly_summary from jarvis_finance.services.budget_transactions import ( confirm_budget_transaction, confirm_reverse_budget_transaction, confirm_transfer, create_budget_transaction_preview, create_transfer_preview, list_budget_transactions, reverse_budget_transaction_preview, ) 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 test_budget_phase1_tables_seeds_and_text_decimal_columns_are_created_idempotently() -> None: conn = db() apply_migrations(conn) assert get_schema_version(conn) == 50 tables = {row["name"] for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")} assert { "budget_accounts", "budget_categories", "budget_tags", "budget_transactions", "budget_transaction_tags", "budget_transfers", }.issubset(tables) tx_cols = {row["name"]: row["type"] for row in conn.execute("PRAGMA table_info(budget_transactions)")} assert tx_cols["amount_original"].upper() == "TEXT" assert tx_cols["fx_rate_to_chf"].upper() == "TEXT" assert tx_cols["amount_chf"].upper() == "TEXT" assert conn.execute("SELECT COUNT(*) FROM budget_categories").fetchone()[0] == 13 assert conn.execute("SELECT COUNT(*) FROM budget_tags").fetchone()[0] == 10 def test_budget_accounts_preview_confirm_archive_and_list() -> None: conn = db() preview = create_budget_account_preview(conn, {"name": "Demo Konto", "account_type": "checking", "currency": "CHF"}) assert preview["summary"] == "Demo Konto · checking · CHF" confirmed = confirm_create_budget_account(conn, preview["payload"]) account_id = confirmed["entity_id"] assert confirmed["audit_id"] assert list_budget_accounts(conn)[0]["name"] == "Demo Konto" archived = __import__("jarvis_finance.services.budget_accounts", fromlist=["archive_budget_account"]).archive_budget_account(conn, account_id) assert archived["status"] == "archived" assert conn.execute("SELECT is_active FROM budget_accounts WHERE budget_account_id=?", (account_id,)).fetchone()[0] == 0 def test_budget_categories_tree_create_and_archive() -> None: conn = db() categories = list_categories(conn) parent = next(c for c in categories if c["name"] == "Essen & Haushalt") preview = create_category_preview(conn, {"name": "Drogerie", "category_type": "expense", "parent_category_id": parent["category_id"]}) confirmed = confirm_create_category(conn, preview["payload"]) category_id = confirmed["entity_id"] def contains(items, name): return any(item["name"] == name or contains(item.get("children", []), name) for item in items) assert contains(list_categories(conn), "Drogerie") archived = __import__("jarvis_finance.services.budget_categories", fromlist=["archive_category"]).archive_category(conn, category_id) assert archived["status"] == "archived" def test_manual_income_expense_transfer_reversal_and_overview_write_audit() -> None: conn = db() account_a = confirm_create_budget_account(conn, {"name": "Haushalt", "account_type": "checking", "currency": "CHF"})["entity_id"] account_b = confirm_create_budget_account(conn, {"name": "Reserve", "account_type": "savings", "currency": "CHF"})["entity_id"] category = next(c for c in list_categories(conn) if c["name"] == "Essen & Haushalt")["category_id"] expense_payload = { "account_id": account_a, "transaction_type": "expense", "transaction_date": "2026-05-16", "description": "Synthetische Ausgabe", "amount_original": "12.34", "currency_original": "CHF", "category_id": category, "tag_names": ["Einmalig"], } preview = create_budget_transaction_preview(conn, expense_payload) assert preview["fx_status"] == "not_needed" confirmed = confirm_budget_transaction(conn, preview["payload"]) tx_id = confirmed["entity_id"] assert confirmed["audit_id"] assert conn.execute("SELECT typeof(amount_original) FROM budget_transactions WHERE budget_transaction_id=?", (tx_id,)).fetchone()[0] == "text" income_preview = create_budget_transaction_preview(conn, {**expense_payload, "transaction_type": "income", "description": "Synthetische Einnahme", "amount_original": "20.00", "category_id": None}) assert confirm_budget_transaction(conn, income_preview["payload"])["status"] == "confirmed" fx_preview = create_budget_transaction_preview(conn, {**expense_payload, "currency_original": "EUR", "amount_original": "5.00"}) assert fx_preview["fx_status"] in {"missing", "ok"} transfer_preview = create_transfer_preview(conn, {"from_account_id": account_a, "to_account_id": account_b, "amount_original": "3.00", "currency_original": "CHF", "transaction_date": "2026-05-16", "description": "Synthetischer Transfer"}) transfer = confirm_transfer(conn, transfer_preview["payload"]) assert transfer["status"] == "confirmed" rev_preview = reverse_budget_transaction_preview(conn, tx_id, {"reason": "Synthetische Korrektur"}) assert rev_preview["summary"].startswith("Reversal") reversal = confirm_reverse_budget_transaction(conn, tx_id, rev_preview["payload"]) assert reversal["status"] == "confirmed" assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE entity_type LIKE 'budget_%'").fetchone()[0] >= 5 assert len(list_budget_transactions(conn)) >= 4 assert get_budget_overview(conn)["current_month"] assert get_monthly_summary(conn, "2026-05")["month"] == "2026-05" assert get_category_summary(conn, "2026-05") def test_budget_api_endpoints_preview_confirm_and_overview() -> None: conn = db() app = create_app() def override_db(): yield conn app.dependency_overrides[get_db] = override_db client = TestClient(app) account_payload = {"name": "API Konto", "account_type": "checking", "currency": "CHF"} preview = client.post("/api/budget/accounts/preview", json=account_payload) assert preview.status_code == 200 confirm = client.post("/api/budget/accounts/confirm", json=preview.json()["payload"]) assert confirm.status_code == 200 account_id = confirm.json()["entity_id"] tx_payload = {"account_id": account_id, "transaction_type": "expense", "transaction_date": "2026-05-16", "description": "API Ausgabe", "amount_original": "10.00", "currency_original": "CHF"} tx_preview = client.post("/api/budget/transactions/preview", json=tx_payload) assert tx_preview.status_code == 200 tx_confirm = client.post("/api/budget/transactions/confirm", json=tx_preview.json()["payload"]) assert tx_confirm.status_code == 200 assert client.get("/api/budget/overview").status_code == 200 assert client.get("/api/budget/transactions").status_code == 200 legacy_categories = client.get("/api/budget/categories/summary?month=2026-05") assert legacy_categories.status_code == 200 assert isinstance(legacy_categories.json(), list) category_envelope = client.get("/api/budget/categories/summary-v2?month=2026-05") assert category_envelope.status_code == 200 assert category_envelope.json()["month"] == "2026-05"