479 lines
30 KiB
Python
479 lines
30 KiB
Python
"""Issue 18 comprehensive settlement acceptance contracts (synthetic SQLite only).
|
|
|
|
Public APIs are resolved at call time so the pre-implementation suite collects and
|
|
fails with an actionable assertion rather than an import/attribute error.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
import pytest
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def api(name):
|
|
fn = getattr(service, name, None)
|
|
assert callable(fn), f"required comprehensive settlement API missing: service.{name}"
|
|
return fn
|
|
|
|
|
|
def dbfile(tmp_path):
|
|
path = str(tmp_path / "comprehensive-synthetic.sqlite3")
|
|
init_db(path)
|
|
return path
|
|
|
|
|
|
def ok(value):
|
|
assert value.get("success") is True, value
|
|
return value
|
|
|
|
|
|
def new_draft(path, month="2026-09", key="draft"):
|
|
created = service.create_manual_asset("Synthetic cash", 100, "2026-09-30", "cash", db_path=path)
|
|
assert created.get("success") is True, created
|
|
return ok(api("create_settlement_draft")(month, "synthetic-operator", key, db_path=path))["draft"]
|
|
|
|
|
|
def close_draft(path, draft, idem="close"):
|
|
review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
|
|
review = review_result
|
|
approved = api("approve_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "synthetic-reviewer", "synthetic approval", "ref-synthetic", db_path=path)
|
|
ok(approved)
|
|
fresh = ok(api("get_settlement_draft")(draft["id"], db_path=path))["draft"]
|
|
return api("finalize_settlement_draft")(draft["id"], fresh["version"], review["review_fingerprint"], idem, db_path=path)
|
|
|
|
|
|
def test_draft_previews_six_totals_without_creating_run(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
draft = new_draft(path)
|
|
result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
|
|
review = result
|
|
required = {"income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth"}
|
|
assert required <= review.keys() and required <= review["totals"].keys()
|
|
assert review["aggregation_version"] == "comprehensive-v1"
|
|
assert review["snapshot_schema_version"] == 1
|
|
assert isinstance(review["cashflow_items"], list)
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
|
|
assert db.execute("SELECT COUNT(*) FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone()[0] == 1
|
|
|
|
|
|
def test_closed_header_persists_six_totals_policy_and_versions(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
draft = new_draft(path)
|
|
result = ok(close_draft(path, draft))
|
|
assert result.get("run_id")
|
|
saved = result
|
|
for key in ("income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth"):
|
|
assert key in saved
|
|
assert saved["aggregation_version"] == "comprehensive-v1"
|
|
assert saved["snapshot_schema_version"] == 1
|
|
with sqlite3.connect(path) as db:
|
|
columns = {row[1] for row in db.execute("PRAGMA table_info(asset_snapshot_runs)")}
|
|
assert {"income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth", "cashflow_policy_json", "aggregation_version", "snapshot_schema_version", "source_snapshot_json"} <= columns
|
|
row = db.execute("SELECT income_total,expense_total,net_cashflow,total_assets,total_liabilities,net_worth FROM asset_snapshot_runs WHERE id=?", (saved["run_id"],)).fetchone()
|
|
assert len(row) == 6
|
|
|
|
|
|
def test_asset_detail_preserves_raw_evidence_and_signed_liability_contributions(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
a = service.create_ledger_account("Synthetic debt A", type="loan", liquidity_bucket="loan", db_path=path)["account"]["id"]
|
|
b = service.create_ledger_account("Synthetic debt B", type="loan", liquidity_bucket="loan", db_path=path)["account"]["id"]
|
|
for account, amount, key in ((a, -900, "loan-a"), (b, 200, "loan-b")):
|
|
v = service.register_account_valuation(account, "2026-09-30", amount, valuation_kind="loan_balance", representative_liability_definition="principal_only", verification_status="verified", idempotency_key=key, db_path=path)
|
|
assert v["success"], v
|
|
draft = new_draft(path)
|
|
review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
|
|
review = review_result
|
|
entries = review.get("items", [])
|
|
assert entries
|
|
assert any(x.get("amount") in (-900, 200) for x in entries)
|
|
result = ok(close_draft(path, draft))
|
|
evidence = json.dumps(result, sort_keys=True)
|
|
assert "applied_amount" in evidence
|
|
assert "-900" in evidence and "200" in evidence
|
|
|
|
|
|
def test_source_snapshot_preserves_reproducible_values_without_raw_secrets(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
account = service.create_ledger_account("Synthetic evidence", last_balance=42, liquidity_bucket="cash", db_path=path)["account"]["id"]
|
|
# Evidence is synthetic; no credentials or user-provided freeform payloads.
|
|
v = service.register_account_valuation(account, "2026-09-30", 42, valuation_kind="balance", verification_status="verified", idempotency_key="value", source_ref="synthetic-ref", db_path=path)
|
|
assert v["success"], v
|
|
result = ok(close_draft(path, new_draft(path)))
|
|
saved = result
|
|
assert saved.get("source_snapshot") or saved.get("source_snapshot_json")
|
|
encoded = json.dumps(saved, sort_keys=True)
|
|
assert "synthetic-ref" in encoded
|
|
for secret in ("raw_payload", "password", "cookie", "token"):
|
|
assert secret not in encoded.lower()
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["ledger_amount", "category_classification"])
|
|
def test_source_policy_and_category_drift_invalidate_approval(tmp_path, mutation):
|
|
path = dbfile(tmp_path)
|
|
category = service.create_ledger_category("Food", "Meal", cashflow_kind="expense", db_path=path)
|
|
assert category.get("success") is True, category
|
|
transaction = service.create_ledger_transaction(transaction_date="2026-09-12", type="expense", amount=17, category="Food", subcategory="Meal", description="synthetic drift", dedupe=False, db_path=path)
|
|
assert transaction.get("success") is True, transaction
|
|
txid = transaction.get("transaction", transaction).get("id")
|
|
draft = new_draft(path)
|
|
review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
|
|
review = review_result
|
|
ok(api("approve_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "synthetic-reviewer", "approved", "ref", db_path=path))
|
|
with sqlite3.connect(path) as db:
|
|
if mutation == "ledger_amount":
|
|
db.execute("UPDATE transactions SET amount=amount+1 WHERE id=?", (txid,))
|
|
else:
|
|
tables = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
assert "ledger_categories" in tables
|
|
cols = {r[1] for r in db.execute("PRAGMA table_info(ledger_categories)")}
|
|
assert {"category", "subcategory", "cashflow_kind"} <= cols
|
|
db.execute("UPDATE ledger_categories SET cashflow_kind='income' WHERE category=? AND subcategory=?", ("Food", "Meal"))
|
|
db.commit()
|
|
closed = api("finalize_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "close", db_path=path)
|
|
assert closed.get("success") is False, closed
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
|
|
|
|
|
|
@pytest.mark.parametrize("failure_point", ["header", "assets", "cashflow", "completion"])
|
|
def test_mid_close_failure_rolls_back_all_settlement_state(tmp_path, failure_point):
|
|
path = dbfile(tmp_path)
|
|
ok(service.create_ledger_category("Work", "Salary", cashflow_kind="income", db_path=path))
|
|
ok(service.create_ledger_transaction("2026-09-10", "income", 50, "Work", "Salary", db_path=path))
|
|
draft = new_draft(path)
|
|
review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
|
|
review = review_result
|
|
ok(api("approve_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "synthetic-reviewer", "approved", "ref", db_path=path))
|
|
with sqlite3.connect(path) as db:
|
|
target = {"header":"asset_snapshot_runs", "assets":"asset_snapshots", "cashflow":"settlement_cashflow_snapshots", "completion":"settlement_drafts"}[failure_point]
|
|
# Missing future table is an explicit RED assertion, not a SQL error masquerading as rollback.
|
|
assert db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (target,)).fetchone(), f"required comprehensive table missing: {target}"
|
|
event = "UPDATE" if failure_point == "completion" else "INSERT"
|
|
condition = " WHEN NEW.status='completed'" if failure_point == "completion" else ""
|
|
db.execute(f"CREATE TRIGGER fail_{failure_point} BEFORE {event} ON {target}{condition} BEGIN SELECT RAISE(ABORT,'synthetic injected failure'); END")
|
|
failed = api("finalize_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "close", db_path=path)
|
|
assert failed.get("success") is False
|
|
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
|
|
assert db.execute("SELECT COUNT(*) FROM settlement_cashflow_snapshots").fetchone()[0] == 0
|
|
assert db.execute("SELECT status FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone()[0] == "active"
|
|
|
|
|
|
def test_close_rejects_stored_detail_sum_tampering(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
ok(service.create_ledger_category("Work", "Salary", cashflow_kind="income", db_path=path))
|
|
ok(service.create_ledger_transaction("2026-09-10", "income", 50, "Work", "Salary", db_path=path))
|
|
draft = new_draft(path)
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT 1 FROM sqlite_master WHERE name='settlement_cashflow_snapshots'").fetchone(), "cashflow detail storage is required"
|
|
db.execute("CREATE TRIGGER corrupt_saved_detail AFTER INSERT ON settlement_cashflow_snapshots BEGIN UPDATE settlement_cashflow_snapshots SET amount=amount+1 WHERE id=NEW.id; END")
|
|
result = close_draft(path, draft)
|
|
assert result.get("success") is False, result
|
|
with sqlite3.connect(path) as db:
|
|
for table in ("asset_snapshot_runs", "asset_snapshots", "settlement_cashflow_snapshots"):
|
|
assert db.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0
|
|
assert db.execute("SELECT status FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone()[0] == "active"
|
|
|
|
|
|
def test_replay_and_correction_keep_each_run_immutable(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
first = new_draft(path)
|
|
initial = ok(close_draft(path, first, "first-close"))
|
|
replay = ok(api("finalize_settlement_draft")(first["id"], first["version"], initial["review_fingerprint"], "first-close", db_path=path))
|
|
assert replay == initial
|
|
correction = ok(api("create_settlement_draft")("2026-09", "synthetic-operator", "correction", correction_of_run_id=initial["run_id"], correction_reason="synthetic correction", db_path=path))["draft"]
|
|
corrected = ok(close_draft(path, correction, "correction-close"))
|
|
assert corrected["run_id"] != initial["run_id"]
|
|
old = ok(api("get_monthly_settlement_snapshot")(run_id=initial["run_id"], db_path=path))
|
|
new = ok(api("get_monthly_settlement_snapshot")(run_id=corrected["run_id"], db_path=path))
|
|
assert old["run_id"] == initial["run_id"] and new["run_id"] == corrected["run_id"]
|
|
|
|
|
|
def test_concurrent_close_retries_return_one_saved_result(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
draft = new_draft(path)
|
|
review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
|
|
review = review_result
|
|
ok(api("approve_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "synthetic-reviewer", "approved", "ref", db_path=path))
|
|
def close():
|
|
return api("finalize_settlement_draft")(draft["id"], draft["version"], review["review_fingerprint"], "same-key", db_path=path)
|
|
with ThreadPoolExecutor(max_workers=4) as pool:
|
|
results = list(pool.map(lambda _: close(), range(4)))
|
|
successes = [r for r in results if r.get("success")]
|
|
assert len(successes) == 4
|
|
assert len({r["run_id"] for r in successes}) == 1
|
|
|
|
|
|
def test_readonly_snapshot_missing_database_does_not_create_file(tmp_path):
|
|
path = tmp_path / "missing.sqlite3"
|
|
result = api("get_monthly_settlement_snapshot")(month="2026-09", db_path=str(path))
|
|
assert result.get("success") is False
|
|
assert not path.exists()
|
|
|
|
|
|
def test_saved_reader_has_six_totals_and_legacy_null_is_not_live_zero(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
# Empty database has no run; a saved reader must identify no snapshot, never fabricate current totals.
|
|
result = api("get_monthly_settlement_snapshot")(month="2026-09", db_path=path)
|
|
assert result.get("success") is False or result.get("source") == "settlement_snapshot"
|
|
for key in ("income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth"):
|
|
assert key not in result or result[key] is None
|
|
|
|
|
|
def test_cashflow_and_legacy_api_shapes_remain_distinct(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
before = service.get_monthly_cashflow_report("2026-09", db_path=path)
|
|
assert before.get("success") is True
|
|
live_keys = set(before)
|
|
draft = new_draft(path)
|
|
review_result = ok(api("review_settlement_draft")(draft["id"], db_path=path))
|
|
review = review_result
|
|
assert {"income_total", "expense_total", "net_cashflow"} <= review.keys()
|
|
after = service.get_monthly_cashflow_report("2026-09", db_path=path)
|
|
assert set(after) == live_keys
|
|
|
|
|
|
def test_policy_setter_persists_explicit_lists_and_changes_review_fingerprint(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
draft = new_draft(path)
|
|
before = ok(api("review_settlement_draft")(draft["id"], db_path=path))["review_fingerprint"]
|
|
policy = {"savings_categories":[{"category":"Savings","subcategory":"Reserve"}], "category_decisions":[]}
|
|
changed = ok(api("set_settlement_cashflow_policy")(draft["id"], draft["version"], "synthetic-operator", policy, db_path=path))
|
|
updated = changed["draft"]
|
|
after = ok(api("review_settlement_draft")(draft["id"], db_path=path))["review_fingerprint"]
|
|
assert updated["version"] == draft["version"] + 1
|
|
assert before != after
|
|
|
|
|
|
def test_completed_get_and_saved_snapshot_ignore_later_live_ledger_changes(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
draft = new_draft(path)
|
|
saved = ok(close_draft(path, draft))
|
|
ok(service.create_ledger_category("Later", "Income", cashflow_kind="income", db_path=path))
|
|
ok(service.create_ledger_transaction("2026-09-20", "income", 99, "Later", "Income", db_path=path))
|
|
history = ok(api("get_settlement_draft")(draft["id"], db_path=path))
|
|
fetched = ok(api("get_monthly_settlement_snapshot")(run_id=saved["run_id"], db_path=path))
|
|
assert history["final_response"] == saved
|
|
assert fetched["income_total"] == saved["income_total"]
|
|
|
|
|
|
def _historical_asset_review_fingerprint(path, draft, review):
|
|
"""Exact f67 asset-only serial, with no new cashflow/policy/version inputs."""
|
|
from jarca_ledger_mcp.application import settlement_draft_service as old
|
|
with sqlite3.connect(path) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
stored = old._items(conn, draft["id"])
|
|
states = []
|
|
for item in stored:
|
|
candidates = [dict(v) for v in conn.execute("SELECT * FROM account_valuations WHERE account_id=? ORDER BY id", (item["source_id"],))] if item["source_type"] == "account" else []
|
|
states.append({"stored": item, "current": old._current_semantics(conn, item), "candidates": candidates})
|
|
serial = {"draft_id": draft["id"], "version": draft["version"], "basis_date": draft["basis_date"], "states": states,
|
|
"warnings": [], "latest_run": old._latest_close(conn, draft["basis_date"]), "correction_parent": draft["correction_of_run_id"],
|
|
"totals": {k: review[k] for k in ("total_assets", "total_liabilities", "net_worth")}}
|
|
return hashlib.sha256(json.dumps(serial, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
|
|
|
|
|
|
def test_legacy_asset_approval_requires_comprehensive_review_before_close(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
draft = new_draft(path)
|
|
review = ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
assert not review["warnings"], "this historical serial fixture is deliberately warning-free"
|
|
historic = _historical_asset_review_fingerprint(path, draft, review)
|
|
if "aggregation_version" not in review:
|
|
assert historic == review["review_fingerprint"], "historical fixture must exactly match the original baseline approval"
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE settlement_drafts SET approval_fingerprint=?,approval_version=?,approved_by='synthetic-reviewer',approved_at='2026-09-30',approval_reason='historic asset review',approval_reference='historic-ref',approval_flags=? WHERE id=?",
|
|
(historic, draft["version"], json.dumps({"allow_missing_exclusions": False, "allow_carried_before": False, "allow_manual_substitute": False}), draft["id"]))
|
|
refused = service.finalize_settlement_draft(draft["id"], draft["version"], historic, "historic-close", db_path=path)
|
|
assert refused["success"] is False, "an asset-only historical approval cannot authorize comprehensive cashflow interpretation"
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
|
|
assert conn.execute("SELECT status,final_response FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone() == ("active", "")
|
|
fresh = ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
assert fresh["review_fingerprint"] != historic
|
|
ok(service.approve_settlement_draft(draft["id"], draft["version"], fresh["review_fingerprint"], "synthetic-reviewer", "comprehensive reapproval", "new-ref", db_path=path))
|
|
saved = ok(service.finalize_settlement_draft(draft["id"], draft["version"], fresh["review_fingerprint"], "fresh-close", db_path=path))
|
|
ok(service.create_ledger_category("Later", "Income", cashflow_kind="income", db_path=path))
|
|
ok(service.create_ledger_transaction("2026-09-20", "income", 99, "Later", "Income", db_path=path))
|
|
assert ok(service.get_settlement_draft(draft["id"], db_path=path))["final_response"] == saved
|
|
assert ok(service.finalize_settlement_draft(draft["id"], draft["version"], fresh["review_fingerprint"], "fresh-close", db_path=path)) == saved
|
|
|
|
|
|
def _seed_reader_run(path, comprehensive=False, source=None):
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("INSERT INTO asset_snapshot_runs(id,basis_date,close_status,created_at) VALUES ('synthetic-run','2026-09-30','complete','2026-09-30')")
|
|
if comprehensive:
|
|
conn.execute("UPDATE asset_snapshot_runs SET settlement_month='2026-09',currency='KRW',income_total=20,expense_total=0,net_cashflow=20,total_assets=0,total_liabilities=0,net_worth=0,cashflow_policy_json='{}',aggregation_version='comprehensive-v1',snapshot_schema_version=1,source_snapshot_json=? WHERE id='synthetic-run'", (json.dumps(source or {}),))
|
|
conn.execute("INSERT INTO settlement_cashflow_snapshots(id,settlement_run_id,cashflow_kind,category,subcategory,amount,transaction_count,currency) VALUES ('synthetic-detail','synthetic-run','income','Salary','Monthly',20,1,'KRW')")
|
|
|
|
|
|
def test_legacy_reader_preserves_unrecorded_currency_and_marks_legacy(tmp_path):
|
|
from jarca_ledger_mcp.application.monthly_settlement_service import get_monthly_settlement_snapshot
|
|
path = str(tmp_path / "reader-legacy-null.sqlite3")
|
|
_seed_reader_run(path)
|
|
result = ok(get_monthly_settlement_snapshot(run_id="synthetic-run", db_path=path))
|
|
assert result["legacy"] is True and result["is_comprehensive"] is False
|
|
assert result["currency"] is None, "legacy currency must not be invented as KRW"
|
|
assert all(value is None for value in result["totals"].values())
|
|
|
|
|
|
def test_legacy_reader_accepts_original_header_without_migration(tmp_path):
|
|
from jarca_ledger_mcp.application.monthly_settlement_service import get_monthly_settlement_snapshot
|
|
path = str(tmp_path / "reader-old-columns.sqlite3")
|
|
_seed_reader_run(path)
|
|
with sqlite3.connect(path) as conn:
|
|
for column in ("settlement_month", "currency", "income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth", "cashflow_policy_json", "aggregation_version", "snapshot_schema_version", "source_snapshot_json"):
|
|
conn.execute(f"ALTER TABLE asset_snapshot_runs DROP COLUMN {column}")
|
|
for kwargs in ({"run_id": "synthetic-run"}, {"month": "2026-09"}):
|
|
result = get_monthly_settlement_snapshot(db_path=path, **kwargs)
|
|
assert result["success"], "pre-comprehensive headers must be read without new-column SQL"
|
|
assert all(value is None for value in result["totals"].values())
|
|
with sqlite3.connect(path) as conn:
|
|
assert "settlement_month" not in {row[1] for row in conn.execute("PRAGMA table_info(asset_snapshot_runs)")}
|
|
|
|
|
|
def test_stored_reader_keeps_required_transaction_date(tmp_path):
|
|
from jarca_ledger_mcp.application.monthly_settlement_service import get_monthly_settlement_snapshot
|
|
path = str(tmp_path / "reader-source-date.sqlite3")
|
|
tx = {"id": "synthetic-tx", "transaction_date": "2026-09-01", "type": "income", "amount": 20, "currency": "KRW", "category": "Salary", "subcategory": "Monthly", "cashflow_kind": "income", "applied_amount": 20}
|
|
_seed_reader_run(path, True, {"transactions": [tx], "assets": []})
|
|
result = ok(get_monthly_settlement_snapshot(run_id="synthetic-run", db_path=path))
|
|
assert result["source_snapshot"]["transactions"] == [tx]
|
|
|
|
|
|
def test_stored_reader_allowlists_source_root_and_nested_decision(tmp_path):
|
|
from jarca_ledger_mcp.application.monthly_settlement_service import get_monthly_settlement_snapshot
|
|
path = str(tmp_path / "reader-source-private.sqlite3")
|
|
marker = "SYNTHETIC_PRIVATE_SNAPSHOT"
|
|
tx = {"id": "synthetic-tx", "transaction_date": "2026-09-01", "type": "income", "amount": 20, "currency": "KRW", "category": "Salary", "subcategory": "Monthly", "cashflow_kind": "income", "applied_amount": 20, "decision": {"category": "Salary", "subcategory": "Monthly", "cashflow_kind": "income", "reason": "synthetic decision", "private_note": marker}}
|
|
_seed_reader_run(path, True, {"transactions": [tx], "assets": [], "private_note": marker})
|
|
result = ok(get_monthly_settlement_snapshot(run_id="synthetic-run", db_path=path))
|
|
assert marker not in json.dumps(result), "unknown root and nested source fields are not public financial evidence"
|
|
|
|
|
|
@pytest.mark.parametrize("target", ["header_total", "asset_amount", "asset_currency", "header_source", "header_policy", "header_version"])
|
|
def test_close_rejects_persisted_header_and_asset_tampering(tmp_path, target):
|
|
path = dbfile(tmp_path)
|
|
ok(service.create_ledger_category("Salary", "Monthly", cashflow_kind="income", db_path=path))
|
|
ok(service.create_ledger_transaction("2026-09-01", "income", 20, "Salary", "Monthly", db_path=path))
|
|
draft = new_draft(path)
|
|
review = ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
ok(service.approve_settlement_draft(draft["id"], draft["version"], review["review_fingerprint"], "reviewer", "synthetic", "ref", db_path=path))
|
|
if target.startswith("asset_"):
|
|
column, expression = ("amount", "amount+1") if target == "asset_amount" else ("currency", "'USD'")
|
|
trigger = f"CREATE TRIGGER synthetic_tamper AFTER INSERT ON asset_snapshots BEGIN UPDATE asset_snapshots SET {column}={expression} WHERE id=NEW.id; END"
|
|
else:
|
|
column, expression = {"header_total": ("income_total", "income_total+1"), "header_source": ("source_snapshot_json", "'{}'"), "header_policy": ("cashflow_policy_json", "'{\"savings_categories\":[{\"category\":\"Other\",\"subcategory\":\"Pair\"}],\"category_decisions\":[]}'"), "header_version": ("aggregation_version", "'unknown-version'")}[target]
|
|
trigger = f"CREATE TRIGGER synthetic_tamper AFTER INSERT ON settlement_cashflow_snapshots BEGIN UPDATE asset_snapshot_runs SET {column}={expression} WHERE id=NEW.settlement_run_id; END"
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute(trigger)
|
|
result = service.finalize_settlement_draft(draft["id"], draft["version"], review["review_fingerprint"], "tamper-close", db_path=path)
|
|
assert result["success"] is False, "persisted header/detail must match the approved result before commit"
|
|
with sqlite3.connect(path) as conn:
|
|
for table in ("asset_snapshot_runs", "asset_snapshots", "settlement_cashflow_snapshots"):
|
|
assert conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0
|
|
assert conn.execute("SELECT status,final_run_id,final_response FROM settlement_drafts WHERE id=?", (draft["id"],)).fetchone() == ("active", "", "")
|
|
|
|
|
|
def test_stored_aggregation_version_drift_invalidates_approval(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
draft = new_draft(path)
|
|
review = ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
ok(service.approve_settlement_draft(draft["id"], draft["version"], review["review_fingerprint"], "reviewer", "synthetic", "ref", db_path=path))
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE settlement_drafts SET aggregation_version='unknown-version' WHERE id=?", (draft["id"],))
|
|
result = service.finalize_settlement_draft(draft["id"], draft["version"], review["review_fingerprint"], "version-drift", db_path=path)
|
|
assert result["success"] is False, "unknown stored logic version cannot silently use current aggregation"
|
|
|
|
|
|
def test_completed_review_returns_saved_totals_and_fingerprint(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
draft = new_draft(path)
|
|
saved = ok(close_draft(path, draft))
|
|
ok(service.create_ledger_category("Later", "Income", cashflow_kind="income", db_path=path))
|
|
ok(service.create_ledger_transaction("2026-09-20", "income", 99, "Later", "Income", db_path=path))
|
|
review = ok(service.review_settlement_draft(draft["id"], db_path=path))
|
|
assert review["totals"] == saved["totals"]
|
|
assert review["review_fingerprint"] == saved["review_fingerprint"]
|
|
|
|
|
|
def test_excluded_loan_source_has_no_applied_contribution(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
values = []
|
|
for name, amount in (("Synthetic A", -900), ("Synthetic B", 200)):
|
|
account = ok(service.create_ledger_account(name, type="loan", liquidity_bucket="loan", db_path=path))["account"]["id"]
|
|
value = ok(service.register_account_valuation(account, "2026-09-30", amount, valuation_kind="loan_balance", representative_liability_definition="principal_only", verification_status="verified", idempotency_key=name, db_path=path))["valuation"]
|
|
values.append((account, value["id"]))
|
|
draft = new_draft(path)
|
|
edited = ok(service.set_settlement_draft_item(draft["id"], "account", values[1][0], draft["version"], "operator", valuation_id=values[1][1], included=False, exclusion_reason="synthetic exclusion", db_path=path))
|
|
review = edited["review"]
|
|
assert review["total_liabilities"] == 900
|
|
excluded = next(x for x in review["source_snapshot"]["assets"] if x["source_id"] == values[1][0])
|
|
assert excluded["amount"] == 200 and excluded["applied_amount"] is None
|
|
|
|
|
|
def test_cashflow_detail_currency_matches_header_without_losing_original(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
ok(service.create_ledger_category("Salary", "Monthly", cashflow_kind="income", db_path=path))
|
|
tx = ok(service.create_ledger_transaction("2026-09-01", "income", 20, "Salary", "Monthly", db_path=path))["transaction"]
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE transactions SET currency='krw' WHERE id=?", (tx["id"],))
|
|
saved = ok(close_draft(path, new_draft(path)))
|
|
assert saved["cashflow_items"][0]["currency"] == "KRW"
|
|
assert saved["source_snapshot"]["transactions"][0]["currency"] == "krw"
|
|
|
|
|
|
def test_saved_reader_retains_asset_authority_and_liability_flags(tmp_path):
|
|
path = dbfile(tmp_path)
|
|
saved = ok(close_draft(path, new_draft(path)))
|
|
snapshot = ok(service.get_monthly_settlement_snapshot(run_id=saved["run_id"], db_path=path))
|
|
asset = snapshot["source_snapshot"]["assets"][0]
|
|
assert asset.get("authority_source") == "manual_assets"
|
|
assert asset.get("is_liability") is False
|
|
assert asset.get("status") == "ready"
|
|
|
|
|
|
@pytest.mark.parametrize("schema_version,aggregation_version", [(1, None), (1, ""), (1, "unknown-version"), (2, "comprehensive-v1")])
|
|
def test_saved_reader_rejects_unsupported_comprehensive_versions_without_writes(tmp_path, schema_version, aggregation_version):
|
|
path = dbfile(tmp_path)
|
|
saved = ok(close_draft(path, new_draft(path)))
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE asset_snapshot_runs SET snapshot_schema_version=?,aggregation_version=? WHERE id=?", (schema_version, aggregation_version, saved["run_id"]))
|
|
conn.commit()
|
|
before = list(conn.iterdump())
|
|
for query in ({"run_id": saved["run_id"]}, {"month": "2026-09"}):
|
|
result = service.get_monthly_settlement_snapshot(db_path=path, **query)
|
|
assert result["success"] is False, "unsupported comprehensive logic/schema is not a recognized authoritative result"
|
|
assert result["error_code"] == "db_error"
|
|
assert "totals" not in result and "cashflow_items" not in result
|
|
with sqlite3.connect(path) as conn:
|
|
assert list(conn.iterdump()) == before
|
|
|
|
|
|
@pytest.mark.parametrize("aggregation_version", ["unknown-version", "comprehensive-v1", ""])
|
|
def test_saved_reader_does_not_mislabel_partial_version_metadata_as_legacy(tmp_path, aggregation_version):
|
|
path = dbfile(tmp_path)
|
|
saved = ok(close_draft(path, new_draft(path)))
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE asset_snapshot_runs SET snapshot_schema_version=NULL,aggregation_version=? WHERE id=?", (aggregation_version, saved["run_id"]))
|
|
conn.commit()
|
|
before = list(conn.iterdump())
|
|
for query in ({"run_id": saved["run_id"]}, {"month": "2026-09"}):
|
|
result = service.get_monthly_settlement_snapshot(db_path=path, **query)
|
|
assert result["success"] is False, "partial version metadata is not a genuine NULL-version legacy run"
|
|
assert result["error_code"] == "db_error"
|
|
assert "totals" not in result
|
|
with sqlite3.connect(path) as conn:
|
|
assert list(conn.iterdump()) == before
|
|
conn.execute("UPDATE asset_snapshot_runs SET aggregation_version=NULL WHERE id=?", (saved["run_id"],))
|
|
legacy = ok(service.get_monthly_settlement_snapshot(run_id=saved["run_id"], db_path=path))
|
|
assert legacy["legacy"] and not legacy["is_comprehensive"]
|
|
assert all(value is None for value in legacy["totals"].values())
|