232 lines
9.0 KiB
Python
232 lines
9.0 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_bank_to_loan_conversion_rejects_existing_undefined_valuation_without_mutation(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
created = service.create_ledger_account(
|
|
"은행", type="bank", last_balance=100, last_synced_at="2026-06-30",
|
|
last_balance_basis_date="2026-06-30", db_path=path,
|
|
)
|
|
account_id = created["account"]["id"]
|
|
before = service.list_account_valuations(account_id, db_path=path)
|
|
assert before["count"] == 1
|
|
assert before["valuations"][0]["representative_liability_definition"] is None
|
|
|
|
rejected = service.update_ledger_account(account_id, type="loan", db_path=path)
|
|
|
|
assert rejected["success"] is False
|
|
assert rejected["error_code"] == "validation_error"
|
|
accounts = service.list_ledger_accounts(active_only=False, db_path=path)["accounts"]
|
|
assert next(account for account in accounts if account["id"] == account_id)["type"] == "bank"
|
|
after = service.list_account_valuations(account_id, db_path=path)
|
|
assert after["count"] == 1
|
|
assert after["valuations"] == before["valuations"]
|
|
|
|
|
|
def test_bank_to_loan_conversion_without_valuations_remains_allowed(tmp_path):
|
|
path = str(tmp_path / "ledger.sqlite3")
|
|
account_id = service.create_ledger_account("은행", type="bank", db_path=path)["account"]["id"]
|
|
|
|
converted = service.update_ledger_account(account_id, type="loan", db_path=path)
|
|
|
|
assert converted["success"] is True
|
|
assert converted["account"]["type"] == "loan"
|
|
assert service.list_account_valuations(account_id, db_path=path)["count"] == 0
|
|
|
|
|
|
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
|