128 lines
4.5 KiB
Python
128 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
from jarca_ledger_mcp import service
|
|
|
|
|
|
def _find_tx(rows: list[dict], tx_id: str) -> dict:
|
|
for row in rows:
|
|
if row["id"] == tx_id:
|
|
return row
|
|
raise AssertionError(f"tx {tx_id} not found")
|
|
|
|
|
|
def test_run_cross_transfer_match_marks_ownership_transfer_as_investment(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
kb_account = service.create_ledger_account("KB 본인", "bank", institution="국민", holder_name="고진환", db_path=db_path)
|
|
toss_account = service.create_ledger_account("토스증권/뱅크", "bank", institution="토스", holder_name="고진환", db_path=db_path)
|
|
|
|
kb_tx = service.create_ledger_transaction(
|
|
transaction_date="2026-06-17",
|
|
type="expense",
|
|
amount=63_000,
|
|
category="이체",
|
|
subcategory="입금",
|
|
merchant="출금",
|
|
counterparty="고진환",
|
|
account_id=kb_account["account"]["id"],
|
|
source="kb_bank_pdf",
|
|
db_path=db_path,
|
|
)
|
|
toss_tx = service.create_ledger_transaction(
|
|
transaction_date="2026-06-17",
|
|
type="income",
|
|
amount=63_000,
|
|
category="이체",
|
|
subcategory="입금",
|
|
merchant="입금",
|
|
counterparty="고진환",
|
|
account_id=toss_account["account"]["id"],
|
|
source="toss_bank_pdf",
|
|
db_path=db_path,
|
|
)
|
|
|
|
assert kb_tx["success"] is True
|
|
assert toss_tx["success"] is True
|
|
|
|
result = service.run_cross_account_transfer_match(db_path=db_path)
|
|
assert result["success"] is True
|
|
assert result["matched_count"] >= 1
|
|
transactions = service.list_ledger_transactions(limit=20, db_path=db_path)["transactions"]
|
|
kb = _find_tx(transactions, kb_tx["transaction"]["id"])
|
|
toss = _find_tx(transactions, toss_tx["transaction"]["id"])
|
|
|
|
assert kb["category"] == "이체"
|
|
assert kb["subcategory"] == "투자이동"
|
|
assert toss["category"] == "이체"
|
|
assert toss["subcategory"] == "투자이동"
|
|
|
|
|
|
def test_run_cross_transfer_match_marks_foreign_name_as_review_needed(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
kb_account = service.create_ledger_account("KB 본인", "bank", institution="국민", holder_name="고진환", db_path=db_path)
|
|
toss_account = service.create_ledger_account("토스뱅크", "bank", institution="토스", holder_name="고진환", db_path=db_path)
|
|
|
|
kb_tx = service.create_ledger_transaction(
|
|
transaction_date="2026-06-20",
|
|
type="expense",
|
|
amount=5_000_000,
|
|
category="이체",
|
|
subcategory="출금",
|
|
merchant="출금",
|
|
counterparty="최진선",
|
|
account_id=kb_account["account"]["id"],
|
|
source="kb_bank_pdf",
|
|
db_path=db_path,
|
|
)
|
|
toss_tx = service.create_ledger_transaction(
|
|
transaction_date="2026-06-20",
|
|
type="income",
|
|
amount=5_000_000,
|
|
category="이체",
|
|
subcategory="입금",
|
|
merchant="입금",
|
|
counterparty="최진선",
|
|
account_id=toss_account["account"]["id"],
|
|
source="toss_bank_pdf",
|
|
db_path=db_path,
|
|
)
|
|
|
|
assert kb_tx["success"] is True
|
|
assert toss_tx["success"] is True
|
|
|
|
result = service.run_cross_account_transfer_match(db_path=db_path)
|
|
assert result["success"] is True
|
|
|
|
transactions = service.list_ledger_transactions(limit=20, db_path=db_path)["transactions"]
|
|
kb = _find_tx(transactions, kb_tx["transaction"]["id"])
|
|
toss = _find_tx(transactions, toss_tx["transaction"]["id"])
|
|
assert kb["subcategory"] == "확인필요"
|
|
assert toss["subcategory"] == "확인필요"
|
|
|
|
|
|
def test_preview_cross_transfer_marks_unmatched_as_review_item(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
kb_account = service.create_ledger_account("KB 본인", "bank", institution="국민", holder_name="고진환", db_path=db_path)
|
|
|
|
kb_tx = service.create_ledger_transaction(
|
|
transaction_date="2026-06-22",
|
|
type="expense",
|
|
amount=12_000,
|
|
category="이체",
|
|
subcategory="출금",
|
|
merchant="출금",
|
|
counterparty="고진환",
|
|
account_id=kb_account["account"]["id"],
|
|
source="kb_bank_pdf",
|
|
db_path=db_path,
|
|
)
|
|
|
|
assert kb_tx["success"] is True
|
|
|
|
result = service.run_cross_account_transfer_match(limit=100, db_path=db_path)
|
|
assert result["success"] is True
|
|
assert result["review_count"] >= 1
|
|
|
|
rows = service.list_cross_transfer_review_queue(db_path=db_path)["transactions"]
|
|
row = _find_tx(rows, kb_tx["transaction"]["id"])
|
|
assert row["memo"].find("cross_transfer_review") != -1
|