253 lines
14 KiB
Python
253 lines
14 KiB
Python
import sqlite3
|
|
import inspect
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.application.account_valuation_service import _public_candidate
|
|
from jarca_ledger_mcp.application.imports_service import _record_import_valuation_candidate
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def test_empty_conditional_close_is_closed_and_second_close_requires_correction(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_ledger_account("미확인 계좌", db_path=path)
|
|
first = service.close_asset_snapshot(
|
|
"2026-06-30", selection_policy="exact", allow_missing_exclusions=True,
|
|
approved_by="reviewer", approval_reason="approved empty close", approval_reference="APR-1", db_path=path,
|
|
)
|
|
assert first["success"] and first["close_status"] == "conditional" and first["inserted_count"] == 0
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
assert report["is_closed_snapshot"] is True and report["snapshot_count"] == 0
|
|
assert report["close_metadata"]["snapshot_run_id"] == first["snapshot_run_id"]
|
|
second = service.close_asset_snapshot(
|
|
"2026-06-30", selection_policy="exact", allow_missing_exclusions=True,
|
|
approved_by="reviewer", approval_reason="again", approval_reference="APR-2", db_path=path,
|
|
)
|
|
assert second["success"] is False and second["error_code"] == "validation_error"
|
|
|
|
|
|
def test_reconciliation_uses_corrected_investment_leaf_without_transaction_count_change(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("투자", type="investment", last_balance=100, last_synced_at="2026-06-30", db_path=path)["account"]["id"]
|
|
original = service.register_account_valuation(account, "2026-06-30", 100, valuation_kind="total_value", verification_status="verified", idempotency_key="v1", db_path=path)
|
|
corrected = service.correct_account_valuation(original["valuation"]["id"], 250, correction_reason="정정", idempotency_key="v2", db_path=path)
|
|
assert corrected["success"]
|
|
report = service.get_balance_reconciliation_report(account_id=account, baseline_at="2026-06-30T00:00:00+09:00", db_path=path)
|
|
assert report["items"][0]["external_confirmed_value"] == 250
|
|
assert report["items"][0]["transaction_count"] == 0
|
|
|
|
|
|
def test_reconciliation_prefers_investment_total_value_over_balance(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"투자", type="investment", last_balance=100,
|
|
last_synced_at="2026-06-30", db_path=path,
|
|
)["account"]["id"]
|
|
total = service.register_account_valuation(
|
|
account, "2026-06-30", 250, valuation_kind="total_value",
|
|
verification_status="verified", idempotency_key="total", db_path=path,
|
|
)
|
|
service.register_account_valuation(
|
|
account, "2026-06-30", 100, valuation_kind="balance",
|
|
verification_status="verified", idempotency_key="balance", db_path=path,
|
|
)
|
|
item = service.get_balance_reconciliation_report(
|
|
account_id=account, baseline_at="2026-06-30T00:00:00+09:00", db_path=path,
|
|
)["items"][0]
|
|
assert item["external_confirmed_value"] == 250
|
|
assert item["valuation_id"] == total["valuation"]["id"]
|
|
assert item["valuation_kind"] == "total_value"
|
|
|
|
|
|
def test_historical_account_update_preserves_current_and_adds_one_candidate(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
created = service.create_ledger_account("계좌", last_balance=900, last_synced_at="2026-06-30", db_path=path)
|
|
account = created["account"]["id"]
|
|
updated = service.update_ledger_account(account, last_balance=700, last_synced_at="2026-06-01", db_path=path)
|
|
assert updated["account"]["last_balance"] == 900 and updated["account"]["last_synced_at"] == "2026-06-30"
|
|
values = service.list_account_valuations(account, db_path=path)["valuations"]
|
|
assert len(values) == 2 and sum(v["basis_date"] == "2026-06-01" for v in values) == 1
|
|
|
|
|
|
def test_schema12_migration_preserves_legacy_rows_indexes_and_is_idempotent(tmp_path):
|
|
path = str(tmp_path / "legacy.sqlite3")
|
|
with sqlite3.connect(path) as conn:
|
|
conn.executescript("""
|
|
CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);
|
|
CREATE TABLE accounts (
|
|
id TEXT PRIMARY KEY, name TEXT NOT NULL, type TEXT NOT NULL DEFAULT 'bank',
|
|
institution TEXT NOT NULL DEFAULT '', currency TEXT NOT NULL DEFAULT 'KRW',
|
|
active INTEGER NOT NULL DEFAULT 1, memo TEXT NOT NULL DEFAULT '',
|
|
created_at TEXT NOT NULL, updated_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE transactions (
|
|
id TEXT PRIMARY KEY, transaction_date TEXT NOT NULL, type TEXT NOT NULL,
|
|
amount INTEGER NOT NULL, category TEXT NOT NULL DEFAULT '',
|
|
subcategory TEXT NOT NULL DEFAULT '', person TEXT NOT NULL DEFAULT '',
|
|
account_id TEXT NOT NULL DEFAULT '', dedupe_key TEXT NOT NULL DEFAULT '',
|
|
created_at TEXT NOT NULL, updated_at TEXT NOT NULL
|
|
);
|
|
CREATE INDEX legacy_transactions_date ON transactions(transaction_date);
|
|
CREATE TABLE asset_snapshots (
|
|
id TEXT PRIMARY KEY, basis_date TEXT NOT NULL, source_type TEXT NOT NULL,
|
|
source_id TEXT NOT NULL DEFAULT '', name TEXT NOT NULL, amount INTEGER,
|
|
currency TEXT NOT NULL DEFAULT 'KRW', liquidity_bucket TEXT NOT NULL DEFAULT 'other',
|
|
valuation_source TEXT NOT NULL DEFAULT '', source_note TEXT NOT NULL DEFAULT '',
|
|
memo TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL
|
|
);
|
|
INSERT INTO schema_meta(key,value) VALUES('schema_version','11');
|
|
INSERT INTO accounts(id,name,type,currency,created_at,updated_at) VALUES('a','old','bank','KRW','t','t');
|
|
INSERT INTO transactions(id,transaction_date,type,amount,created_at,updated_at) VALUES('t','2026-01-01','expense',10,'t','t');
|
|
INSERT INTO asset_snapshots(id,basis_date,source_type,name,amount,created_at) VALUES('s','2026-01-01','account','old',10,'t');
|
|
""")
|
|
init_db(path)
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT name FROM accounts WHERE id='a'").fetchone()
|
|
assert conn.execute("SELECT id FROM transactions WHERE id='t'").fetchone()
|
|
assert conn.execute("SELECT name FROM asset_snapshots WHERE id='s'").fetchone()
|
|
indexes = {r[1] for r in conn.execute("PRAGMA index_list(transactions)")}
|
|
assert {"legacy_transactions_date", "idx_transactions_datetime"} <= indexes
|
|
assert conn.execute("SELECT value FROM schema_meta WHERE key='schema_version'").fetchone()[0] == "12"
|
|
|
|
|
|
def test_snapshot_close_idempotency_fingerprint_and_concurrent_first_close(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_manual_asset("현금", 100, "2026-06-30", "cash", currency="KRW", db_path=path)
|
|
first = service.close_asset_snapshot(
|
|
"2026-06-30", source_note="source-a", memo="memo-a",
|
|
idempotency_key="close-1", db_path=path,
|
|
)
|
|
same = service.close_asset_snapshot(
|
|
"2026-06-30", source_note="source-a", memo="memo-a",
|
|
idempotency_key="close-1", db_path=path,
|
|
)
|
|
changed = service.close_asset_snapshot(
|
|
"2026-06-30", source_note="source-a", memo="memo-b",
|
|
idempotency_key="close-1", db_path=path,
|
|
)
|
|
assert first["success"] and same["success"] and same["reused"] is True
|
|
assert changed["success"] is False and changed["error_code"] == "idempotency_conflict"
|
|
|
|
other_asset = first["included_items"][0]["source_id"]
|
|
service.update_manual_asset(other_asset, memo="changed evidence metadata", db_path=path)
|
|
changed_evidence = service.close_asset_snapshot(
|
|
"2026-06-30", source_note="source-a", memo="memo-a",
|
|
idempotency_key="close-1", db_path=path,
|
|
)
|
|
assert changed_evidence["success"] is False
|
|
assert changed_evidence["error_code"] == "idempotency_conflict"
|
|
|
|
other_path = str(tmp_path / "concurrent.sqlite3")
|
|
service.create_manual_asset("현금", 100, "2026-06-30", "cash", currency="KRW", db_path=other_path)
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = list(pool.map(lambda _: service.close_asset_snapshot("2026-06-30", db_path=other_path), range(2)))
|
|
assert sum(bool(result.get("success")) for result in results) == 1
|
|
with sqlite3.connect(other_path) as conn:
|
|
assert conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs WHERE basis_date='2026-06-30'").fetchone()[0] == 1
|
|
|
|
|
|
def test_import_candidate_helper_reuses_deterministic_duplicate_on_same_connection(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("계좌", db_path=path)["account"]["id"]
|
|
with sqlite3.connect(path) as raw:
|
|
raw.row_factory = sqlite3.Row
|
|
first = _record_import_valuation_candidate(raw, account, 100, "2026-06-30", "sha")
|
|
second = _record_import_valuation_candidate(raw, account, 100, "2026-06-30", "sha")
|
|
count = raw.execute("SELECT count(*) FROM account_valuations WHERE account_id=?", (account,)).fetchone()[0]
|
|
assert first is None and second is None and count == 1
|
|
|
|
|
|
def test_valuation_currency_and_loan_kind_are_rejected(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
bank = service.create_ledger_account("은행", currency="KRW", db_path=path)["account"]["id"]
|
|
loan = service.create_ledger_account("대출", type="loan", db_path=path)["account"]["id"]
|
|
assert service.register_account_valuation(bank, "2026-06-30", 1, currency="USD", db_path=path)["success"] is False
|
|
assert service.register_account_valuation(bank, "2026-06-30", 1, valuation_kind="loan", representative_liability_definition="principal_only", db_path=path)["success"] is False
|
|
assert service.register_account_valuation(loan, "2026-06-30", 1, valuation_kind="loan", representative_liability_definition="principal_only", db_path=path)["success"] is True
|
|
|
|
|
|
def test_loan_representative_is_always_aggregated_as_liability(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
loan = service.create_ledger_account(
|
|
"대출", type="loan", last_balance=100,
|
|
last_synced_at="2026-06-30", db_path=path,
|
|
)["account"]["id"]
|
|
service.register_account_valuation(
|
|
loan, "2026-06-30", 100, valuation_kind="loan",
|
|
representative_liability_definition="principal_only",
|
|
verification_status="verified", db_path=path,
|
|
)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
assert report["total_assets"] == 0
|
|
assert report["total_liabilities"] == 100
|
|
assert report["net_worth"] == -100
|
|
assert report["included_items"][0]["liquidity_bucket"] == "loan"
|
|
|
|
|
|
def test_loan_balance_fallback_without_representative_definition_is_excluded(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
loan = service.create_ledger_account(
|
|
"대출", type="loan", last_balance=100,
|
|
last_synced_at="2026-06-30", db_path=path,
|
|
)["account"]["id"]
|
|
service.register_account_valuation(
|
|
loan, "2026-06-30", 100, valuation_kind="balance",
|
|
verification_status="verified", db_path=path,
|
|
)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
assert report["total_liabilities"] == 0
|
|
assert report["net_worth"] == 0
|
|
assert report["excluded_items"][0]["reason"] == "missing_representative_liability_definition"
|
|
|
|
|
|
def test_warning_candidate_public_payload_is_minimal():
|
|
class Row(dict):
|
|
__getattr__ = dict.__getitem__
|
|
row = Row(id="v", basis_date="2026-06-30", currency="KRW", valuation_kind="balance", verification_status="unverified", memo="secret", source_ref="secret", name="secret")
|
|
payload = _public_candidate(row)
|
|
assert set(payload) == {"id", "basis_date", "currency", "valuation_kind", "verification_status"}
|
|
|
|
|
|
def test_snapshot_report_redacts_unverified_candidate_sensitive_fields(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("계좌", db_path=path)["account"]["id"]
|
|
service.register_account_valuation(
|
|
account, "2026-06-30", 10, verification_status="unverified",
|
|
source_ref="TOPSECRET", memo="PASSWORD=secret", idempotency_key="private",
|
|
db_path=path,
|
|
)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
encoded = repr(report)
|
|
assert "TOPSECRET" not in encoded and "PASSWORD=secret" not in encoded
|
|
candidates = report["excluded_items"][0]["unverified_candidates"]
|
|
assert set(candidates[0]) == {"id", "basis_date", "currency", "valuation_kind", "verification_status"}
|
|
|
|
|
|
def test_snapshot_rejects_bogus_selection_policy_and_manual_usd_is_excluded(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
assert service.preview_asset_snapshot_close("2026-06-30", selection_policy="bogus", db_path=path)["success"] is False
|
|
service.create_manual_asset("달러", 100, "2026-06-30", "other", currency="USD", db_path=path)
|
|
report = service.get_asset_liquidity_report(db_path=path)
|
|
assert report["success"] and report["total_assets"] == 0
|
|
assert any(item.get("reason") == "non_krw_without_fx" for item in report["excluded_items"])
|
|
|
|
|
|
def test_account_write_rolls_back_when_conn_level_candidate_write_fails(tmp_path, monkeypatch):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
import jarca_ledger_mcp.application.accounts_write_services as writes
|
|
monkeypatch.setattr(writes, "_record_balance_candidate", lambda *_args: (_ for _ in ()).throw(RuntimeError("injected")))
|
|
try:
|
|
service.create_ledger_account("원자성", last_balance=1, last_synced_at="2026-06-30", db_path=path)
|
|
except RuntimeError:
|
|
pass
|
|
assert service.list_ledger_accounts(active_only=False, db_path=path)["count"] == 0
|
|
|
|
|
|
def test_mcp_registry_keeps_category_report_and_correction_signature_is_explicit():
|
|
from jarca_ledger_mcp.adapters.mcp import server
|
|
source = inspect.getsource(server)
|
|
assert "def get_category_spending_report" in source
|
|
signature = inspect.signature(server.correct_account_valuation)
|
|
assert not any(param.kind is inspect.Parameter.VAR_KEYWORD for param in signature.parameters.values())
|