199 lines
7.4 KiB
Python
199 lines
7.4 KiB
Python
import sqlite3
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.application.account_valuation_service import record_account_valuation_candidate
|
|
from jarca_ledger_mcp.application.imports_service import _record_import_valuation_candidate
|
|
|
|
|
|
def test_loan_balance_requires_representative_liability_definition(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
bank = service.create_ledger_account("은행", type="bank", db_path=path)["account"]["id"]
|
|
loan = service.create_ledger_account("대출", type="loan", db_path=path)["account"]["id"]
|
|
|
|
missing_definition = service.register_account_valuation(
|
|
loan,
|
|
"2026-06-30",
|
|
100,
|
|
valuation_kind="balance",
|
|
verification_status="verified",
|
|
db_path=path,
|
|
)
|
|
assert missing_definition["success"] is False
|
|
assert missing_definition["error_code"] == "validation_error"
|
|
assert service.list_account_valuations(loan, db_path=path)["count"] == 0
|
|
|
|
ordinary_balance = service.register_account_valuation(
|
|
bank,
|
|
"2026-06-30",
|
|
100,
|
|
valuation_kind="balance",
|
|
verification_status="verified",
|
|
db_path=path,
|
|
)
|
|
defined_loan_balance = service.register_account_valuation(
|
|
loan,
|
|
"2026-06-30",
|
|
100,
|
|
valuation_kind="balance",
|
|
representative_liability_definition="principal_only",
|
|
verification_status="verified",
|
|
db_path=path,
|
|
)
|
|
assert ordinary_balance["success"] is True
|
|
assert defined_loan_balance["success"] is True
|
|
|
|
|
|
def test_correction_replays_by_evidence_and_conflicts_on_changed_payload(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("계좌", db_path=path)["account"]["id"]
|
|
parent = service.register_account_valuation(
|
|
account,
|
|
"2026-06-30",
|
|
100,
|
|
source_ref="statement-1",
|
|
verification_status="verified",
|
|
db_path=path,
|
|
)["valuation"]
|
|
|
|
first = service.correct_account_valuation(
|
|
parent["id"],
|
|
110,
|
|
"statement correction",
|
|
db_path=path,
|
|
)
|
|
replay = service.correct_account_valuation(
|
|
parent["id"],
|
|
110,
|
|
"statement correction",
|
|
db_path=path,
|
|
)
|
|
changed_amount = service.correct_account_valuation(
|
|
parent["id"],
|
|
111,
|
|
"statement correction",
|
|
db_path=path,
|
|
)
|
|
changed_reason = service.correct_account_valuation(
|
|
parent["id"],
|
|
110,
|
|
"different correction",
|
|
db_path=path,
|
|
)
|
|
changed_source = service.correct_account_valuation(
|
|
parent["id"],
|
|
110,
|
|
"statement correction",
|
|
source_ref="statement-2",
|
|
db_path=path,
|
|
)
|
|
|
|
assert first["success"] is True
|
|
assert replay["success"] is True
|
|
assert replay["reused"] is True
|
|
assert replay["valuation"]["id"] == first["valuation"]["id"]
|
|
for changed in (changed_amount, changed_reason, changed_source):
|
|
assert changed["success"] is False
|
|
assert changed["error_code"] == "correction_conflict"
|
|
assert service.list_account_valuations(account, db_path=path)["count"] == 2
|
|
|
|
|
|
def test_direct_loan_candidate_requires_liability_definition_and_writes_nothing(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
loan = service.create_ledger_account("대출", type="loan", db_path=path)["account"]["id"]
|
|
|
|
with sqlite3.connect(path) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
rejected = record_account_valuation_candidate(
|
|
conn, loan, "2026-06-30", 100, "KRW", "import", "source", "candidate", "memo",
|
|
)
|
|
|
|
assert rejected["success"] is False
|
|
assert rejected["error_code"] == "validation_error"
|
|
assert service.list_account_valuations(loan, db_path=path)["count"] == 0
|
|
|
|
|
|
def test_import_loan_candidate_requires_liability_definition_and_writes_nothing(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
loan = service.create_ledger_account("대출", type="loan", db_path=path)["account"]["id"]
|
|
|
|
with sqlite3.connect(path) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
assert _record_import_valuation_candidate(conn, loan, 100, "2026-06-30", "source-hash") is None
|
|
|
|
assert service.list_account_valuations(loan, db_path=path)["count"] == 0
|
|
|
|
|
|
def test_dated_loan_create_and_historical_update_preserve_current_cache_without_valuations(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
created = service.create_ledger_account(
|
|
"대출", type="loan", last_balance=900, last_synced_at="2026-06-30",
|
|
last_balance_basis_date="2026-06-30", db_path=path,
|
|
)
|
|
loan = created["account"]["id"]
|
|
assert created["account"]["last_balance"] == 900
|
|
assert created["account"]["last_synced_at"] == "2026-06-30"
|
|
assert service.list_account_valuations(loan, db_path=path)["count"] == 0
|
|
|
|
updated = service.update_ledger_account(
|
|
loan, 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
|
|
assert updated["account"]["last_synced_at"] == "2026-06-30"
|
|
assert service.list_account_valuations(loan, db_path=path)["count"] == 0
|
|
|
|
|
|
def test_loan_cash_balance_and_total_value_are_rejected_without_writes(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
loan = service.create_ledger_account("대출", type="loan", db_path=path)["account"]["id"]
|
|
|
|
for valuation_kind in ("cash_balance", "total_value"):
|
|
rejected = service.register_account_valuation(
|
|
loan, "2026-06-30", 100, valuation_kind=valuation_kind, 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
|
|
|
|
|
|
def test_non_loan_candidate_still_succeeds_and_reuses_idempotently(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", "candidate", "memo",
|
|
)
|
|
replay = record_account_valuation_candidate(
|
|
conn, account, "2026-06-30", 100, "KRW", "import", "source", "candidate", "memo",
|
|
)
|
|
|
|
assert first["success"] is True
|
|
assert first["reused"] is False
|
|
assert replay["success"] is True
|
|
assert replay["reused"] is True
|
|
assert replay["valuation"]["id"] == first["valuation"]["id"]
|
|
assert service.list_account_valuations(account, db_path=path)["count"] == 1
|
|
|
|
|
|
def test_candidate_invalid_amount_returns_controlled_error_with_existing_row(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", "candidate", "memo",
|
|
)
|
|
rejected = record_account_valuation_candidate(
|
|
conn, account, "2026-06-30", "not-an-integer", "KRW", "import", "source", "candidate", "memo",
|
|
)
|
|
|
|
assert first["success"] is True
|
|
assert rejected["success"] is False
|
|
assert rejected["error_code"] == "validation_error"
|
|
assert service.list_account_valuations(account, db_path=path)["count"] == 1
|