From 31d1c11dcf397ddf6a23e98d6d9062f96bab0194 Mon Sep 17 00:00:00 2001 From: gitea-impl-agent Date: Wed, 7 Oct 2026 22:09:02 +0000 Subject: [PATCH] Reject invalid balances on account updates --- .../application/accounts_write_services.py | 13 +++- .../test_import_current_balance_authority.py | 72 +++++++++++++++++++ 2 files changed, 82 insertions(+), 3 deletions(-) diff --git a/jarca_ledger_mcp/application/accounts_write_services.py b/jarca_ledger_mcp/application/accounts_write_services.py index 9d3e9b1..6cc90ce 100644 --- a/jarca_ledger_mcp/application/accounts_write_services.py +++ b/jarca_ledger_mcp/application/accounts_write_services.py @@ -48,8 +48,15 @@ def update_ledger_account(account_id: str, db_path: str | None = None, **fields: if current is None: return fail("not_found", "account not found") balance_update = "last_balance" in fields and fields.get("last_balance") is not None basis_update = "last_balance_basis_date" in fields and fields.get("last_balance_basis_date") is not None - if basis_update and (not balance_update or fields.get("last_balance") == ""): + if basis_update and not balance_update: return fail("validation_error", "last_balance_basis_date requires last_balance") + normalized_balance = None + if balance_update: + try: + from .account_valuation_service import _strict_int + normalized_balance = _strict_int(fields["last_balance"], "last_balance", allow_none=False) + except (ValueError, TypeError) as e: + return fail("validation_error", str(e)) normalized_basis = None if basis_update: try: @@ -64,7 +71,7 @@ def update_ledger_account(account_id: str, db_path: str | None = None, **fields: if k == "statement_status": v = normalize_statement_status(str(v)) if k == "liquidity_bucket": v = normalize_liquidity_bucket(str(v)) if k == "active": v = int(bool(v)) - if k == "last_balance" and v != "": v = int(v) + if k == "last_balance": v = normalized_balance if k == "currency": v = str(v).upper() if k == "last_balance_basis_date": v = normalized_basis except Exception as e: return fail("validation_error", str(e)) @@ -83,7 +90,7 @@ def update_ledger_account(account_id: str, db_path: str | None = None, **fields: stored_date = str(current["last_balance_basis_date"] or "")[:10] incoming_date = normalized_basis if basis_update else "" historical = bool(balance_update and incoming_date and stored_date and incoming_date < stored_date) - candidate_balance = fields.get("last_balance", current["last_balance"]) + candidate_balance = normalized_balance if balance_update else current["last_balance"] if historical: kept=[(u,p) for u,p in zip(updates,params) if not u.startswith(("last_balance=","last_balance_basis_date=","last_synced_at="))] updates=[u for u,_ in kept]; params=[p for _,p in kept] diff --git a/tests/test_import_current_balance_authority.py b/tests/test_import_current_balance_authority.py index 1b9a32e..31e00a4 100644 --- a/tests/test_import_current_balance_authority.py +++ b/tests/test_import_current_balance_authority.py @@ -121,6 +121,78 @@ def test_create_rejects_whitespace_or_non_integer_dated_balance_without_mutation assert service.list_account_valuations(baseline["account"]["id"], db_path=path)["valuations"] == before_valuations +def test_update_rejects_bool_or_float_dated_balance_without_mutation(tmp_path): + path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account( + "baseline", + last_balance=900, + last_balance_basis_date="2026-07-01", + db_path=path, + )["account"]["id"] + before = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] + before_valuations = service.list_account_valuations(account, db_path=path)["valuations"] + + for invalid_balance in (True, False, 1.2, -1.2): + rejected = service.update_ledger_account( + account, + last_balance=invalid_balance, + last_balance_basis_date="2026-07-02", + db_path=path, + ) + + assert rejected["success"] is False + assert rejected["error_code"] == "validation_error" + assert service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] == before + assert service.list_account_valuations(account, db_path=path)["valuations"] == before_valuations + + +def test_update_rejects_empty_undated_balance_without_mutation(tmp_path): + path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account( + "baseline", + last_balance=900, + last_balance_basis_date="2026-07-01", + db_path=path, + )["account"]["id"] + before = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] + before_valuations = service.list_account_valuations(account, db_path=path)["valuations"] + + rejected = service.update_ledger_account(account, last_balance="", db_path=path) + + assert rejected["success"] is False + assert rejected["error_code"] == "validation_error" + assert service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] == before + assert service.list_account_valuations(account, db_path=path)["valuations"] == before_valuations + + +def test_update_preserves_zero_and_signed_integer_balance_semantics(tmp_path): + path = str(tmp_path / "ledger.sqlite3") + account = service.create_ledger_account("baseline", db_path=path)["account"]["id"] + + zero = service.update_ledger_account( + account, + last_balance=" +0 ", + last_balance_basis_date="2026-07-01", + db_path=path, + ) + signed = service.update_ledger_account( + account, + last_balance="-12", + last_balance_basis_date="2026-07-02", + db_path=path, + ) + + assert zero["success"] is True + assert zero["account"]["last_balance"] == 0 + assert signed["success"] is True + assert signed["account"]["last_balance"] == -12 + assert signed["account"]["last_balance_basis_date"] == "2026-07-02" + assert [(item["basis_date"], item["amount"]) for item in service.list_account_valuations(account, db_path=path)["valuations"]] == [ + ("2026-07-01", 0), + ("2026-07-02", -12), + ] + + def test_zero_dated_balance_and_valid_import_recovery(tmp_path, monkeypatch): path = str(tmp_path / "ledger.sqlite3") zero = service.create_ledger_account(