191 lines
11 KiB
Python
191 lines
11 KiB
Python
from types import SimpleNamespace
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.application import imports_service
|
|
|
|
|
|
def _account(path, *, account_type="bank", currency="KRW"):
|
|
return service.create_ledger_account("계좌", type=account_type, currency=currency, db_path=path)["account"]["id"]
|
|
|
|
|
|
def test_valuation_round_trip_preserves_null_zero_and_physical_amount_mapping(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
investment = _account(path, account_type="investment")
|
|
loan = _account(path, account_type="loan")
|
|
inv = service.register_account_valuation(
|
|
investment, "2026-06-30", 1_000, valuation_kind="total_value",
|
|
principal=0, reported_principal_kind="contributed", reported_principal_label="납입원금",
|
|
interest=None, fee=None, idempotency_key="inv", db_path=path,
|
|
)
|
|
debt = service.register_account_valuation(
|
|
loan, "2026-06-30", 9_000, valuation_kind="loan",
|
|
representative_liability_definition="principal_interest_fees", principal=9_000,
|
|
interest=300, fee=20, idempotency_key="loan", db_path=path,
|
|
)
|
|
assert inv["success"] and debt["success"]
|
|
inv_value = service.list_account_valuations(investment, db_path=path)["valuations"][0]
|
|
debt_value = service.list_account_valuations(loan, db_path=path)["valuations"][0]
|
|
assert inv_value["principal"] == 0
|
|
assert inv_value["interest"] is None and inv_value["fee"] is None
|
|
assert inv_value["reported_principal_kind"] == "contributed"
|
|
assert inv_value["reported_principal_label"] == "납입원금"
|
|
assert debt_value["amount"] == 9_000
|
|
assert debt_value["principal"] == 9_000 and debt_value["interest"] == 300 and debt_value["fee"] == 20
|
|
|
|
|
|
def test_direct_correction_requires_matching_existing_parent_and_reason(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = _account(path)
|
|
parent = service.register_account_valuation(account, "2026-06-30", 100, db_path=path)["valuation"]
|
|
for kwargs in (
|
|
{"supersedes_id": "missing", "correction_reason": "정정"},
|
|
{"supersedes_id": parent["id"], "correction_reason": ""},
|
|
):
|
|
result = service.register_account_valuation(account, "2026-06-30", 200, db_path=path, **kwargs)
|
|
assert result["success"] is False
|
|
other = _account(path, currency="USD")
|
|
result = service.register_account_valuation(
|
|
other, "2026-06-30", 200, currency="USD", supersedes_id=parent["id"],
|
|
correction_reason="cross account", db_path=path,
|
|
)
|
|
assert result["success"] is False
|
|
assert service.list_account_valuations(account, db_path=path)["count"] == 1
|
|
|
|
|
|
def test_correction_rejects_currency_and_kind_override_but_preserves_valid_overrides(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = _account(path, account_type="investment")
|
|
parent = service.register_account_valuation(
|
|
account, "2026-06-30", 100, valuation_kind="total_value", memo="old", db_path=path,
|
|
)["valuation"]
|
|
bad_currency = service.correct_account_valuation(parent["id"], 110, "정정", currency="USD", db_path=path)
|
|
bad_kind = service.correct_account_valuation(parent["id"], 110, "정정", valuation_kind="balance", db_path=path)
|
|
good = service.correct_account_valuation(parent["id"], 110, "정정", memo="new", db_path=path)
|
|
assert bad_currency["success"] is False and bad_kind["success"] is False
|
|
assert good["success"] and good["valuation"]["memo"] == "new"
|
|
|
|
|
|
def _fake_tx(date, balance, *, row=1, kind="bank"):
|
|
if kind == "kb":
|
|
return SimpleNamespace(
|
|
transaction_date=date, transaction_datetime=f"{date} 10:00:00", amount=100,
|
|
counterparty="상대", raw_summary="이체", branch="", remittance_memo="",
|
|
balance_after=balance, type="income", page=1,
|
|
external_id=lambda source_hash: f"kb:{source_hash}:{row}",
|
|
raw_payload=lambda: "{}",
|
|
)
|
|
if kind == "tosssec":
|
|
return SimpleNamespace(
|
|
transaction_date=date, amount=100, action="매수", name="종목", code="A", section="거래",
|
|
quantity=1, unit_price=100, fee=0, tax=0, balance_after=balance, type="income",
|
|
external_id=lambda source_hash: f"sec:{source_hash}:{row}", raw_payload=lambda: "{}",
|
|
)
|
|
return SimpleNamespace(
|
|
transaction_date=date, transaction_datetime=f"{date} 10:00:00", amount=100,
|
|
action="입금", counterparty="상대", balance_after=balance, type="income", raw_summary="입금",
|
|
external_id=lambda source_hash: f"bank:{source_hash}:{row}", raw_payload=lambda: "{}",
|
|
)
|
|
|
|
|
|
def _parsed(tx, source, metadata):
|
|
return {
|
|
"transactions": [tx], "validation": {"ok": True}, "source_hash": f"hash-{source}",
|
|
"metadata": metadata, "file_path": "/tmp/synthetic.pdf", "page_count": 1,
|
|
}
|
|
|
|
|
|
def test_import_historical_kb_does_not_replace_newer_current_balance(monkeypatch, tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("KB", institution="KB국민은행", last_balance=900, last_synced_at="2026-06-30", db_path=path)["account"]["id"]
|
|
monkeypatch.setattr(imports_service, "parse_kb_bank_pdf", lambda *_: _parsed(_fake_tx("2026-06-01", 700, kind="kb"), "kb", {"final_balance": 700, "account_number_masked": ""}))
|
|
result = imports_service.import_kb_bank_pdf("ignored", "pw", account_id=account, db_path=path)
|
|
current = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0]
|
|
assert result["success"] and current["last_balance"] == 900 and current["last_synced_at"] == "2026-06-30"
|
|
assert service.list_account_valuations(account, db_path=path)["count"] == 2
|
|
|
|
|
|
def test_import_historical_toss_bank_does_not_replace_newer_current_balance(monkeypatch, tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("Toss", institution="토스뱅크", last_balance=900, last_synced_at="2026-06-30", db_path=path)["account"]["id"]
|
|
monkeypatch.setattr(imports_service, "parse_toss_bank_pdf", lambda *_: _parsed(_fake_tx("2026-06-01", 700), "bank", {"account_number_masked": ""}))
|
|
result = imports_service.import_toss_bank_pdf("ignored", account_id=account, db_path=path)
|
|
current = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0]
|
|
assert result["success"] and current["last_balance"] == 900 and current["last_synced_at"] == "2026-06-30"
|
|
assert service.list_account_valuations(account, db_path=path)["count"] == 2
|
|
|
|
|
|
def test_import_historical_toss_securities_does_not_replace_newer_current_balance(monkeypatch, tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("Securities", type="investment", institution="토스증권", last_balance=900, last_synced_at="2026-06-30", db_path=path)["account"]["id"]
|
|
monkeypatch.setattr(imports_service, "parse_toss_securities_pdf", lambda *_: _parsed(_fake_tx("2026-06-01", 700, kind="tosssec"), "sec", {"account_number_masked": ""}))
|
|
result = imports_service.import_toss_securities_pdf("ignored", account_id=account, db_path=path)
|
|
current = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0]
|
|
assert result["success"] and current["last_balance"] == 900 and current["last_synced_at"] == "2026-06-30"
|
|
assert service.list_account_valuations(account, db_path=path)["count"] == 2
|
|
|
|
|
|
def test_source_ref_is_account_scoped_evidence_identity_without_explicit_key(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
first = _account(path)
|
|
second = _account(path)
|
|
one = service.register_account_valuation(first, "2026-06-30", 100, source_ref="statement", db_path=path)
|
|
same = service.register_account_valuation(first, "2026-06-30", 100, source_ref="statement", db_path=path)
|
|
changed = service.register_account_valuation(first, "2026-06-30", 101, source_ref="statement", db_path=path)
|
|
independent = service.register_account_valuation(second, "2026-06-30", 101, source_ref="statement", db_path=path)
|
|
empty_a = service.register_account_valuation(first, "2026-07-01", 1, source_ref="", db_path=path)
|
|
empty_b = service.register_account_valuation(first, "2026-07-01", 1, source_ref="", db_path=path)
|
|
assert one["success"] and same["reused"] is True
|
|
assert changed["success"] is False and changed["error_code"] == "idempotency_conflict"
|
|
assert independent["success"] and independent["reused"] is False
|
|
assert empty_a["success"] and empty_b["success"] and empty_b["reused"] is False
|
|
|
|
|
|
def test_correction_rejects_self_parent_and_cycles_without_write(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = _account(path)
|
|
parent = service.register_account_valuation(account, "2026-06-30", 100, db_path=path)["valuation"]
|
|
child = service.register_account_valuation(account, "2026-06-30", 101, supersedes_id=parent["id"], correction_reason="r", db_path=path)["valuation"]
|
|
from jarca_ledger_mcp.db import db
|
|
with db(path) as conn:
|
|
conn.execute("UPDATE account_valuations SET supersedes_id=? WHERE id=?", (child["id"], child["id"]))
|
|
cyclic = service.register_account_valuation(account, "2026-06-30", 102, supersedes_id=child["id"], correction_reason="r", db_path=path)
|
|
assert cyclic["success"] is False and cyclic["error_code"] == "validation_error"
|
|
assert service.list_account_valuations(account, db_path=path)["count"] == 2
|
|
|
|
|
|
def test_correction_preserves_source_ref_without_evidence_identity_conflict(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = _account(path)
|
|
parent = service.register_account_valuation(
|
|
account, "2026-06-30", 100, source_ref="statement-1", db_path=path,
|
|
)["valuation"]
|
|
|
|
corrected = service.correct_account_valuation(
|
|
parent["id"], 110, "정정", idempotency_key="correction-1", db_path=path,
|
|
)
|
|
replay = service.correct_account_valuation(
|
|
parent["id"], 110, "정정", idempotency_key="correction-1", db_path=path,
|
|
)
|
|
unrelated = service.register_account_valuation(
|
|
account, "2026-06-30", 120, source_ref="statement-1", idempotency_key="unrelated", db_path=path,
|
|
)
|
|
listed = service.list_account_valuations(account, db_path=path)
|
|
|
|
assert corrected["success"]
|
|
assert corrected["valuation"]["source_ref"] == "statement-1"
|
|
assert replay["success"] and replay["reused"] is True
|
|
assert replay["valuation"]["id"] == corrected["valuation"]["id"]
|
|
assert unrelated["success"] is False and unrelated["error_code"] == "idempotency_conflict"
|
|
assert listed["count"] == 2
|
|
assert sum(item["supersedes_id"] is not None for item in listed["valuations"]) == 1
|
|
|
|
|
|
def test_source_ref_conflicts_even_when_idempotency_keys_differ(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = _account(path)
|
|
first = service.register_account_valuation(account, "2026-06-30", 100, source_ref="statement", idempotency_key="a", db_path=path)
|
|
same = service.register_account_valuation(account, "2026-06-30", 100, source_ref="statement", idempotency_key="b", db_path=path)
|
|
changed = service.register_account_valuation(account, "2026-06-30", 101, source_ref="statement", idempotency_key="c", db_path=path)
|
|
assert first["success"] and same["reused"] is True
|
|
assert changed["success"] is False and changed["error_code"] == "idempotency_conflict"
|