673 lines
32 KiB
Python
673 lines
32 KiB
Python
import sqlite3
|
|
import inspect
|
|
import json
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.application import assets_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_preview_marks_approval_eligible_missing_valuation_as_incomplete_draft(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_ledger_account("미확인 계좌", db_path=path)
|
|
preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path)
|
|
assert preview["success"] is True
|
|
assert preview["close_status"] == "incomplete_draft"
|
|
assert any(warning["code"] == "missing_verified_exact_valuation" for warning in preview["warnings"])
|
|
|
|
|
|
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, last_balance_basis_date="2026-06-30")
|
|
account = created["account"]["id"]
|
|
updated = service.update_ledger_account(account, last_balance=700, last_synced_at="2026-06-01", last_balance_basis_date="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 == first
|
|
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
|
|
conflict = next(result for result in results if not result.get("success"))
|
|
assert "detail" not in conflict
|
|
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_snapshot_integrity_conflict_does_not_expose_sqlite_detail(tmp_path, monkeypatch):
|
|
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", db_path=path)
|
|
real_new_id = assets_service.new_id
|
|
|
|
def colliding_run_id(prefix):
|
|
return first["snapshot_run_id"] if prefix == "snaprun" else real_new_id(prefix)
|
|
|
|
monkeypatch.setattr(assets_service, "new_id", colliding_run_id)
|
|
conflict = service.close_asset_snapshot(
|
|
"2026-06-30", correction_reason="retry collision", db_path=path,
|
|
)
|
|
assert conflict["success"] is False
|
|
assert conflict["error_code"] == "snapshot_conflict"
|
|
assert conflict["error"] == "another snapshot close already owns this basis date"
|
|
assert "detail" not in conflict
|
|
|
|
|
|
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_concurrent_valuation_idempotency_reuses_same_payload(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("계좌", db_path=path)["account"]["id"]
|
|
|
|
def register(_):
|
|
return service.register_account_valuation(
|
|
account, "2026-06-30", 100, verification_status="verified",
|
|
idempotency_key="same", db_path=path,
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = list(pool.map(register, range(2)))
|
|
assert all(result["success"] for result in results)
|
|
assert len({result["valuation"]["id"] for result in results}) == 1
|
|
assert sorted(result["reused"] for result in results) == [False, True]
|
|
|
|
|
|
def test_concurrent_correction_idempotency_reuses_same_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, verification_status="verified", db_path=path,
|
|
)["valuation"]
|
|
|
|
def correct(_):
|
|
return service.correct_account_valuation(
|
|
original["id"], 200, "정정", idempotency_key="same-correction", db_path=path,
|
|
)
|
|
|
|
with ThreadPoolExecutor(max_workers=2) as pool:
|
|
results = list(pool.map(correct, range(2)))
|
|
assert all(result["success"] for result in results)
|
|
assert len({result["valuation"]["id"] for result in results}) == 1
|
|
assert sorted(result["reused"] for result in results) == [False, True]
|
|
|
|
|
|
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_without_representative_definition_is_rejected_without_write(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"]
|
|
rejected = service.register_account_valuation(
|
|
loan, "2026-06-30", 100, valuation_kind="balance",
|
|
verification_status="verified", db_path=path,
|
|
)
|
|
assert rejected["success"] is False
|
|
assert rejected["error_code"] == "validation_error"
|
|
assert service.list_account_valuations(loan, db_path=path)["count"] == 0
|
|
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_verified_exact_valuation"
|
|
|
|
|
|
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_requires_balance_coverage_through_requested_basis(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"계좌", last_balance=1_000, last_synced_at="2026-06-30", db_path=path,
|
|
)["account"]["id"]
|
|
service.create_ledger_transaction(
|
|
"2026-06-01", "income", 1_000, account_id=account,
|
|
balance_after=1_000, db_path=path,
|
|
)
|
|
|
|
item = service.get_balance_reconciliation_report(
|
|
account_id=account, baseline_balance=1_000,
|
|
baseline_at="2026-06-30T00:00:00+09:00", db_path=path,
|
|
)["items"][0]
|
|
|
|
assert item["calculation_basis"] == "latest_transaction_balance_after"
|
|
assert item["calculation_complete"] is False
|
|
assert item["status"] == "INCOMPLETE_CALCULATION"
|
|
assert item["difference"] is None
|
|
|
|
|
|
def test_reconciliation_accepts_balance_coverage_at_requested_basis(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"계좌", last_balance=1_000, last_synced_at="2026-06-30", db_path=path,
|
|
)["account"]["id"]
|
|
service.create_ledger_transaction(
|
|
"2026-06-30", "income", 1_000, account_id=account,
|
|
balance_after=1_000, db_path=path,
|
|
)
|
|
|
|
item = service.get_balance_reconciliation_report(
|
|
account_id=account, baseline_balance=1_000,
|
|
baseline_at="2026-06-30T00:00:00+09:00", db_path=path,
|
|
)["items"][0]
|
|
|
|
assert item["calculation_complete"] is True
|
|
assert item["status"] == "OK"
|
|
assert item["difference"] == 0
|
|
|
|
|
|
def test_reconciliation_requires_start_balance_for_transaction_coverage(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"계좌", last_balance=1_000, last_synced_at="2026-06-30", db_path=path,
|
|
)["account"]["id"]
|
|
service.create_ledger_transaction(
|
|
"2026-06-01", "income", 1_000, account_id=account, db_path=path,
|
|
)
|
|
service.create_ledger_transaction(
|
|
"2026-06-30", "expense", 100, account_id=account,
|
|
balance_after=900, db_path=path,
|
|
)
|
|
|
|
item = service.get_balance_reconciliation_report(
|
|
account_id=account, baseline_balance=900,
|
|
baseline_at="2026-06-30T00:00:00+09:00", db_path=path,
|
|
)["items"][0]
|
|
|
|
assert item["calculation_complete"] is False
|
|
assert item["status"] == "INCOMPLETE_CALCULATION"
|
|
assert item["difference"] is None
|
|
|
|
|
|
def test_reconciliation_requires_authoritative_opening_evidence_for_multi_transaction_coverage(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"계좌", last_balance=1_000, last_synced_at="2026-06-30", db_path=path,
|
|
)["account"]["id"]
|
|
service.create_ledger_transaction(
|
|
"2026-06-15", "income", 100, account_id=account,
|
|
balance_after=900, db_path=path,
|
|
)
|
|
service.create_ledger_transaction(
|
|
"2026-06-30", "income", 100, account_id=account,
|
|
balance_after=1_000, db_path=path,
|
|
)
|
|
|
|
item = service.get_balance_reconciliation_report(
|
|
account_id=account, baseline_balance=1_000,
|
|
baseline_at="2026-06-30T00:00:00+09:00", db_path=path,
|
|
)["items"][0]
|
|
|
|
assert item["calculation_complete"] is False
|
|
assert item["status"] == "INCOMPLETE_CALCULATION"
|
|
assert item["difference"] is None
|
|
|
|
|
|
def test_reconciliation_accepts_continuous_transactions_with_verified_opening_evidence(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"계좌", last_balance=1_000, last_synced_at="2026-06-30", db_path=path,
|
|
)["account"]["id"]
|
|
service.register_account_valuation(
|
|
account, "2026-06-14", 800, valuation_kind="balance",
|
|
verification_status="verified", db_path=path,
|
|
)
|
|
service.create_ledger_transaction(
|
|
"2026-06-15", "income", 100, account_id=account,
|
|
balance_after=900, db_path=path,
|
|
)
|
|
service.create_ledger_transaction(
|
|
"2026-06-30", "income", 100, account_id=account,
|
|
balance_after=1_000, db_path=path,
|
|
)
|
|
|
|
item = service.get_balance_reconciliation_report(
|
|
account_id=account, baseline_balance=1_000,
|
|
baseline_at="2026-06-30T00:00:00+09:00", db_path=path,
|
|
)["items"][0]
|
|
|
|
assert item["calculation_complete"] is True
|
|
assert item["status"] == "OK"
|
|
assert item["difference"] == 0
|
|
|
|
|
|
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_non_krw_account_valuation_is_excluded_without_fx_conversion(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("달러 계좌", currency="USD", db_path=path)["account"]["id"]
|
|
service.register_account_valuation(
|
|
account, "2026-06-30", 100, currency="USD",
|
|
verification_status="verified", db_path=path,
|
|
)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
assert report["total_assets"] == 0
|
|
assert any(item.get("reason") == "non_krw_without_fx" for item in report["excluded_items"])
|
|
|
|
|
|
def test_legacy_lowercase_krw_account_currency_selects_normalized_valuation(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("계좌", currency="KRW", db_path=path)["account"]["id"]
|
|
service.register_account_valuation(
|
|
account, "2026-06-30", 100, currency="KRW",
|
|
verification_status="verified", db_path=path,
|
|
)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE accounts SET currency='krw' WHERE id=?", (account,))
|
|
conn.execute("UPDATE account_valuations SET currency='krw' WHERE account_id=?", (account,))
|
|
selected = service.select_account_valuation(
|
|
account, "2026-06-30", currency="KRW", db_path=path,
|
|
)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
assert selected["valuation"]["amount"] == 100
|
|
assert report["total_assets"] == 100
|
|
assert report["excluded_items"] == []
|
|
|
|
|
|
def test_rejected_valuation_is_not_an_unverified_candidate(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", 100, verification_status="verified", db_path=path,
|
|
)
|
|
service.register_account_valuation(
|
|
account, "2026-06-30", 999, verification_status="rejected", db_path=path,
|
|
)
|
|
selected = service.select_account_valuation(account, "2026-06-30", db_path=path)
|
|
preview = service.preview_asset_snapshot_close("2026-06-30", db_path=path)
|
|
assert selected.get("warning") is None
|
|
assert selected.get("unverified_candidates", []) == []
|
|
assert not any(warning.get("code") == "unverified_candidate" for warning in preview["warnings"])
|
|
|
|
|
|
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_snapshot_public_api_preserves_legacy_db_path_position(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
init_db(path)
|
|
preview = service.preview_asset_snapshot_close("2026-06-30", path)
|
|
closed = service.close_asset_snapshot("2026-06-30", "", "", "", path)
|
|
report = service.get_asset_snapshot_report("2026-06-30", path)
|
|
assert preview["success"] is True
|
|
assert closed["success"] is True
|
|
assert report["success"] is True
|
|
assert report["is_closed_snapshot"] is True
|
|
|
|
|
|
def test_legacy_snapshot_requires_explicit_correction_before_new_close(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_asset_snapshot(
|
|
"2026-06-30", "account", "legacy", "legacy", 999, "cash", db_path=path,
|
|
)
|
|
close = service.close_asset_snapshot("2026-06-30", db_path=path)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
assert close["success"] is False
|
|
assert close["error_code"] == "legacy_snapshot_correction_required"
|
|
assert report["source"] == "legacy_asset_snapshots"
|
|
assert report["net_worth"] == 999
|
|
assert report["is_closed_snapshot"] is False
|
|
|
|
|
|
def test_loan_balance_fallback_with_representative_definition_is_liability(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("대출", type="loan", db_path=path)["account"]["id"]
|
|
recorded = service.register_account_valuation(
|
|
account, "2026-06-30", 100, valuation_kind="balance",
|
|
representative_liability_definition="principal_only",
|
|
verification_status="verified", db_path=path,
|
|
)
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
assert recorded["success"] is True
|
|
assert report["total_assets"] == 0
|
|
assert report["total_liabilities"] == 100
|
|
assert report["net_worth"] == -100
|
|
|
|
|
|
def test_loan_balance_fallback_rejects_unknown_representative_definition(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("대출", type="loan", db_path=path)["account"]["id"]
|
|
recorded = service.register_account_valuation(
|
|
account, "2026-06-30", 100, valuation_kind="balance",
|
|
representative_liability_definition="outstanding_principal",
|
|
verification_status="verified", db_path=path,
|
|
)
|
|
assert recorded["success"] is False
|
|
assert recorded["error_code"] == "validation_error"
|
|
|
|
|
|
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())
|