import sqlite3 import inspect import json from concurrent.futures import ThreadPoolExecutor from jarca_ledger_mcp import service from jarca_ledger_mcp.application.account_valuation_service import ( _public_candidate, record_account_valuation_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_candidate_idempotency_rejects_changed_payload(tmp_path): path = str(tmp_path / "ledger.sqlite3") account = service.create_ledger_account("계좌", db_path=path)["account"]["id"] with sqlite3.connect(path) as conn: conn.row_factory = sqlite3.Row first = record_account_valuation_candidate( conn, account, "2026-06-30", 100, "KRW", "import", "source-a", "candidate-key", "memo-a", ) conflict = record_account_valuation_candidate( conn, account, "2026-07-01", 200, "KRW", "import", "source-b", "candidate-key", "memo-b", ) assert first["success"] is True assert conflict["success"] is False assert conflict["error_code"] == "idempotency_conflict" def test_correction_idempotency_compares_effective_preserved_payload(tmp_path): path = str(tmp_path / "ledger.sqlite3") account = service.create_ledger_account("계좌", db_path=path)["account"]["id"] original = service.register_account_valuation( account, "2026-06-30", 100, memo="original", verification_status="verified", idempotency_key="original", db_path=path, )["valuation"] first = service.correct_account_valuation( original["id"], 200, "정정", memo="override", idempotency_key="correction", db_path=path, ) retry_with_omitted_memo = service.correct_account_valuation( original["id"], 200, "정정", idempotency_key="correction", db_path=path, ) assert first["success"] is True assert retry_with_omitted_memo["success"] is False assert retry_with_omitted_memo["error_code"] == "idempotency_conflict" 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_preview_warns_about_unverified_candidate_when_verified_value_is_selected(tmp_path): path = str(tmp_path / "ledger.sqlite3") account = service.create_ledger_account( "투자", type="investment", db_path=path, )["account"]["id"] service.register_account_valuation( account, "2026-06-30", 100, valuation_kind="total_value", verification_status="verified", idempotency_key="verified", db_path=path, ) service.register_account_valuation( account, "2026-06-30", 999, valuation_kind="total_value", verification_status="unverified", idempotency_key="candidate", db_path=path, ) preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path) warning = next(item for item in preview["warnings"] if item["code"] == "unverified_candidate") assert warning["source_id"] == account assert warning["candidates"][0]["verification_status"] == "unverified" closed = service.close_asset_snapshot("2026-06-30", db_path=path) assert closed["success"] is True with sqlite3.connect(path) as conn: evidence = json.loads(conn.execute( "SELECT excluded_evidence FROM asset_snapshot_runs WHERE id=?", (closed["snapshot_run_id"],), ).fetchone()[0]) assert evidence[0]["reason"] == "unverified_candidate" assert evidence[0]["candidate"]["verification_status"] == "unverified" def test_snapshot_idempotency_binds_redacted_candidate_state(tmp_path): path = str(tmp_path / "ledger.sqlite3") account = service.create_ledger_account( "투자", type="investment", db_path=path, )["account"]["id"] service.register_account_valuation( account, "2026-06-30", 100, valuation_kind="total_value", verification_status="verified", idempotency_key="verified", db_path=path, ) candidate = service.register_account_valuation( account, "2026-06-30", 999, valuation_kind="total_value", verification_status="unverified", idempotency_key="candidate", db_path=path, )["valuation"] first = service.close_asset_snapshot( "2026-06-30", idempotency_key="close-candidate", db_path=path, ) with sqlite3.connect(path) as conn: conn.execute( "UPDATE account_valuations SET amount=?, source_ref=?, memo=? WHERE id=?", (1000, "changed-source", "changed-memo", candidate["id"]), ) retry = service.close_asset_snapshot( "2026-06-30", idempotency_key="close-candidate", db_path=path, ) assert first["success"] is True assert retry["success"] is False assert retry["error_code"] == "idempotency_conflict" def test_reconciliation_fails_closed_on_ambiguous_verified_leaf(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"] for key, amount in (("v1", 100), ("v2", 200)): service.register_account_valuation( account, "2026-06-30", amount, valuation_kind="total_value", verification_status="verified", idempotency_key=key, 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["status"] == "AMBIGUOUS_VALUATION" assert item["external_confirmed_value"] is None assert item["valuation_id"] == "" 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_legacy_snapshot_rows_do_not_override_live_authority_or_claim_closed(tmp_path): path = str(tmp_path / "ledger.sqlite3") service.create_asset_snapshot( "2026-06-30", "account", "legacy", "legacy", 999, "cash", db_path=path, ) account = service.create_ledger_account("현재 계좌", db_path=path)["account"]["id"] service.register_account_valuation( account, "2026-06-30", 100, verification_status="verified", db_path=path, ) report = service.get_asset_snapshot_report("2026-06-30", db_path=path) assert report["source"] == "live_authority" assert report["is_closed_snapshot"] is False assert report["total_assets"] == 100 assert report["legacy_snapshot_row_count"] == 1 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())