291 lines
20 KiB
Python
291 lines
20 KiB
Python
"""Public-API regressions for frozen settlement-draft policy consistency."""
|
|
import json
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from jarca_ledger_mcp import service as s
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def ok(value):
|
|
assert value["success"], value
|
|
return value
|
|
|
|
|
|
def draft(path, month="2026-09"):
|
|
return ok(s.create_settlement_draft(month, "synthetic creator", "draft-key", db_path=path))["draft"]
|
|
|
|
|
|
def frozen_items(path, did):
|
|
with sqlite3.connect(path) as db:
|
|
return db.execute("SELECT source_id, evidence_json FROM settlement_draft_items WHERE draft_id=? ORDER BY source_id", (did,)).fetchall()
|
|
|
|
|
|
def approve(path, d, **flags):
|
|
r = ok(s.review_settlement_draft(d["id"], db_path=path))
|
|
ok(s.approve_settlement_draft(d["id"], d["version"], r["review_fingerprint"], "reviewer", "synthetic approval", "ref", db_path=path, **flags))
|
|
return r
|
|
|
|
|
|
@pytest.mark.parametrize("debts", [(-40,), (40,), (40, -10, -15), (10, -40)])
|
|
def test_signed_manual_loan_totals_remain_consistent_across_close_and_reports(tmp_path, debts):
|
|
path = str(tmp_path / "signed-loans.sqlite3")
|
|
init_db(path)
|
|
ids = [ok(s.create_manual_asset(f"debt {i}", amount, "2026-09-30", "loan", db_path=path))["manual_asset"]["id"] for i, amount in enumerate(debts)]
|
|
ok(s.create_manual_asset("asset", 100, "2026-09-30", "real_estate", db_path=path))
|
|
with sqlite3.connect(path) as db:
|
|
before = db.execute("SELECT id, valuation_amount, currency, source_note, memo, valuation_date FROM manual_assets ORDER BY id").fetchall()
|
|
d = draft(path)
|
|
frozen_before = frozen_items(path, d["id"])
|
|
review = ok(s.review_settlement_draft(d["id"], db_path=path))
|
|
liabilities = abs(sum(debts)); net = 100 - liabilities
|
|
for totals in (review["totals"], review):
|
|
assert totals["total_assets"] == 100
|
|
assert totals["total_liabilities"] == liabilities
|
|
assert totals["net_worth"] == net
|
|
ok(s.approve_settlement_draft(d["id"], d["version"], review["review_fingerprint"], "reviewer", "synthetic review", "ref", db_path=path))
|
|
closed = ok(s.finalize_settlement_draft(d["id"], d["version"], review["review_fingerprint"], "close-key", db_path=path))
|
|
for totals in (closed["totals"], closed):
|
|
assert totals["total_assets"] == 100 and totals["total_liabilities"] == liabilities and totals["net_worth"] == net
|
|
lookup = ok(s.get_settlement_draft(d["id"], db_path=path))
|
|
assert lookup["final_response"]["totals"] == closed["totals"]
|
|
replay = s.finalize_settlement_draft(d["id"], d["version"], review["review_fingerprint"], "close-key", db_path=path)
|
|
assert {k:v for k,v in replay.items() if k != "already_existing"} == {k:v for k,v in closed.items() if k != "already_existing"}
|
|
report = ok(s.get_asset_snapshot_report("2026-09-30", db_path=path))
|
|
assert (report["total_assets"], report["total_liabilities"], report["net_worth"]) == (100, liabilities, net)
|
|
assert report["close_metadata"]["draft_id"] == d["id"]
|
|
point = ok(s.get_net_worth_timeseries("2026-09-01", "2026-09-30", db_path=path))["points"][0]
|
|
assert (point["total_assets"], point["total_liabilities"], point["net_worth"]) == (100, liabilities, net)
|
|
monthly = ok(s.get_monthly_settlement_summary("2026-09", db_path=path))
|
|
assert monthly["source_coverage"]["assets"]["draft_id"] == d["id"]
|
|
with sqlite3.connect(path) as db:
|
|
after = db.execute("SELECT id, valuation_amount, currency, source_note, memo, valuation_date FROM manual_assets ORDER BY id").fetchall()
|
|
assert after == before and len(ids) == len(debts)
|
|
assert frozen_items(path, d["id"]) == frozen_before
|
|
assert "evidence_json" not in closed
|
|
|
|
|
|
def test_lowercase_supported_currency_normalizes_without_mutating_frozen_evidence(tmp_path):
|
|
path = str(tmp_path / "lowercase-currency.sqlite3")
|
|
init_db(path)
|
|
asset = ok(s.create_manual_asset("Synthetic KRW asset", 25, "2026-09-30", "real_estate", currency="krw", db_path=path))["manual_asset"]["id"]
|
|
with sqlite3.connect(path) as db:
|
|
raw_before = db.execute("SELECT valuation_amount, currency, source_note, memo FROM manual_assets WHERE id=?", (asset,)).fetchone()
|
|
assert raw_before[1] == "krw"
|
|
d = draft(path)
|
|
r = ok(s.review_settlement_draft(d["id"], db_path=path))
|
|
item = next(x for x in r["items"] if x["source_id"] == asset)
|
|
assert item["currency"] == "KRW" and item["amount"] == 25
|
|
assert not r["missing_items"] and not any(w.get("code") == "non_krw" for w in r["warnings"])
|
|
ok(s.approve_settlement_draft(d["id"], d["version"], r["review_fingerprint"], "reviewer", "approve", "ref", db_path=path))
|
|
closed = ok(s.finalize_settlement_draft(d["id"], d["version"], r["review_fingerprint"], "close", db_path=path))
|
|
frozen = dict(frozen_items(path, d["id"]))
|
|
assert json.loads(frozen[asset])["source"]["currency"] == "krw"
|
|
assert "evidence_json" not in closed
|
|
assert closed["totals"]["total_assets"] == closed["total_assets"] == 25
|
|
assert ok(s.get_settlement_draft(d["id"], db_path=path))["final_response"] == closed
|
|
assert ok(s.finalize_settlement_draft(d["id"], d["version"], r["review_fingerprint"], "close", db_path=path)) == closed
|
|
assert ok(s.get_asset_snapshot_report("2026-09-30", db_path=path))["total_assets"] == 25
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT valuation_amount, currency, source_note, memo FROM manual_assets WHERE id=?", (asset,)).fetchone() == raw_before
|
|
|
|
|
|
def test_verified_cash_balance_can_be_explicitly_reselected_without_losing_authority(tmp_path):
|
|
path = str(tmp_path / "cash-reselection.sqlite3")
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account("Synthetic bank", type="bank", liquidity_bucket="cash", db_path=path))["account"]["id"]
|
|
val = ok(s.register_account_valuation(account, "2026-09-30", 20, valuation_kind="cash_balance", verification_status="verified", idempotency_key="cash", db_path=path))["valuation"]["id"]
|
|
d = draft(path)
|
|
r = ok(s.review_settlement_draft(d["id"], db_path=path))
|
|
row = next(x for x in r["included_items"] if x["source_id"] == account)
|
|
assert row["valuation_id"] == val and row["amount"] == 20
|
|
selected = ok(s.set_settlement_draft_item(d["id"], "account", account, d["version"], "selector", valuation_id=val, db_path=path))["draft"]
|
|
r2 = ok(s.review_settlement_draft(selected["id"], db_path=path))
|
|
row2 = next(x for x in r2["included_items"] if x["source_id"] == account)
|
|
assert row2["valuation_id"] == val and row2["amount"] == 20
|
|
ok(s.approve_settlement_draft(selected["id"], selected["version"], r2["review_fingerprint"], "reviewer", "approve", "ref", db_path=path))
|
|
final = ok(s.finalize_settlement_draft(selected["id"], selected["version"], r2["review_fingerprint"], "cash-close", db_path=path))
|
|
assert final["totals"]["total_assets"] == 20
|
|
assert ok(s.get_settlement_draft(selected["id"], db_path=path))["final_response"]["totals"] == final["totals"]
|
|
|
|
|
|
@pytest.mark.parametrize("mode,when,flags", [("exact", "2026-09-30", {}), ("carried_before", "2026-09-29", {"allow_carried_before": True}), ("manual_substitute", "2026-10-02", {"allow_manual_substitute": True})])
|
|
@pytest.mark.parametrize("kind", ["balance", "cash_balance"])
|
|
def test_bank_cash_balance_selection_modes_preserve_explicit_authority(tmp_path, mode, when, flags, kind):
|
|
path = str(tmp_path / (mode + ".sqlite3"))
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account("Synthetic bank", type="bank", liquidity_bucket="cash", db_path=path))["account"]["id"]
|
|
val = ok(s.register_account_valuation(account, when, 20, valuation_kind=kind, verification_status="verified", idempotency_key="cash", db_path=path))["valuation"]["id"]
|
|
d = draft(path)
|
|
kwargs = {"selection_mode": mode, "selection_reason": "dated bank statement"} if mode != "exact" else {}
|
|
d = ok(s.set_settlement_draft_item(d["id"], "account", account, d["version"], "selector", valuation_id=val, db_path=path, **kwargs))["draft"]
|
|
r = ok(s.review_settlement_draft(d["id"], db_path=path))
|
|
row = next(x for x in r["included_items"] if x["source_id"] == account)
|
|
assert row["valuation_id"] == val and row["amount"] == 20
|
|
frozen_before = frozen_items(path, d["id"])
|
|
assert row["valuation_date"] == when
|
|
ok(s.approve_settlement_draft(d["id"], d["version"], r["review_fingerprint"], "reviewer", "accepted dated evidence", "ref", db_path=path, **flags))
|
|
final = ok(s.finalize_settlement_draft(d["id"], d["version"], r["review_fingerprint"], "mode-close", db_path=path))
|
|
assert final["totals"]["total_assets"] == final["total_assets"] == 20
|
|
assert final["contains_substitutions"] == (mode == "manual_substitute")
|
|
assert final["contains_carried"] == (mode == "carried_before")
|
|
assert frozen_items(path, d["id"]) == frozen_before
|
|
assert ok(s.get_settlement_draft(d["id"], db_path=path))["final_response"] == final
|
|
assert ok(s.finalize_settlement_draft(d["id"], d["version"], r["review_fingerprint"], "mode-close", db_path=path)) == final
|
|
|
|
|
|
def test_invalid_future_exact_and_unverified_bank_candidate_cannot_supply_authority(tmp_path):
|
|
path = str(tmp_path / "invalid-bank.sqlite3")
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account("Synthetic bank", type="bank", liquidity_bucket="cash", db_path=path))["account"]["id"]
|
|
future = ok(s.register_account_valuation(account, "2026-10-02", 20, valuation_kind="cash_balance", verification_status="verified", idempotency_key="future", db_path=path))["valuation"]["id"]
|
|
d = draft(path)
|
|
bad = s.set_settlement_draft_item(d["id"], "account", account, d["version"], "selector", valuation_id=future, selection_mode="exact", db_path=path)
|
|
assert not bad["success"] and bad["error_code"] == "validation_error"
|
|
assert ok(s.get_settlement_draft(d["id"], db_path=path))["draft"]["version"] == d["version"]
|
|
|
|
|
|
@pytest.mark.parametrize("currency,amount,expected", [("USD", 25, None), ("KRW", 0, 0), ("KRW", None, None), (" krw ", 25, None)])
|
|
def test_manual_currency_and_missing_value_boundaries_match_legacy_policy(tmp_path, currency, amount, expected):
|
|
path = str(tmp_path / "manual-boundary.sqlite3")
|
|
init_db(path)
|
|
asset = ok(s.create_manual_asset("Synthetic boundary", amount, "2026-09-30", "real_estate", currency=currency, db_path=path))["manual_asset"]["id"]
|
|
d = draft(path)
|
|
before = frozen_items(path, d["id"])
|
|
r = ok(s.review_settlement_draft(d["id"], db_path=path))
|
|
row = next(x for x in r["items"] if x["source_id"] == asset)
|
|
assert row["amount"] == expected
|
|
assert row["status"] == ("missing" if expected is None else "ready")
|
|
assert r["total_assets"] == (expected or 0)
|
|
assert frozen_items(path, d["id"]) == before
|
|
|
|
|
|
@pytest.mark.parametrize("mode,when,reason", [("exact", "2026-10-02", ""), ("carried_before", "2026-09-30", "carry"), ("carried_before", "2026-10-02", "carry"), ("manual_substitute", "2026-10-02", "")])
|
|
def test_invalid_bank_date_or_missing_reason_keeps_draft_unchanged(tmp_path, mode, when, reason):
|
|
path = str(tmp_path / "invalid-selection.sqlite3")
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account("Synthetic bank", db_path=path))["account"]["id"]
|
|
val = ok(s.register_account_valuation(account, when, 20, valuation_kind="cash_balance", verification_status="verified", db_path=path))["valuation"]["id"]
|
|
d = draft(path)
|
|
before = ok(s.get_settlement_draft(d["id"], db_path=path))
|
|
raw = frozen_items(path, d["id"])
|
|
result = s.set_settlement_draft_item(d["id"], "account", account, d["version"], "selector", valuation_id=val, selection_mode=mode, selection_reason=reason, db_path=path)
|
|
assert not result["success"] and result["error_code"] == "validation_error"
|
|
assert ok(s.get_settlement_draft(d["id"], db_path=path)) == before
|
|
assert frozen_items(path, d["id"]) == raw
|
|
|
|
|
|
def test_unverified_bank_evidence_remains_missing_after_explicit_review_selection(tmp_path):
|
|
path = str(tmp_path / "unverified.sqlite3")
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account("Synthetic bank", db_path=path))["account"]["id"]
|
|
val = ok(s.register_account_valuation(account, "2026-09-30", 20, valuation_kind="cash_balance", verification_status="unverified", db_path=path))["valuation"]["id"]
|
|
d = draft(path)
|
|
r = ok(s.review_settlement_draft(d["id"], db_path=path))
|
|
assert r["items"][0]["valuation_id"] == "" and r["items"][0]["candidate_amount"] == 20
|
|
assert r["items"][0]["amount"] is None
|
|
rejected = s.set_settlement_draft_item(d["id"], "account", account, d["version"], "selector", valuation_id=val, db_path=path)
|
|
assert not rejected["success"] and rejected["error_code"] == "validation_error"
|
|
d = ok(s.set_settlement_draft_item(d["id"], "account", account, d["version"], "selector", valuation_id=val, selection_mode="manual_substitute", selection_reason="review only", db_path=path))["draft"]
|
|
r = approve(path, d, allow_missing_exclusions=True, allow_manual_substitute=True)
|
|
assert r["items"][0]["amount"] is None and r["total_assets"] == 0
|
|
closed = ok(s.finalize_settlement_draft(d["id"], d["version"], r["review_fingerprint"], "unverified-close", db_path=path))
|
|
assert closed["total_assets"] == closed["totals"]["total_assets"] == 0
|
|
assert closed["excluded_items"][0]["amount"] is None
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT verification_status FROM account_valuations WHERE id=?", (val,)).fetchone()[0] == "unverified"
|
|
|
|
|
|
@pytest.mark.parametrize("account_type,kind,definition,registration_allowed,selection_allowed", [("bank", "total_value", None, True, False), ("investment", "loan_balance", "principal_only", False, False), ("loan", "total_value", None, False, False), ("loan", "balance", None, False, False), ("loan", "balance", "principal_only", True, True), ("investment", "total_value", None, True, True)])
|
|
def test_account_type_purpose_and_representative_liability_boundaries(tmp_path, account_type, kind, definition, registration_allowed, selection_allowed):
|
|
path = str(tmp_path / "purpose.sqlite3")
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account("Synthetic purpose", type=account_type, db_path=path))["account"]["id"]
|
|
registered = s.register_account_valuation(account, "2026-09-30", 20, valuation_kind=kind, representative_liability_definition=definition, verification_status="verified", db_path=path)
|
|
assert registered["success"] == registration_allowed
|
|
if not registration_allowed:
|
|
assert registered["error_code"] == "validation_error"
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT COUNT(*) FROM account_valuations").fetchone()[0] == 0
|
|
return
|
|
d = draft(path)
|
|
before = ok(s.get_settlement_draft(d["id"], db_path=path))
|
|
selected = s.set_settlement_draft_item(d["id"], "account", account, d["version"], "selector", valuation_id=registered["valuation"]["id"], db_path=path)
|
|
assert selected["success"] == selection_allowed
|
|
if not selection_allowed:
|
|
assert selected["error_code"] == "validation_error"
|
|
assert ok(s.get_settlement_draft(d["id"], db_path=path)) == before
|
|
|
|
|
|
@pytest.mark.parametrize("amount", [None, True, 1.5])
|
|
def test_account_valuation_amount_requires_a_strict_integer(tmp_path, amount):
|
|
path = str(tmp_path / "strict-amount.sqlite3")
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account("Synthetic strict", db_path=path))["account"]["id"]
|
|
rejected = s.register_account_valuation(account, "2026-09-30", amount, verification_status="verified", db_path=path)
|
|
assert not rejected["success"] and rejected["error_code"] == "validation_error"
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT COUNT(*) FROM account_valuations").fetchone()[0] == 0
|
|
|
|
|
|
|
|
def test_legacy_amount_approval_requires_fresh_review_after_policy_correction(tmp_path):
|
|
import hashlib
|
|
|
|
path = str(tmp_path / "legacy-approval.sqlite3")
|
|
init_db(path)
|
|
ok(s.create_manual_asset("Synthetic asset", 100, "2026-09-30", "real_estate", db_path=path))
|
|
ok(s.create_manual_asset("Synthetic signed debt", -40, "2026-09-30", "loan", db_path=path))
|
|
d = draft(path)
|
|
frozen_before = frozen_items(path, d["id"])
|
|
# Historical persisted fingerprint schema: immutable baseline _review serial
|
|
# did not include normalized totals. Reconstruct its exact manual-only inputs,
|
|
# rather than treating the current fingerprint as a historical approval.
|
|
with sqlite3.connect(path) as db:
|
|
db.row_factory = sqlite3.Row
|
|
stored = [dict(x) for x in db.execute("SELECT * FROM settlement_draft_items WHERE draft_id=? ORDER BY source_type,source_id", (d["id"],))]
|
|
states = []
|
|
for item in stored:
|
|
current = {"source": dict(db.execute("SELECT * FROM manual_assets WHERE id=?", (item["source_id"],)).fetchone())}
|
|
states.append({"stored": item, "current": current, "candidates": []})
|
|
assert db.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
|
|
serial = json.dumps({"draft_id": d["id"], "version": d["version"], "basis_date": d["basis_date"], "states": states, "warnings": [], "latest_run": "", "correction_parent": d["correction_of_run_id"]}, sort_keys=True, ensure_ascii=False)
|
|
legacy_fingerprint = hashlib.sha256(serial.encode()).hexdigest()
|
|
flags = {"allow_missing_exclusions": False, "allow_carried_before": False, "allow_manual_substitute": False}
|
|
with sqlite3.connect(path) as db:
|
|
db.execute("UPDATE settlement_drafts SET approval_fingerprint=?,approval_version=?,approved_by=?,approved_at=?,approval_reason=?,approval_reference=?,approval_flags=? WHERE id=?", (legacy_fingerprint, d["version"], "synthetic legacy reviewer", "2026-10-10T00:00:00Z", "reviewed legacy net 140", "synthetic reference", json.dumps(flags), d["id"]))
|
|
rejected = s.finalize_settlement_draft(d["id"], d["version"], legacy_fingerprint, "legacy-close", db_path=path)
|
|
assert not rejected["success"], "a historical amount approval must not authorize corrected totals"
|
|
assert rejected["error_code"] == "conflict"
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
|
|
assert db.execute("SELECT COUNT(*) FROM asset_snapshots").fetchone()[0] == 0
|
|
row = db.execute("SELECT status,version,final_run_id FROM settlement_drafts WHERE id=?", (d["id"],)).fetchone()
|
|
assert row == ("active", d["version"], "")
|
|
assert frozen_items(path, d["id"]) == frozen_before
|
|
reviewed = approve(path, d)
|
|
assert reviewed["review_fingerprint"] != legacy_fingerprint
|
|
final = ok(s.finalize_settlement_draft(d["id"], d["version"], reviewed["review_fingerprint"], "legacy-close", db_path=path))
|
|
assert final["totals"] == {"total_assets": 100, "total_liabilities": 40, "net_worth": 60, "income_total": 0, "expense_total": 0, "net_cashflow": 0}
|
|
assert ok(s.get_settlement_draft(d["id"], db_path=path))["final_response"] == final
|
|
assert ok(s.finalize_settlement_draft(d["id"], d["version"], reviewed["review_fingerprint"], "legacy-close", db_path=path)) == final
|
|
assert frozen_items(path, d["id"]) == frozen_before
|
|
|
|
|
|
def test_completed_close_replay_uses_immutable_response_without_revalidation(tmp_path, monkeypatch):
|
|
from jarca_ledger_mcp.application import settlement_draft_service
|
|
|
|
path = str(tmp_path / "completed-replay.sqlite3")
|
|
init_db(path)
|
|
ok(s.create_manual_asset("Synthetic asset", 100, "2026-09-30", "real_estate", db_path=path))
|
|
d = draft(path)
|
|
reviewed = approve(path, d)
|
|
final = ok(s.finalize_settlement_draft(d["id"], d["version"], reviewed["review_fingerprint"], "completed-close", db_path=path))
|
|
|
|
def forbidden_review(*args):
|
|
raise AssertionError("completed history must not be reinterpreted")
|
|
|
|
monkeypatch.setattr(settlement_draft_service, "_review", forbidden_review)
|
|
assert ok(s.finalize_settlement_draft(d["id"], d["version"], reviewed["review_fingerprint"], "completed-close", db_path=path)) == final
|
|
assert ok(s.get_settlement_draft(d["id"], db_path=path))["final_response"] == final
|