311 lines
18 KiB
Python
311 lines
18 KiB
Python
"""Cashflow policy lifecycle and explicit-classification contract tests."""
|
|
import json
|
|
import sqlite3
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def _capability(name):
|
|
candidate = getattr(service, name, None)
|
|
assert callable(candidate), f"required public service capability {name} is absent"
|
|
return candidate
|
|
|
|
|
|
def _ok(result):
|
|
assert result["success"], result
|
|
return result
|
|
|
|
|
|
def _draft(path, month="2026-09"):
|
|
return _ok(service.create_settlement_draft(month, "synthetic creator", "synthetic-key", db_path=path))["draft"]
|
|
|
|
|
|
def _policy(decisions=None):
|
|
return {"savings_categories": [{"category": "Savings", "subcategory": "Monthly"}], "category_decisions": decisions or []}
|
|
|
|
|
|
def _add_category(name, subcategory, kind, path):
|
|
return _ok(service.create_ledger_category(name, subcategory, cashflow_kind=kind, db_path=path))
|
|
|
|
|
|
def _add_tx(day, typ, amount, category, subcategory, path):
|
|
result = service.create_ledger_transaction(day, typ, amount, category, subcategory, db_path=path)
|
|
assert result["success"], result
|
|
assert isinstance(result["transaction"], dict)
|
|
return result["transaction"]["id"]
|
|
|
|
|
|
def test_explicit_savings_refund_and_movement_precedence(tmp_path):
|
|
path = str(tmp_path / "policy-precedence.sqlite3")
|
|
init_db(path)
|
|
draft = _draft(path)
|
|
setter = _capability("set_settlement_cashflow_policy")
|
|
_capability("get_monthly_settlement_snapshot")
|
|
_add_category("Savings", "Monthly", "investment", path)
|
|
_add_category("Savings", "Fund", "investment", path)
|
|
_add_category("Movement", "Transfer", "transfer", path)
|
|
_add_category("Voucher", "Load", "expense", path)
|
|
_add_category("Voucher", "Consumption", "expense", path)
|
|
_add_category("Loan", "Debt payment", "debt_payment", path)
|
|
_add_category("Food", "Meals", "expense", path)
|
|
ids = {
|
|
"save": _add_tx("2026-09-01", "expense", 100, "Savings", "Monthly", path),
|
|
"unlisted_investment": _add_tx("2026-09-02", "expense", 50, "Savings", "Fund", path),
|
|
"transfer": _add_tx("2026-09-03", "transfer", 200, "Movement", "Transfer", path),
|
|
"voucher_load": _add_tx("2026-09-04", "expense", 300, "Voucher", "Load", path),
|
|
"voucher_consumption": _add_tx("2026-09-05", "expense", 30, "Voucher", "Consumption", path),
|
|
"debt": _add_tx("2026-09-06", "expense", 40, "Loan", "Debt payment", path),
|
|
"food": _add_tx("2026-09-07", "expense", 80, "Food", "Meals", path),
|
|
"refund": _add_tx("2026-09-30", "refund", 20, "Food", "Meals", path),
|
|
}
|
|
policy = _policy([{"category": "Voucher", "subcategory": "Load", "cashflow_kind": "excluded", "reason": "synthetic asset movement"}])
|
|
changed = _ok(setter(draft["id"], draft["version"], "synthetic operator", policy, db_path=path))
|
|
saved = _ok(service.get_settlement_draft(draft["id"], db_path=path))["draft"]
|
|
assert saved["cashflow_policy"] == policy and changed["draft"]["version"] == draft["version"] + 1
|
|
reviewed = _ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
assert "cashflow_items" in reviewed
|
|
grouped = {(x["cashflow_kind"], x["category"], x["subcategory"]): x for x in reviewed["cashflow_items"]}
|
|
assert grouped[("expense", "Savings", "Monthly")]["amount"] == 100
|
|
assert grouped[("expense", "Savings", "Monthly")]["transaction_count"] == 1
|
|
assert grouped[("excluded", "Voucher", "Load")]["transaction_count"] == 1
|
|
assert grouped[("expense", "Voucher", "Consumption")]["amount"] == 30
|
|
assert grouped[("expense", "Loan", "Debt payment")]["amount"] == 40
|
|
assert grouped[("expense", "Food", "Meals")]["amount"] == 60
|
|
assert not reviewed["cashflow_blockers"]
|
|
source = reviewed["source_snapshot"]["transactions"]
|
|
by_id = {x["id"]: x for x in source}
|
|
for txid in ids.values():
|
|
assert {"id", "transaction_date", "type", "amount", "currency", "category", "subcategory", "cashflow_kind", "applied_amount"} <= by_id[txid].keys()
|
|
assert by_id[ids["refund"]]["amount"] == 20
|
|
assert by_id[ids["refund"]]["applied_amount"] == -20
|
|
assert by_id[ids["transfer"]]["cashflow_kind"] == "excluded"
|
|
assert by_id[ids["unlisted_investment"]]["cashflow_kind"] == "excluded"
|
|
assert reviewed["expense_total"] == 230
|
|
|
|
|
|
def test_unknown_classification_blocks_and_explicit_exclusion_unblocks(tmp_path):
|
|
path = str(tmp_path / "unknown-blocker.sqlite3")
|
|
init_db(path)
|
|
draft = _draft(path)
|
|
_add_category("Mystery", "Other", "unknown", path)
|
|
_add_tx("2026-09-14", "expense", 60, "Mystery", "Other", path)
|
|
reviewed = _ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
assert reviewed["cashflow_blockers"]
|
|
assert not service.approve_settlement_draft(draft["id"], draft["version"], reviewed["review_fingerprint"], "reviewer", "synthetic", "ref", db_path=path, allow_missing_exclusions=True)["success"]
|
|
explicit = _policy([{"category": "Mystery", "subcategory": "Other", "cashflow_kind": "excluded", "reason": "synthetic operator exclusion"}])
|
|
edited = _ok(_capability("set_settlement_cashflow_policy")(draft["id"], draft["version"], "operator", explicit, db_path=path))["draft"]
|
|
reviewed = _ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
assert not reviewed["cashflow_blockers"]
|
|
assert reviewed["version"] == edited["version"]
|
|
assert service.approve_settlement_draft(draft["id"], edited["version"], reviewed["review_fingerprint"], "reviewer", "synthetic", "ref", db_path=path)["success"]
|
|
|
|
|
|
def test_policy_edit_persists_versions_and_invalidates_approval(tmp_path):
|
|
path = str(tmp_path / "policy-approval.sqlite3")
|
|
init_db(path)
|
|
_ok(service.create_manual_asset("Synthetic cash", 100, "2026-09-30", "cash", db_path=path))
|
|
draft = _draft(path)
|
|
reviewed = _ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
assert not reviewed["cashflow_blockers"]
|
|
_ok(service.approve_settlement_draft(draft["id"], draft["version"], reviewed["review_fingerprint"], "reviewer", "synthetic", "ref", db_path=path))
|
|
result = _ok(_capability("set_settlement_cashflow_policy")(draft["id"], draft["version"], "editor", _policy(), db_path=path))
|
|
saved = _ok(service.get_settlement_draft(draft["id"], db_path=path))["draft"]
|
|
assert saved["cashflow_policy"] == _policy()
|
|
assert result["draft"]["version"] == draft["version"] + 1
|
|
assert saved["approved_by"] == ""
|
|
|
|
|
|
def test_invalid_policy_payload_is_rejected_without_write(tmp_path):
|
|
path = str(tmp_path / "policy-invalid.sqlite3")
|
|
init_db(path)
|
|
draft = _draft(path)
|
|
before = _ok(service.get_settlement_draft(draft["id"], db_path=path))
|
|
invalid = {"savings_categories": [{"category": "Savings"}], "category_decisions": []}
|
|
result = _capability("set_settlement_cashflow_policy")(draft["id"], draft["version"], "operator", invalid, db_path=path)
|
|
assert not result["success"]
|
|
assert _ok(service.get_settlement_draft(draft["id"], db_path=path)) == before
|
|
|
|
|
|
def test_review_contract_and_stored_reader_never_recomputes_live_preview(tmp_path):
|
|
path = str(tmp_path / "stored-reader.sqlite3")
|
|
init_db(path)
|
|
reader = _capability("get_monthly_settlement_snapshot")
|
|
empty = reader(month="2026-09", db_path=path)
|
|
assert empty["success"] is False and empty["error_code"] == "not_found"
|
|
_ok(service.create_manual_asset("Synthetic cash", 100, "2026-09-30", "cash", db_path=path))
|
|
draft = _draft(path)
|
|
reviewed = _ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
assert "review_fingerprint" in reviewed and "review" not in reviewed
|
|
approved = _ok(service.approve_settlement_draft(draft["id"], draft["version"], reviewed["review_fingerprint"], "reviewer", "synthetic", "ref", db_path=path))
|
|
final = _ok(service.finalize_settlement_draft(draft["id"], approved["draft"]["version"], reviewed["review_fingerprint"], "close-key", db_path=path))
|
|
assert "final_response" not in final and "snapshot_run_id" in final
|
|
snapshot = _ok(reader(run_id=final["snapshot_run_id"], db_path=path))
|
|
assert snapshot["is_closed_snapshot"] is True
|
|
assert snapshot["snapshot_run_id"] == final["snapshot_run_id"]
|
|
assert snapshot["total_assets"] == 100
|
|
assert _ok(reader(month="2026-09", db_path=path))["snapshot_run_id"] == final["snapshot_run_id"]
|
|
_ok(service.create_manual_asset("Later live cash", 900, "2026-09-30", "cash", db_path=path))
|
|
persisted = _ok(reader(run_id=final["snapshot_run_id"], db_path=path))
|
|
assert persisted["total_assets"] == 100 and persisted["snapshot_run_id"] == final["snapshot_run_id"]
|
|
assert "live_authority" not in persisted.values()
|
|
|
|
|
|
def test_month_boundaries_refund_month_negative_total_and_fx_blocker(tmp_path):
|
|
path = str(tmp_path / "month-boundaries.sqlite3")
|
|
init_db(path)
|
|
_add_category("Food", "Meals", "expense", path)
|
|
_add_tx("2026-08-31", "expense", 10, "Food", "Meals", path)
|
|
_add_tx("2026-09-01", "expense", 5, "Food", "Meals", path)
|
|
_add_tx("2026-10-01", "expense", 7, "Food", "Meals", path)
|
|
_add_tx("2026-09-15", "refund", 20, "Food", "Meals", path)
|
|
_add_tx("2026-09-16", "expense", 3, "Food", "Meals", path)
|
|
d = _draft(path)
|
|
reviewed = _ok(service.review_settlement_draft(d["id"], db_path=path))
|
|
food = next(x for x in reviewed["cashflow_items"] if x["cashflow_kind"] == "expense" and x["category"] == "Food")
|
|
assert food["amount"] == -12
|
|
assert reviewed["expense_total"] == -12
|
|
assert reviewed["cashflow_blockers"] == []
|
|
_add_category("Foreign", "Purchase", "expense", path)
|
|
_add_tx("2026-09-20", "expense", 12, "Foreign", "Purchase", path)
|
|
with sqlite3.connect(path) as db:
|
|
db.execute("UPDATE transactions SET currency='USD' WHERE category='Foreign'")
|
|
reviewed = _ok(service.review_settlement_draft(d["id"], db_path=path))
|
|
assert reviewed["cashflow_blockers"]
|
|
assert any("fx" in json.dumps(x).lower() or "currency" in json.dumps(x).lower() for x in reviewed["cashflow_blockers"])
|
|
assert reviewed["expense_total"] == -12
|
|
|
|
|
|
def test_known_refund_only_category_is_retained_at_zero(tmp_path):
|
|
path = str(tmp_path / "zero-refund.sqlite3")
|
|
init_db(path)
|
|
_add_category("Food", "Meals", "expense", path)
|
|
_add_tx("2026-09-01", "expense", 20, "Food", "Meals", path)
|
|
_add_tx("2026-09-02", "refund", 20, "Food", "Meals", path)
|
|
reviewed = _ok(service.review_settlement_draft(_draft(path)["id"], db_path=path))
|
|
food = next(x for x in reviewed["cashflow_items"] if x["cashflow_kind"] == "expense" and x["category"] == "Food")
|
|
assert food["amount"] == 0 and food["transaction_count"] == 2
|
|
|
|
|
|
def test_policy_capabilities_are_checked_at_call_time(tmp_path):
|
|
path = str(tmp_path / "future-capabilities.sqlite3")
|
|
init_db(path)
|
|
assert callable(_capability("set_settlement_cashflow_policy"))
|
|
assert callable(_capability("get_monthly_settlement_snapshot"))
|
|
assert _capability("get_monthly_settlement_snapshot")(month="", run_id="", db_path=path)["error_code"] == "not_found"
|
|
|
|
|
|
def _private_cashflow_review(path, policy=None):
|
|
from jarca_ledger_mcp.application.settlement_cashflow_service import _review_cashflow
|
|
with sqlite3.connect(path) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
return _review_cashflow(conn, "2026-09", policy or {})
|
|
|
|
|
|
def test_cashflow_group_field_matches_persisted_detail_contract(tmp_path):
|
|
path = str(tmp_path / "group-field.sqlite3")
|
|
init_db(path)
|
|
_add_category("Food", "Meals", "expense", path)
|
|
_add_tx("2026-09-01", "expense", 20, "Food", "Meals", path)
|
|
result = _private_cashflow_review(path)
|
|
assert len(result["cashflow_items"]) == 1
|
|
assert "cashflow_kind" in result["cashflow_items"][0], "group shape must use the persisted/public cashflow_kind field"
|
|
|
|
|
|
def test_refund_reduces_debt_payment_and_selected_savings(tmp_path):
|
|
for kind, category, subcategory in [("debt_payment", "Loan", "Payment"), ("investment", "Savings", "Monthly")]:
|
|
path = str(tmp_path / (kind + "-refund.sqlite3"))
|
|
init_db(path)
|
|
_add_category(category, subcategory, kind, path)
|
|
txid = _add_tx("2026-09-01", "refund", 20, category, subcategory, path)
|
|
result = _private_cashflow_review(path, _policy())
|
|
assert not result["cashflow_blockers"]
|
|
assert result["expense_total"] == -20, "refund must reduce expense even for debt payment or selected savings"
|
|
assert result["source_snapshot"]["transactions"][0]["id"] == txid
|
|
assert result["source_snapshot"]["transactions"][0]["applied_amount"] == -20
|
|
|
|
|
|
def test_unknown_cashflow_source_is_not_recorded_as_excluded(tmp_path):
|
|
path = str(tmp_path / "unknown-source.sqlite3")
|
|
init_db(path)
|
|
_add_category("Mystery", "Other", "unknown", path)
|
|
_add_tx("2026-09-01", "expense", 20, "Mystery", "Other", path)
|
|
result = _private_cashflow_review(path)
|
|
assert result["cashflow_blockers"] and not result["cashflow_items"]
|
|
assert result["source_snapshot"]["transactions"][0]["cashflow_kind"] != "excluded", "unknown is not an authorized exclusion"
|
|
|
|
|
|
def test_nonstring_cashflow_decision_kind_has_validation_error():
|
|
from jarca_ledger_mcp.application.settlement_cashflow_service import _normalize_cashflow_policy
|
|
malformed = {"category_decisions": [{"category": "Food", "subcategory": "Meals", "cashflow_kind": [], "reason": "synthetic"}]}
|
|
error = None
|
|
try:
|
|
_normalize_cashflow_policy(malformed)
|
|
except Exception as caught:
|
|
error = caught
|
|
assert isinstance(error, ValueError), "malformed kind must have the validator's ValueError contract, not an unhandled TypeError"
|
|
|
|
|
|
def test_inactive_category_uses_raw_fallback_through_public_close(tmp_path):
|
|
path = str(tmp_path / "inactive-category-fallback.sqlite3")
|
|
init_db(path)
|
|
# A close run requires a nonempty asset valuation; cashflow-only drafts cannot finalize.
|
|
_ok(service.create_manual_asset("Synthetic close cash", 100, "2026-09-30", "cash", db_path=path))
|
|
cases = [
|
|
# SQL fallback is category-based: ordinary legacy categories become expense,
|
|
# while canonical legacy categories retain their established classifications.
|
|
("Old exclude", "A", "exclude", "expense", 11, "expense"),
|
|
("Old income", "B", "income", "expense", 13, "expense"),
|
|
("Old transfer", "C", "transfer", "expense", 17, "expense"),
|
|
("Old unknown", "D", "unknown", "income", 19, "expense"),
|
|
("Old refund", "E", "exclude", "refund", 5, "expense"),
|
|
("Old movement", "F", "exclude", "transfer", 7, "excluded"),
|
|
("수입", "Legacy", "exclude", "income", 29, "income"),
|
|
("Old expense", "H", "expense", "expense", 37, "expense"),
|
|
("Old investment", "G", "investment", "expense", 31, "expense"),
|
|
]
|
|
ids = {}
|
|
for index, (category, subcategory, kind, raw_type, amount, expected) in enumerate(cases):
|
|
_add_category(category, subcategory, kind, path)
|
|
txid = _add_tx(f"2026-09-{index + 1:02d}", raw_type, amount, category, subcategory, path)
|
|
assert service.delete_ledger_category(category, subcategory, force=True, db_path=path)["success"]
|
|
ids[txid] = (expected, kind, raw_type, amount)
|
|
draft = _draft(path)
|
|
reviewed = _ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
by_id = {row["id"]: row for row in reviewed["source_snapshot"]["transactions"]}
|
|
for txid, (expected, _kind, raw_type, amount) in ids.items():
|
|
row = by_id[txid]
|
|
applied = -amount if raw_type == "refund" else amount
|
|
assert row["cashflow_kind"] == expected
|
|
assert row["originalcashflow_kind"] == ("income" if row["category"] == "수입" else ("unknown" if row["category"] == "미분류" else "expense"))
|
|
assert row["applied_amount"] == applied # excluded movement retains its amount as evidence
|
|
assert row["sourceprovenance"] == "fallback"
|
|
assert row["category_cashflow_kind"] is None and row["category_active"] is None
|
|
assert reviewed["expense_total"] == 123 # 11 + 13 + 17 + 19 - 5 + 31 + 37
|
|
assert reviewed["income_total"] == 29
|
|
approved = _ok(service.approve_settlement_draft(draft["id"], draft["version"], reviewed["review_fingerprint"], "reviewer", "synthetic", "ref", db_path=path))
|
|
closed = _ok(service.finalize_settlement_draft(draft["id"], approved["draft"]["version"], reviewed["review_fingerprint"], "inactive-category-close", db_path=path))
|
|
snapshot = _ok(service.get_monthly_settlement_snapshot(run_id=closed["snapshot_run_id"], db_path=path))
|
|
persisted = {row["id"]: row for row in snapshot["source_snapshot"]["transactions"]}
|
|
for txid in ids:
|
|
assert persisted[txid]["sourceprovenance"] == "fallback"
|
|
assert persisted[txid]["category_active"] is None
|
|
assert persisted[txid]["category_cashflow_kind"] is None
|
|
assert snapshot["expense_total"] == 123 and snapshot["income_total"] == 29
|
|
|
|
|
|
def test_explicit_decision_classifies_only_unresolved_inactive_fallback(tmp_path):
|
|
path = str(tmp_path / "inactive-explicit-decision.sqlite3")
|
|
init_db(path)
|
|
_add_category("미분류", "Retired", "unknown", path)
|
|
unresolved = _add_tx("2026-09-10", "expense", 23, "미분류", "Retired", path)
|
|
assert service.delete_ledger_category("미분류", "Retired", force=True, db_path=path)["success"]
|
|
draft = _draft(path)
|
|
policy = _policy([{"category": "미분류", "subcategory": "Retired", "cashflow_kind": "income", "reason": "synthetic explicit decision"}])
|
|
_ok(_capability("set_settlement_cashflow_policy")(draft["id"], draft["version"], "operator", policy, db_path=path))
|
|
reviewed = _ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
row = next(x for x in reviewed["source_snapshot"]["transactions"] if x["id"] == unresolved)
|
|
assert row["cashflow_kind"] == "income" and row["decision"]["cashflow_kind"] == "income"
|
|
assert row["originalcashflow_kind"] == "unknown" and row["sourceprovenance"] == "fallback"
|
|
assert reviewed["income_total"] == 23
|