307 lines
17 KiB
Python
307 lines
17 KiB
Python
import json
|
|
import sqlite3
|
|
|
|
from jarca_ledger_mcp import service
|
|
|
|
|
|
def test_dated_manual_asset_is_verified_and_preview_needs_approval(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("현금", 100, "2026-06-01", "cash", db_path=path)
|
|
|
|
preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path, selection_policy="as_of")
|
|
|
|
assert preview["close_status"] == "incomplete_draft"
|
|
item = preview["included_items"][0]
|
|
assert item["verification_status"] == "verified"
|
|
assert item["evidence_kind"] == "manual_confirmed"
|
|
assert item["valuation_date"] == "2026-06-01"
|
|
assert item["carried_forward"] is True
|
|
assert item["age_days"] == 29
|
|
|
|
|
|
def test_approved_carried_close_is_conditional_and_persists_provenance(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("현금", 100, "2026-06-01", "cash", db_path=path)
|
|
|
|
closed = service.close_asset_snapshot(
|
|
"2026-06-30", db_path=path, selection_policy="as_of",
|
|
allow_as_of_carry_forward=True, approved_by="reviewer",
|
|
approval_reason="approved carry", approval_reference="APR-59",
|
|
)
|
|
assert closed["close_status"] == "conditional"
|
|
item = closed["included_items"][0]
|
|
assert item["verification_status"] == "verified"
|
|
assert item["evidence_kind"] == "manual_confirmed"
|
|
with sqlite3.connect(path) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
row = conn.execute("SELECT applied_evidence FROM asset_snapshot_runs WHERE id=?", (closed["snapshot_run_id"],)).fetchone()
|
|
evidence = json.loads(row["applied_evidence"])
|
|
assert evidence[0]["valuation_date"] == "2026-06-01"
|
|
assert evidence[0]["age_days"] == 29
|
|
assert evidence[0]["carried_forward"] is True
|
|
assert closed["approval"] == {
|
|
"approved_by": "reviewer",
|
|
"approved_at": closed["closed_at"],
|
|
"approval_reason": "approved carry",
|
|
"approval_reference": "APR-59",
|
|
}
|
|
assert closed["correction"]["is_correction"] is False
|
|
assert closed["excluded_items"] == []
|
|
assert closed["provenance"]["carried_forward"] is True
|
|
assert closed["provenance"]["age_days"] == 29
|
|
assert closed["provenance"]["verification_status"] == "verified"
|
|
assert closed["provenance"]["evidence_kind"] == "manual_confirmed"
|
|
assert closed["provenance"]["approval"] == closed["approval"]
|
|
assert closed["provenance"]["correction"] == closed["correction"]
|
|
|
|
|
|
def test_reconciliation_uses_currency_and_loan_priority_and_canonical_provenance(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
investment = service.create_ledger_account("투자", type="investment", currency="USD", last_balance=100, db_path=path)["account"]["id"]
|
|
loan = service.create_ledger_account("대출", type="loan", last_balance=100, db_path=path)["account"]["id"]
|
|
service.register_account_valuation(investment, "2026-06-30", 100, currency="KRW", valuation_kind="total_value", verification_status="verified", db_path=path)
|
|
loan_value = service.register_account_valuation(loan, "2026-06-30", 80, valuation_kind="loan_balance", representative_liability_definition="principal_only", verification_status="verified", db_path=path)
|
|
|
|
report = service.get_balance_reconciliation_report(account_id=loan, baseline_at="2026-06-30T00:00:00+09:00", db_path=path)
|
|
item = report["items"][0]
|
|
assert item["external_confirmed_value"] == 80
|
|
assert item["valuation_id"] == loan_value["valuation"]["id"]
|
|
assert item["valuation_kind"] == "loan_balance"
|
|
assert item["provenance"]["valuation_date"] == "2026-06-30"
|
|
assert item["provenance"]["verification_status"] == "verified"
|
|
|
|
|
|
def test_investment_without_verified_valuation_and_without_opening_balance_is_incomplete(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("투자", type="investment", db_path=path)["account"]["id"]
|
|
report = service.get_balance_reconciliation_report(account_id=account, baseline_at="2026-06-30T00:00:00+09:00", db_path=path)
|
|
item = report["items"][0]
|
|
assert item["status"] == "VALUATION_UNAVAILABLE"
|
|
assert item["calculation_complete"] is False
|
|
assert item["difference"] is None
|
|
assert item["external_confirmed_value"] is None
|
|
|
|
|
|
def test_timeseries_and_monthly_expose_canonical_asset_provenance(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("현금", 100, "2026-06-30", "cash", db_path=path)
|
|
closed = service.close_asset_snapshot("2026-06-30", db_path=path)
|
|
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
timeseries = service.get_net_worth_timeseries(start_date="2026-06-01", end_date="2026-06-30", db_path=path)
|
|
monthly = service.get_monthly_settlement_summary("2026-06", db_path=path)
|
|
|
|
for payload in (report, timeseries["points"][0], monthly["reports"]["assets"], monthly["source_coverage"]["assets"]):
|
|
assert "provenance" in payload
|
|
assert "valuation_date" in payload["provenance"]
|
|
assert "warnings" in payload["provenance"]
|
|
assert "exclusions" in payload["provenance"]
|
|
assert report["provenance"]["snapshot_run_id"] == closed["snapshot_run_id"]
|
|
|
|
|
|
def test_reconciliation_does_not_mark_last_balance_without_opening_evidence_complete(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("manual", last_balance=1000, last_synced_at="2026-06-30", db_path=path)["account"]["id"]
|
|
item = service.get_balance_reconciliation_report(account_id=account, db_path=path)["items"][0]
|
|
assert item["calculation_complete"] is False
|
|
assert item["difference"] is None
|
|
assert item["status"] == "INCOMPLETE_CALCULATION"
|
|
|
|
|
|
def test_monthly_settlement_propagates_snapshot_provenance_id_and_actual_valuation_date(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("cash", 100, "2026-06-01", "cash", db_path=path)
|
|
closed = service.close_asset_snapshot("2026-06-30", db_path=path, selection_policy="as_of", allow_as_of_carry_forward=True, approved_by="r", approval_reason="r", approval_reference="r")
|
|
monthly = service.get_monthly_settlement_summary("2026-06", db_path=path)
|
|
assets = monthly["source_coverage"]["assets"]
|
|
assert assets["monthly_source_snapshot_id"] == closed["snapshot_run_id"]
|
|
assert assets["provenance"]["valuation_date"] == "2026-06-01"
|
|
assert "warnings" in assets and "excluded_items" in assets and "carried_items" in assets
|
|
|
|
|
|
def test_approved_carry_provenance_is_identical_across_closed_downstream_reports(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("cash", 100, "2026-06-01", "cash", db_path=path)
|
|
closed = service.close_asset_snapshot(
|
|
"2026-06-30", db_path=path, selection_policy="as_of",
|
|
allow_as_of_carry_forward=True, approved_by="reviewer",
|
|
approval_reason="approved carry", approval_reference="APR-59",
|
|
)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
point = service.get_net_worth_timeseries(start_date="2026-06-30", end_date="2026-06-30", db_path=path)["points"][0]
|
|
monthly = service.get_monthly_settlement_summary("2026-06", db_path=path)
|
|
payloads = [report, point, monthly["reports"]["assets"], monthly["source_coverage"]["assets"]]
|
|
expected = {key: closed[key] for key in ("warnings", "excluded_items", "carried_items", "approval", "correction")}
|
|
expected.update({key: closed["provenance"][key] for key in ("valuation_date", "age_days", "carried_forward", "verification_status", "evidence_kind")})
|
|
for payload in payloads:
|
|
assert payload.get("approval") == expected["approval"]
|
|
assert payload.get("correction") == expected["correction"]
|
|
assert payload.get("warnings") == expected["warnings"]
|
|
assert payload.get("excluded_items") == expected["excluded_items"]
|
|
assert payload.get("carried_items") == expected["carried_items"]
|
|
provenance = payload["provenance"]
|
|
for key in ("valuation_date", "age_days", "carried_forward", "verification_status", "evidence_kind"):
|
|
assert provenance[key] == expected[key]
|
|
assert report["approval"] == expected["approval"]
|
|
assert report["correction"] == expected["correction"]
|
|
assert point["approval"] == expected["approval"]
|
|
assert point["correction"] == expected["correction"]
|
|
assert monthly["source_coverage"]["assets"]["monthly_source_snapshot_id"] == closed["snapshot_run_id"]
|
|
|
|
|
|
def test_preview_and_live_report_redact_excluded_source_note_and_memo(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
sentinel = "SHOULD_NOT_LEAK_C10"
|
|
account = service.create_ledger_account(
|
|
"unverified account", memo=sentinel, last_balance=100, last_synced_at="2026-06-30", db_path=path,
|
|
)["account"]["id"]
|
|
service.register_account_valuation(account, "2026-06-30", 100, verification_status="unverified", db_path=path)
|
|
service.create_manual_asset("missing manual", None, "2026-06-30", "cash", source_note=sentinel, memo=sentinel, db_path=path)
|
|
|
|
preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path)
|
|
live = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
for payload in (preview, live):
|
|
assert sentinel not in json.dumps(payload, ensure_ascii=False)
|
|
assert all(sentinel not in json.dumps(item, ensure_ascii=False) for item in payload.get("warnings", []))
|
|
assert all(sentinel not in json.dumps(item, ensure_ascii=False) for item in payload.get("excluded_items", []))
|
|
assert sentinel not in json.dumps(payload.get("provenance", {}), ensure_ascii=False)
|
|
|
|
|
|
def test_live_report_and_monthly_propagate_canonical_carry_forward_metadata(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("cash", 100, "2026-06-01", "cash", db_path=path)
|
|
|
|
live = service.get_asset_snapshot_report("2026-06-30", db_path=path, selection_policy="as_of")
|
|
monthly = service.get_monthly_settlement_summary("2026-06", db_path=path)
|
|
exact = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
|
|
expected_keys = {"warnings", "carried_items", "approval", "correction"}
|
|
assert expected_keys.issubset(live)
|
|
assert live["provenance"]["carried_forward"] is True
|
|
assert any(warning["code"] == "stale_valuation" for warning in live["warnings"])
|
|
assets = monthly["source_coverage"]["assets"]
|
|
assert expected_keys.issubset(assets)
|
|
assert assets["warnings"] == exact["warnings"]
|
|
assert assets["carried_items"] == exact["carried_items"]
|
|
assert assets["approval"] == exact["approval"]
|
|
assert assets["correction"] == exact["correction"]
|
|
|
|
|
|
def test_public_included_items_redact_source_note_and_memo_recursively(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
sentinel = "SHOULD_NOT_LEAK_C10_INCLUDED"
|
|
service.create_manual_asset("cash", 100, "2026-06-01", "cash", source_note=sentinel, memo=sentinel, db_path=path)
|
|
|
|
preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path, selection_policy="as_of")
|
|
closed = service.close_asset_snapshot(
|
|
"2026-06-30", db_path=path, selection_policy="as_of",
|
|
allow_as_of_carry_forward=True, approved_by="reviewer",
|
|
approval_reason="approved carry", approval_reference="APR-59",
|
|
)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
payloads = [preview, closed, report]
|
|
for payload in payloads:
|
|
assert sentinel not in json.dumps(payload.get("included_items", []), ensure_ascii=False)
|
|
assert all("source_note" not in item and "memo" not in item for item in payload.get("included_items", []))
|
|
assert sentinel not in json.dumps(report, ensure_ascii=False)
|
|
|
|
|
|
def test_snapshot_null_amount_remains_null_and_is_excluded_from_totals(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_asset_snapshot("2026-06-30", "manual_asset", "unvalued", "unvalued", None, "cash", db_path=path)
|
|
service.create_asset_snapshot("2026-06-30", "manual_asset", "zero", "zero", 0, "cash", db_path=path)
|
|
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
|
|
amounts = {item["name"]: item["amount"] for item in report["included_items"]}
|
|
assert amounts == {"unvalued": None, "zero": 0}
|
|
assert report["total_assets"] == 0
|
|
assert report["buckets"][0]["item_count"] == 1
|
|
assert report["close_status"] == "incomplete_draft"
|
|
assert any(item["name"] == "unvalued" and item["reason"] == "missing_amount" for item in report["excluded_items"])
|
|
|
|
|
|
def test_idempotent_snapshot_close_replay_returns_original_immutable_response(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("cash", 100, "2026-06-30", "cash", db_path=path)
|
|
kwargs = dict(basis_date="2026-06-30", db_path=path, idempotency_key="close-replay")
|
|
|
|
original = service.close_asset_snapshot(**kwargs)
|
|
replay = service.close_asset_snapshot(**kwargs)
|
|
|
|
assert replay == original
|
|
assert "reused" not in replay
|
|
for key in ("provenance", "approval", "correction", "warnings", "excluded_items", "carried_items", "buckets", "bucket_totals", "total_assets", "total_liabilities", "net_worth"):
|
|
assert replay[key] == original[key]
|
|
|
|
|
|
def test_asset_close_and_report_expose_provenance_aliases(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("cash", 100, "2026-06-01", "cash", db_path=path)
|
|
closed = service.close_asset_snapshot(
|
|
"2026-06-30", db_path=path, selection_policy="as_of",
|
|
allow_as_of_carry_forward=True, approved_by="reviewer",
|
|
approval_reason="approved carry", approval_reference="APR-59",
|
|
)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
for payload in (closed, report):
|
|
assert payload["valuation_date"] == payload["provenance"]["valuation_date"] == "2026-06-01"
|
|
assert payload["age_days"] == payload["provenance"]["age_days"] == 29
|
|
assert payload["verification_status"] == payload["provenance"]["verification_status"] == "verified"
|
|
assert payload["evidence_kind"] == payload["provenance"]["evidence_kind"] == "manual_confirmed"
|
|
assert payload["monthly_source_snapshot_id"] == payload["provenance"]["monthly_source_snapshot_id"] == closed["snapshot_run_id"]
|
|
|
|
|
|
def test_report_invalid_basis_date_returns_stable_error_envelope(tmp_path):
|
|
result = service.get_asset_snapshot_report("not-a-date", db_path=str(tmp_path / "ledger.sqlite3"))
|
|
assert result == {"success": False, "error_code": "validation_error", "error": "invalid date: not-a-date"}
|
|
|
|
|
|
def test_manual_asset_evidence_has_canonical_currency_before_close(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("cash", 100, "2026-06-30", "cash", currency="krw", db_path=path)
|
|
preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path)
|
|
assert preview["included_items"][0]["currency"] == "KRW"
|
|
closed = service.close_asset_snapshot("2026-06-30", db_path=path)
|
|
assert closed["included_items"][0]["currency"] == "KRW"
|
|
with sqlite3.connect(path) as conn:
|
|
evidence = json.loads(conn.execute("SELECT applied_evidence FROM asset_snapshot_runs").fetchone()[0])
|
|
assert evidence[0]["currency"] == "KRW"
|
|
|
|
|
|
def test_close_rejects_authority_drift_since_preview(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("cash", 100, "2026-06-30", "cash", db_path=path)
|
|
preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path)
|
|
assert preview["preview_fingerprint"]
|
|
service.update_manual_asset("asset-does-not-exist", valuation_amount=101, db_path=path)
|
|
# Mutate the real authority row without changing the public API result shape.
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE manual_assets SET valuation_amount=101")
|
|
result = service.close_asset_snapshot("2026-06-30", db_path=path, preview_fingerprint=preview["preview_fingerprint"])
|
|
assert result["success"] is False
|
|
assert result["error_code"] == "authority_drift"
|
|
|
|
|
|
def test_close_metadata_is_allowlisted_recursively(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("cash", 100, "2026-06-30", "cash", db_path=path)
|
|
service.close_asset_snapshot("2026-06-30", db_path=path)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE asset_snapshot_runs SET applied_evidence=?", ('[{"valuation_date":"2026-06-30","future_secret":{"source_note":"LEAK"}}]',))
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
assert "future_secret" not in json.dumps(report["close_metadata"])
|
|
assert "LEAK" not in json.dumps(report["close_metadata"])
|
|
|
|
|
|
def test_public_evidence_redacts_nested_notes_without_mutating_input_or_hash():
|
|
from jarca_ledger_mcp.application.assets_service import _evidence_state_hash, _public_evidence
|
|
|
|
item = {"amount": 0, "source_note": "top", "nested": {"memo": "deep", "keep": [1, {"source_note": "list"}]}}
|
|
before_hash = _evidence_state_hash(item)
|
|
redacted = _public_evidence(item)
|
|
|
|
assert redacted == {"amount": 0, "nested": {"keep": [1, {}]}}
|
|
assert item["nested"]["memo"] == "deep"
|
|
assert _evidence_state_hash(item) == before_hash
|