263 lines
11 KiB
Python
263 lines
11 KiB
Python
from types import SimpleNamespace
|
|
import sqlite3
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.application import imports_service
|
|
|
|
|
|
def _parsed(transaction, source_hash):
|
|
return {
|
|
"transactions": [transaction],
|
|
"validation": {"ok": True},
|
|
"source_hash": source_hash,
|
|
"metadata": {"final_balance": transaction.balance_after, "account_number_masked": ""},
|
|
"file_path": "/tmp/synthetic.pdf",
|
|
"page_count": 1,
|
|
}
|
|
|
|
|
|
def _transaction(date, balance, row):
|
|
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,
|
|
action="입금",
|
|
external_id=lambda source_hash: f"kb:{source_hash}:{row}",
|
|
raw_payload=lambda: "{}",
|
|
)
|
|
|
|
|
|
def test_late_import_current_balance_is_authoritative_by_evidence_basis_date(monkeypatch, tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"KB", institution="KB국민은행", db_path=path,
|
|
)["account"]["id"]
|
|
imports = iter(
|
|
(
|
|
(_transaction("2026-06-01", 700, 1), "hash-old"),
|
|
(_transaction("2026-06-30", 900, 2), "hash-new"),
|
|
)
|
|
)
|
|
monkeypatch.setattr(imports_service, "parse_kb_bank_pdf", lambda *_: _parsed(*next(imports)))
|
|
sync_times = iter(("2026-07-01T09:00:00", "2026-07-02T09:00:00"))
|
|
monkeypatch.setattr(imports_service, "now_iso", lambda: next(sync_times))
|
|
|
|
older = imports_service.import_kb_bank_pdf("older", "pw", account_id=account, db_path=path)
|
|
newer = imports_service.import_kb_bank_pdf("newer", "pw", account_id=account, db_path=path)
|
|
|
|
current = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0]
|
|
valuations = service.list_account_valuations(account, db_path=path)["valuations"]
|
|
|
|
assert older["success"] and newer["success"]
|
|
assert current["last_balance"] == 900
|
|
assert current["last_synced_at"] == "2026-07-02T09:00:00"
|
|
assert current["last_balance_basis_date"] == "2026-06-30"
|
|
assert {(item["basis_date"], item["amount"]) for item in valuations} >= {
|
|
("2026-06-01", 700),
|
|
("2026-06-30", 900),
|
|
}
|
|
|
|
|
|
def test_empty_balance_with_basis_date_is_rejected_without_mutation_before_import(tmp_path, monkeypatch):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("KB", institution="KB국민은행", db_path=path)["account"]["id"]
|
|
before = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0]
|
|
before_valuations = service.list_account_valuations(account, db_path=path)["valuations"]
|
|
|
|
rejected = service.update_ledger_account(
|
|
account,
|
|
last_balance="",
|
|
last_balance_basis_date="2026-07-01",
|
|
db_path=path,
|
|
)
|
|
|
|
after = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0]
|
|
after_valuations = service.list_account_valuations(account, db_path=path)["valuations"]
|
|
assert rejected["success"] is False
|
|
assert rejected["error_code"] == "validation_error"
|
|
assert after == before
|
|
assert after_valuations == before_valuations
|
|
|
|
monkeypatch.setattr(
|
|
imports_service,
|
|
"parse_kb_bank_pdf",
|
|
lambda *_: _parsed(_transaction("2026-06-30", 900, 1), "hash-valid"),
|
|
)
|
|
monkeypatch.setattr(imports_service, "now_iso", lambda: "2026-07-02T09:00:00")
|
|
imported = imports_service.import_kb_bank_pdf("synthetic", "pw", account_id=account, db_path=path)
|
|
|
|
current = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0]
|
|
valuations = service.list_account_valuations(account, db_path=path)["valuations"]
|
|
assert imported["success"] is True
|
|
assert current["last_balance"] == 900
|
|
assert current["last_balance_basis_date"] == "2026-06-30"
|
|
assert {(item["basis_date"], item["amount"]) for item in valuations} == {("2026-06-30", 900)}
|
|
|
|
|
|
def test_create_rejects_whitespace_or_non_integer_dated_balance_without_mutation(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
baseline = service.create_ledger_account("baseline", db_path=path)
|
|
before = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"]
|
|
before_valuations = service.list_account_valuations(baseline["account"]["id"], db_path=path)["valuations"]
|
|
|
|
for invalid_balance in (" ", "not-an-integer", True, False, 1.2, -1.2):
|
|
rejected = service.create_ledger_account(
|
|
"invalid",
|
|
last_balance=invalid_balance,
|
|
last_balance_basis_date="2026-07-01",
|
|
db_path=path,
|
|
)
|
|
|
|
assert rejected["success"] is False
|
|
assert rejected["error_code"] == "validation_error"
|
|
assert service.list_ledger_accounts(active_only=False, db_path=path)["accounts"] == before
|
|
assert service.list_account_valuations(baseline["account"]["id"], db_path=path)["valuations"] == before_valuations
|
|
|
|
|
|
def test_update_rejects_bool_or_float_dated_balance_without_mutation(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"baseline",
|
|
last_balance=900,
|
|
last_balance_basis_date="2026-07-01",
|
|
db_path=path,
|
|
)["account"]["id"]
|
|
before = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0]
|
|
before_valuations = service.list_account_valuations(account, db_path=path)["valuations"]
|
|
|
|
for invalid_balance in (True, False, 1.2, -1.2):
|
|
rejected = service.update_ledger_account(
|
|
account,
|
|
last_balance=invalid_balance,
|
|
last_balance_basis_date="2026-07-02",
|
|
db_path=path,
|
|
)
|
|
|
|
assert rejected["success"] is False
|
|
assert rejected["error_code"] == "validation_error"
|
|
assert service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] == before
|
|
assert service.list_account_valuations(account, db_path=path)["valuations"] == before_valuations
|
|
|
|
|
|
def test_update_rejects_empty_undated_balance_without_mutation(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"baseline",
|
|
last_balance=900,
|
|
last_balance_basis_date="2026-07-01",
|
|
db_path=path,
|
|
)["account"]["id"]
|
|
before = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0]
|
|
before_valuations = service.list_account_valuations(account, db_path=path)["valuations"]
|
|
|
|
rejected = service.update_ledger_account(account, last_balance="", db_path=path)
|
|
|
|
assert rejected["success"] is False
|
|
assert rejected["error_code"] == "validation_error"
|
|
assert service.list_ledger_accounts(active_only=False, db_path=path)["accounts"][0] == before
|
|
assert service.list_account_valuations(account, db_path=path)["valuations"] == before_valuations
|
|
|
|
|
|
def test_update_preserves_zero_and_signed_integer_balance_semantics(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("baseline", db_path=path)["account"]["id"]
|
|
|
|
zero = service.update_ledger_account(
|
|
account,
|
|
last_balance=" +0 ",
|
|
last_balance_basis_date="2026-07-01",
|
|
db_path=path,
|
|
)
|
|
signed = service.update_ledger_account(
|
|
account,
|
|
last_balance="-12",
|
|
last_balance_basis_date="2026-07-02",
|
|
db_path=path,
|
|
)
|
|
|
|
assert zero["success"] is True
|
|
assert zero["account"]["last_balance"] == 0
|
|
assert signed["success"] is True
|
|
assert signed["account"]["last_balance"] == -12
|
|
assert signed["account"]["last_balance_basis_date"] == "2026-07-02"
|
|
assert [(item["basis_date"], item["amount"]) for item in service.list_account_valuations(account, db_path=path)["valuations"]] == [
|
|
("2026-07-01", 0),
|
|
("2026-07-02", -12),
|
|
]
|
|
|
|
|
|
def test_zero_dated_balance_and_valid_import_recovery(tmp_path, monkeypatch):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
zero = service.create_ledger_account(
|
|
"zero",
|
|
last_balance=0,
|
|
last_balance_basis_date="2026-06-01",
|
|
db_path=path,
|
|
)
|
|
|
|
assert zero["success"] is True
|
|
assert zero["account"]["last_balance"] == 0
|
|
assert [(item["basis_date"], item["amount"]) for item in service.list_account_valuations(zero["account"]["id"], db_path=path)["valuations"]] == [("2026-06-01", 0)]
|
|
|
|
account = service.create_ledger_account("recoverable", db_path=path)["account"]["id"]
|
|
monkeypatch.setattr(
|
|
imports_service,
|
|
"parse_kb_bank_pdf",
|
|
lambda *_: _parsed(_transaction("2026-06-30", 900, 1), "hash-valid"),
|
|
)
|
|
monkeypatch.setattr(imports_service, "now_iso", lambda: "2026-07-02T09:00:00")
|
|
|
|
imported = imports_service.import_kb_bank_pdf("synthetic", "pw", account_id=account, db_path=path)
|
|
|
|
current = next(item for item in service.list_ledger_accounts(active_only=False, db_path=path)["accounts"] if item["id"] == account)
|
|
assert imported["success"] is True
|
|
assert current["last_balance"] == 900
|
|
assert current["last_balance_basis_date"] == "2026-06-30"
|
|
|
|
|
|
def test_account_write_sync_timestamp_does_not_infer_balance_basis_or_candidate(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
unknown = service.create_ledger_account(
|
|
"기준일 미상",
|
|
last_balance=500,
|
|
last_synced_at="2026-07-01T09:00:00+09:00",
|
|
db_path=path,
|
|
)
|
|
|
|
assert unknown["success"] is True
|
|
assert unknown["account"]["last_balance_basis_date"] == ""
|
|
assert service.list_account_valuations(unknown["account"]["id"], db_path=path)["count"] == 0
|
|
|
|
dated = service.create_ledger_account(
|
|
"기준일 확인",
|
|
last_balance=600,
|
|
last_synced_at="2026-07-01T09:00:00+09:00",
|
|
db_path=path,
|
|
last_balance_basis_date="2026-06-30",
|
|
)
|
|
valuations = service.list_account_valuations(dated["account"]["id"], db_path=path)["valuations"]
|
|
assert dated["account"]["last_balance_basis_date"] == "2026-06-30"
|
|
assert [(item["basis_date"], item["amount"]) for item in valuations] == [("2026-06-30", 600)]
|
|
|
|
|
|
def test_unknown_legacy_balance_basis_is_not_reported_as_valuation_date(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("legacy", db_path=path)["account"]["id"]
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute(
|
|
"UPDATE accounts SET last_balance=?, last_synced_at=?, last_balance_basis_date='' WHERE id=?",
|
|
(500, "2026-07-01T09:00:00+09:00", account),
|
|
)
|
|
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=path)
|
|
excluded = next(item for item in report["excluded_items"] if item["source_id"] == account)
|
|
assert excluded["valuation_date"] == ""
|
|
assert excluded["last_synced_at"] == "2026-07-01T09:00:00+09:00"
|