78 lines
3.3 KiB
Python
78 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
import os
|
|
|
|
import pytest
|
|
|
|
from jarca_ledger_mcp import service
|
|
|
|
PDF_PATH = Path("/root/.hermes/cache/documents/doc_5a12af727ae0_KB거래내역조회_5704_2606171043.pdf")
|
|
PDF_PASSWORD = os.environ.get("KB_BANK_PDF_TEST_PASSWORD", "")
|
|
|
|
|
|
def _sample_available(path: Path, password: str) -> bool:
|
|
if not password:
|
|
return False
|
|
try:
|
|
return path.is_file()
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
SAMPLE_AVAILABLE = _sample_available(PDF_PATH, PDF_PASSWORD)
|
|
|
|
|
|
@pytest.mark.skipif(not SAMPLE_AVAILABLE, reason="sample KB Bank PDF/password is not available")
|
|
def test_preview_kb_bank_pdf_import_real_sample(tmp_path):
|
|
result = service.preview_kb_bank_pdf_import(str(PDF_PATH), PDF_PASSWORD, preview_limit=3, db_path=str(tmp_path / "ledger.sqlite3"))
|
|
assert result["success"] is True
|
|
assert result["row_count"] == 499
|
|
assert result["page_count"] == 20
|
|
assert result["metadata"]["account_number_masked"] == "****5704"
|
|
assert result["metadata"]["final_balance"] == 3_041_678
|
|
assert result["validation"]["ok"] is True
|
|
assert result["validation"]["total_deposit"] == 125_730_389
|
|
assert result["validation"]["total_withdrawal"] == 123_565_861
|
|
assert result["preview"][0]["transaction_datetime"] == "2026-01-03T09:05:10"
|
|
assert result["preview"][0]["type"] == "expense"
|
|
assert result["preview"][0]["amount"] == 200
|
|
|
|
|
|
@pytest.mark.skipif(not SAMPLE_AVAILABLE, reason="sample KB Bank PDF/password is not available")
|
|
def test_import_kb_bank_pdf_real_sample_and_rollback(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
result = service.import_kb_bank_pdf(str(PDF_PATH), PDF_PASSWORD, db_path=db_path)
|
|
assert result["success"] is True
|
|
assert result["parsed_row_count"] == 499
|
|
assert result["inserted_count"] == 499
|
|
assert result["skipped_duplicate_count"] == 0
|
|
assert result["metadata"]["account_number_masked"] == "****5704"
|
|
|
|
duplicated = service.import_kb_bank_pdf(str(PDF_PATH), PDF_PASSWORD, db_path=db_path)
|
|
assert duplicated["success"] is True
|
|
assert duplicated["inserted_count"] == 0
|
|
assert duplicated["skipped_duplicate_count"] == 499
|
|
|
|
listed = service.list_ledger_transactions(start_date="2026-01-01", end_date="2026-06-17", limit=1000, db_path=db_path)
|
|
assert listed["success"] is True
|
|
assert listed["count"] == 499
|
|
sample = next(tx for tx in listed["transactions"] if tx["transaction_datetime"] == "2026-06-15T16:50:50")
|
|
assert sample["merchant"] == "오아시스마켓"
|
|
assert sample["balance_after"] == 3_041_678
|
|
assert sample["branch"] == "KB카드"
|
|
|
|
accounts = service.list_ledger_accounts(db_path=db_path)
|
|
assert accounts["count"] == 1
|
|
assert accounts["accounts"][0]["institution"] == "KB국민은행"
|
|
assert accounts["accounts"][0]["last_balance"] == 3_041_678
|
|
|
|
batches = service.list_ledger_import_batches(source="kb_bank_pdf", db_path=db_path)
|
|
assert batches["count"] == 2
|
|
|
|
rollback = service.rollback_ledger_import_batch(result["import_batch_id"], db_path=db_path)
|
|
assert rollback["success"] is True
|
|
assert rollback["affected_transactions"] == 499
|
|
after = service.list_ledger_transactions(start_date="2026-01-01", end_date="2026-06-17", limit=1000, db_path=db_path)
|
|
assert after["count"] == 0
|