544 lines
21 KiB
Python
544 lines
21 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import json
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.infrastructure.sqlite import init_db
|
|
from jarca_ledger_mcp.provenance_policy import (
|
|
EXPOSURE_LEVEL_DEBUG_GATED,
|
|
EXPOSURE_LEVEL_MINIMAL,
|
|
EXPOSURE_LEVEL_OPERATOR_SUMMARY,
|
|
sanitize_provenance_for_exposure,
|
|
sanitize_provenance,
|
|
sanitize_provenance_envelope,
|
|
)
|
|
from jarca_ledger_mcp.kb_bank_pdf import KbPdfTransaction
|
|
from jarca_ledger_mcp.seoulpay_xlsx import SeoulPayTransaction
|
|
from jarca_ledger_mcp.toss_bank_pdf import TossBankTransaction
|
|
from jarca_ledger_mcp.toss_securities_pdf import TossSecuritiesTransaction
|
|
|
|
|
|
SECRET_SOURCE_PATH = "/very/secret/home/alice/Documents/private/statement.pdf"
|
|
SOURCE_HASH = "abcdef1234567890fedcba0987654321abcdef1234567890fedcba0987654321"
|
|
|
|
|
|
def _expected_filename(file_path: str, source_hash: str = SOURCE_HASH) -> str:
|
|
return f"{Path(file_path).name}#{source_hash[:12]}"
|
|
|
|
|
|
def _stored_batch_and_payloads(db_path: str) -> tuple[str, list[dict[str, object]]]:
|
|
with sqlite3.connect(db_path) as conn:
|
|
batch_filename = conn.execute("SELECT filename FROM import_batches ORDER BY imported_at DESC LIMIT 1").fetchone()[0]
|
|
payload_rows = conn.execute("SELECT raw_payload FROM transactions ORDER BY created_at").fetchall()
|
|
return batch_filename, [json.loads(row[0]) for row in payload_rows]
|
|
|
|
|
|
def test_preview_ledger_csv_import_redacts_full_path_and_raw_columns(tmp_path):
|
|
csv_path = tmp_path / "import-preview.csv"
|
|
with csv_path.open("w", encoding="utf-8", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=["date", "amount", "merchant", "secret_note", "local_path"])
|
|
writer.writeheader()
|
|
writer.writerow(
|
|
{
|
|
"date": "2026-06-18",
|
|
"amount": "42000",
|
|
"merchant": "테스트상점",
|
|
"secret_note": "do not leak this raw note",
|
|
"local_path": "/Users/alice/Desktop/private.csv",
|
|
}
|
|
)
|
|
|
|
result = service.preview_ledger_csv_import(str(csv_path))
|
|
|
|
assert result["success"] is True
|
|
assert result["filename"] == _expected_filename(str(csv_path), result["source_hash"])
|
|
assert result["preview"] == [{"row_number": 1, "date": "2026-06-18", "amount": "42000", "merchant": "테스트상점"}]
|
|
rendered = json.dumps(result, ensure_ascii=False)
|
|
assert str(tmp_path) not in rendered
|
|
assert "secret_note" not in rendered
|
|
assert "local_path" not in rendered
|
|
|
|
|
|
def test_preview_and_import_kb_bank_pdf_sanitize_path_and_payload(monkeypatch, tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
init_db(db_path)
|
|
parsed = {
|
|
"file_path": SECRET_SOURCE_PATH,
|
|
"source_hash": SOURCE_HASH,
|
|
"page_count": 4,
|
|
"metadata": {"account_name": "생활비", "account_number_masked": "****5704", "holder_name": "홍길동", "final_balance": 123456},
|
|
"transactions": [
|
|
KbPdfTransaction(
|
|
page=7,
|
|
transaction_datetime="2026-06-18T09:00:00",
|
|
transaction_date="2026-06-18",
|
|
raw_summary="적요-민감메모",
|
|
counterparty="테스트상대",
|
|
withdrawal=42000,
|
|
deposit=0,
|
|
balance_after=123456,
|
|
remittance_memo="숨기고싶은메모",
|
|
branch="강남지점",
|
|
)
|
|
],
|
|
"validation": {"ok": True, "errors": []},
|
|
}
|
|
monkeypatch.setattr(service, "parse_kb_bank_pdf", lambda file_path, password: parsed)
|
|
|
|
preview = service.preview_kb_bank_pdf_import("ignored.pdf", "pw", db_path=db_path)
|
|
imported = service.import_kb_bank_pdf("ignored.pdf", "pw", db_path=db_path)
|
|
|
|
assert preview["success"] is True
|
|
assert preview["file_path"] == _expected_filename(SECRET_SOURCE_PATH)
|
|
assert SECRET_SOURCE_PATH not in json.dumps(preview, ensure_ascii=False)
|
|
assert imported["success"] is True
|
|
|
|
batch_filename, payloads = _stored_batch_and_payloads(db_path)
|
|
assert batch_filename == _expected_filename(SECRET_SOURCE_PATH)
|
|
assert payloads == [{"page": 7}]
|
|
|
|
|
|
def test_preview_and_import_seoulpay_xlsx_sanitize_path_and_payload(monkeypatch, tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
init_db(db_path)
|
|
parsed = {
|
|
"file_path": "/very/secret/home/alice/Documents/private/seoulpay.xlsx",
|
|
"source_hash": SOURCE_HASH,
|
|
"sheet_name": "결제내역",
|
|
"metadata": {"holder_name": "홍길동", "period": "20260601~20260620"},
|
|
"transactions": [
|
|
SeoulPayTransaction(
|
|
row_index=11,
|
|
channel="서울페이",
|
|
transaction_no="TRX-001",
|
|
transaction_datetime="2026-06-18 18:27:46",
|
|
transaction_date="2026-06-18",
|
|
merchant="땡겨요",
|
|
business_no="2028102637",
|
|
merchant_id="AF2402815444",
|
|
transaction_kind="온라인 결제",
|
|
payment_kind="결제",
|
|
amount=42000,
|
|
supply_amount=38182,
|
|
service_fee=0,
|
|
vat=3818,
|
|
tax_status="과세",
|
|
)
|
|
],
|
|
"validation": {"ok": True, "parsed_rows": 1, "errors": []},
|
|
}
|
|
monkeypatch.setattr(service, "parse_seoulpay_xlsx", lambda file_path, password="": parsed)
|
|
|
|
preview = service.preview_seoulpay_xlsx_import("ignored.xlsx", db_path=db_path)
|
|
imported = service.import_seoulpay_xlsx("ignored.xlsx", db_path=db_path)
|
|
|
|
assert preview["success"] is True
|
|
assert preview["file_path"] == _expected_filename(parsed["file_path"])
|
|
assert parsed["file_path"] not in json.dumps(preview, ensure_ascii=False)
|
|
assert imported["success"] is True
|
|
|
|
batch_filename, payloads = _stored_batch_and_payloads(db_path)
|
|
assert batch_filename == _expected_filename(parsed["file_path"])
|
|
assert payloads == [{"row_index": 11}]
|
|
|
|
|
|
def test_preview_and_import_toss_bank_pdf_sanitize_path_and_payload(monkeypatch, tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
init_db(db_path)
|
|
parsed = {
|
|
"file_path": SECRET_SOURCE_PATH,
|
|
"source_hash": SOURCE_HASH,
|
|
"page_count": 3,
|
|
"metadata": {"holder_name": "홍길동", "account_name": "토스 통장", "account_number_masked": "100-1234-5678", "final_balance": 820000},
|
|
"transactions": [
|
|
TossBankTransaction(
|
|
row_index=9,
|
|
transaction_datetime="2026-06-18 09:00:00",
|
|
transaction_date="2026-06-18",
|
|
action="이체",
|
|
counterparty="테스트상대",
|
|
amount=12000,
|
|
type="expense",
|
|
balance_after=820000,
|
|
raw_summary="상대방 요약",
|
|
raw_lines=[SECRET_SOURCE_PATH, "ssh private key"],
|
|
)
|
|
],
|
|
"validation": {"ok": True, "parsed_rows": 1, "errors": []},
|
|
}
|
|
monkeypatch.setattr(service, "parse_toss_bank_pdf", lambda file_path: parsed)
|
|
|
|
preview = service.preview_toss_bank_pdf_import("ignored.pdf", db_path=db_path)
|
|
imported = service.import_toss_bank_pdf("ignored.pdf", db_path=db_path)
|
|
|
|
assert preview["success"] is True
|
|
assert preview["file_path"] == _expected_filename(SECRET_SOURCE_PATH)
|
|
rendered = json.dumps(preview, ensure_ascii=False)
|
|
assert SECRET_SOURCE_PATH not in rendered
|
|
assert "raw_lines" not in rendered
|
|
assert imported["success"] is True
|
|
|
|
batch_filename, payloads = _stored_batch_and_payloads(db_path)
|
|
assert batch_filename == _expected_filename(SECRET_SOURCE_PATH)
|
|
assert payloads == [{"row_index": 9}]
|
|
|
|
|
|
def test_preview_and_import_toss_securities_pdf_sanitize_path_and_payload(monkeypatch, tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
init_db(db_path)
|
|
parsed = {
|
|
"file_path": SECRET_SOURCE_PATH,
|
|
"source_hash": SOURCE_HASH,
|
|
"page_count": 5,
|
|
"metadata": {"holder_name": "홍길동", "account_name": "토스증권 계좌", "account_number_masked": "123-01-048829"},
|
|
"transactions": [
|
|
TossSecuritiesTransaction(
|
|
row_index=5,
|
|
section="USD",
|
|
transaction_date="2026-06-18",
|
|
action="외화증권배당금입금",
|
|
name="ACME(A123)",
|
|
code="A123",
|
|
amount=4300,
|
|
type="income",
|
|
quantity=1.5,
|
|
unit_price=25.2,
|
|
fee=120,
|
|
tax=33,
|
|
balance_after=910000,
|
|
raw_numbers=["1.5", "25.2", "4300"],
|
|
raw_block=[SECRET_SOURCE_PATH, "api_token=secret"],
|
|
)
|
|
],
|
|
"validation": {"ok": True, "parsed_rows": 1, "errors": []},
|
|
}
|
|
monkeypatch.setattr(service, "parse_toss_securities_pdf", lambda file_path: parsed)
|
|
|
|
preview = service.preview_toss_securities_pdf_import("ignored.pdf", db_path=db_path)
|
|
imported = service.import_toss_securities_pdf("ignored.pdf", db_path=db_path)
|
|
|
|
assert preview["success"] is True
|
|
assert preview["file_path"] == _expected_filename(SECRET_SOURCE_PATH)
|
|
rendered = json.dumps(preview, ensure_ascii=False)
|
|
assert SECRET_SOURCE_PATH not in rendered
|
|
assert "raw_block" not in rendered
|
|
assert imported["success"] is True
|
|
|
|
batch_filename, payloads = _stored_batch_and_payloads(db_path)
|
|
assert batch_filename == _expected_filename(SECRET_SOURCE_PATH)
|
|
assert payloads == [{"row_index": 5, "section": "USD", "quantity": 1.5, "unit_price": 25.2, "fee": 120, "tax": 33}]
|
|
|
|
|
|
def test_sanitize_provenance_by_exposure_levels_filters_sensitive_values():
|
|
envelope = {
|
|
"source": "kb_bank_pdf",
|
|
"source_hash": SOURCE_HASH,
|
|
"file_path": SECRET_SOURCE_PATH,
|
|
"filename": SECRET_SOURCE_PATH,
|
|
"parser": "kb",
|
|
"import_batch_id": "batch-1",
|
|
"external_id": "ext-1",
|
|
"dedupe_key": "dedupe-1",
|
|
"row_number": 7,
|
|
"page": 3,
|
|
"sheet": "A1",
|
|
"validation": {"ok": True, "errors": []},
|
|
"raw_payload": {
|
|
"row_index": 7,
|
|
"page": 3,
|
|
"action": "transfer",
|
|
"raw_text": "raw statement text",
|
|
"password": "p@ss",
|
|
"counterparty": "김대리",
|
|
"secret": "should-hide",
|
|
"transaction_path": "/very/sensitive/path.csv",
|
|
"raw_document": "full text dump",
|
|
},
|
|
"debug": {
|
|
"parser_diag": "ok",
|
|
"raw_text": "should-be-hidden",
|
|
"raw_payload_dump": "hide-this",
|
|
"token": "token-001",
|
|
"message": "keep? no",
|
|
"notes": ["safe", "path=/tmp/should-not-leak", {"secret": "still-hidden", "message": "ok"}],
|
|
},
|
|
"local_path": "/another/secret/path",
|
|
}
|
|
|
|
minimal = sanitize_provenance(envelope, exposure_level=EXPOSURE_LEVEL_MINIMAL)
|
|
assert minimal["source"] == "kb_bank_pdf"
|
|
assert minimal["parser"] == "kb"
|
|
assert minimal["source_hash"] == SOURCE_HASH
|
|
assert minimal["source_hash_prefix"] == SOURCE_HASH[:12]
|
|
assert "file_path" not in minimal
|
|
assert "filename" not in minimal
|
|
assert "raw_payload" not in minimal
|
|
assert "debug" not in minimal
|
|
rendered = json.dumps(minimal, ensure_ascii=False)
|
|
assert SECRET_SOURCE_PATH not in rendered
|
|
assert "/tmp/should-not-leak" not in rendered
|
|
assert "password" not in rendered
|
|
assert "token" not in rendered
|
|
|
|
operator_summary = sanitize_provenance(
|
|
envelope,
|
|
exposure_level=EXPOSURE_LEVEL_OPERATOR_SUMMARY,
|
|
)
|
|
assert operator_summary["source"] == "kb_bank_pdf"
|
|
assert operator_summary["parser"] == "kb"
|
|
assert operator_summary["filename"] == _expected_filename(SECRET_SOURCE_PATH)
|
|
assert operator_summary["source_hash_prefix"] == SOURCE_HASH[:12]
|
|
assert operator_summary["validation"] == {"ok": True, "errors": []}
|
|
assert "raw_payload" not in operator_summary
|
|
assert operator_summary["filename"] == _expected_filename(SECRET_SOURCE_PATH)
|
|
assert "raw_text" not in json.dumps(operator_summary, ensure_ascii=False)
|
|
|
|
debug = sanitize_provenance(
|
|
envelope,
|
|
exposure_level=EXPOSURE_LEVEL_DEBUG_GATED,
|
|
)
|
|
assert debug["filename"] == _expected_filename(SECRET_SOURCE_PATH)
|
|
assert debug["source_hash_prefix"] == SOURCE_HASH[:12]
|
|
assert "raw_payload" in debug
|
|
debug_payload = debug["raw_payload"]
|
|
assert debug_payload == {
|
|
"row_index": 7,
|
|
"page": 3,
|
|
"action": "transfer",
|
|
"counterparty": "김대리",
|
|
}
|
|
assert "password" not in json.dumps(debug, ensure_ascii=False)
|
|
assert "raw_text" not in json.dumps(debug, ensure_ascii=False)
|
|
assert "token" not in json.dumps(debug, ensure_ascii=False)
|
|
assert "raw_document" not in json.dumps(debug, ensure_ascii=False)
|
|
|
|
|
|
def test_sanitize_provenance_envelope_alias_matches_core_sanitizer():
|
|
envelope = {
|
|
"source": "toss_securities_pdf",
|
|
"source_hash": SOURCE_HASH,
|
|
"file_path": SECRET_SOURCE_PATH,
|
|
"raw_payload": '{"row_index":1,"page":2,"secret":"x"}',
|
|
}
|
|
|
|
assert sanitize_provenance_envelope(envelope, EXPOSURE_LEVEL_MINIMAL) == sanitize_provenance(
|
|
envelope,
|
|
EXPOSURE_LEVEL_MINIMAL,
|
|
)
|
|
assert sanitize_provenance_envelope(envelope, EXPOSURE_LEVEL_OPERATOR_SUMMARY) == sanitize_provenance(
|
|
envelope,
|
|
EXPOSURE_LEVEL_OPERATOR_SUMMARY,
|
|
)
|
|
assert sanitize_provenance_envelope(envelope, EXPOSURE_LEVEL_DEBUG_GATED) == sanitize_provenance(
|
|
envelope,
|
|
EXPOSURE_LEVEL_DEBUG_GATED,
|
|
)
|
|
|
|
assert sanitize_provenance_for_exposure(envelope, EXPOSURE_LEVEL_OPERATOR_SUMMARY) == sanitize_provenance(
|
|
envelope,
|
|
EXPOSURE_LEVEL_OPERATOR_SUMMARY,
|
|
)
|
|
|
|
|
|
def test_list_ledger_import_batches_defaults_to_minimal_provenance(monkeypatch, tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
parsed = {
|
|
"file_path": SECRET_SOURCE_PATH,
|
|
"source_hash": SOURCE_HASH,
|
|
"page_count": 4,
|
|
"metadata": {"account_name": "생활비", "account_number_masked": "****5704", "holder_name": "홍길동", "final_balance": 123456},
|
|
"transactions": [
|
|
KbPdfTransaction(
|
|
page=7,
|
|
transaction_datetime="2026-06-18T09:00:00",
|
|
transaction_date="2026-06-18",
|
|
raw_summary="적요-민감메모",
|
|
counterparty="테스트상대",
|
|
withdrawal=42000,
|
|
deposit=0,
|
|
balance_after=123456,
|
|
remittance_memo="숨기고싶은메모",
|
|
branch="강남지점",
|
|
)
|
|
],
|
|
"validation": {"ok": True, "errors": []},
|
|
}
|
|
monkeypatch.setattr(service, "parse_kb_bank_pdf", lambda file_path, password: parsed)
|
|
|
|
imported = service.import_kb_bank_pdf("ignored.pdf", "pw", db_path=db_path)
|
|
assert imported["success"] is True
|
|
|
|
listed = service.list_ledger_import_batches(source="kb_bank_pdf", db_path=db_path)
|
|
|
|
assert listed["success"] is True
|
|
assert listed["count"] == 1
|
|
batch = listed["import_batches"][0]
|
|
assert batch["memo"] == ""
|
|
assert batch["filename"] == _expected_filename(SECRET_SOURCE_PATH)
|
|
assert batch["provenance"] == {
|
|
"source": "kb_bank_pdf",
|
|
"parser": "kb_bank_pdf",
|
|
"import_batch_id": imported["import_batch_id"],
|
|
"source_hash": SOURCE_HASH,
|
|
"source_hash_prefix": SOURCE_HASH[:12],
|
|
}
|
|
rendered = json.dumps(batch, ensure_ascii=False)
|
|
assert SECRET_SOURCE_PATH not in rendered
|
|
assert "account_number_masked" not in rendered
|
|
assert "validation" not in rendered
|
|
|
|
|
|
def test_list_ledger_import_batches_operator_summary_exposes_masked_filename_and_validation(monkeypatch, tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
parsed = {
|
|
"file_path": "/very/secret/home/alice/Documents/private/seoulpay.xlsx",
|
|
"source_hash": SOURCE_HASH,
|
|
"sheet_name": "결제내역",
|
|
"metadata": {"holder_name": "홍길동", "period": "20260601~20260620"},
|
|
"transactions": [
|
|
SeoulPayTransaction(
|
|
row_index=11,
|
|
channel="서울페이",
|
|
transaction_no="TRX-001",
|
|
transaction_datetime="2026-06-18 18:27:46",
|
|
transaction_date="2026-06-18",
|
|
merchant="땡겨요",
|
|
business_no="2028102637",
|
|
merchant_id="AF2402815444",
|
|
transaction_kind="온라인 결제",
|
|
payment_kind="결제",
|
|
amount=42000,
|
|
supply_amount=38182,
|
|
service_fee=0,
|
|
vat=3818,
|
|
tax_status="과세",
|
|
)
|
|
],
|
|
"validation": {"ok": True, "parsed_rows": 1, "errors": []},
|
|
}
|
|
monkeypatch.setattr(service, "parse_seoulpay_xlsx", lambda file_path, password="": parsed)
|
|
|
|
imported = service.import_seoulpay_xlsx("ignored.xlsx", db_path=db_path)
|
|
assert imported["success"] is True
|
|
|
|
listed = service.list_ledger_import_batches(
|
|
source="seoulpay_xlsx",
|
|
provenance_exposure=EXPOSURE_LEVEL_OPERATOR_SUMMARY,
|
|
db_path=db_path,
|
|
)
|
|
|
|
assert listed["success"] is True
|
|
batch = listed["import_batches"][0]
|
|
assert batch["memo"] == ""
|
|
assert batch["provenance"] == {
|
|
"source": "seoulpay_xlsx",
|
|
"parser": "seoulpay_xlsx",
|
|
"filename": _expected_filename(parsed["file_path"]),
|
|
"import_batch_id": imported["import_batch_id"],
|
|
"sheet_name": "결제내역",
|
|
"source_hash": SOURCE_HASH,
|
|
"source_hash_prefix": SOURCE_HASH[:12],
|
|
"validation": {"ok": True, "parsed_rows": 1, "errors": []},
|
|
}
|
|
rendered = json.dumps(batch, ensure_ascii=False)
|
|
assert parsed["file_path"] not in rendered
|
|
assert "holder_name" not in rendered
|
|
assert "period" not in rendered
|
|
|
|
|
|
def test_list_ledger_import_batches_requires_explicit_gate_for_debug(monkeypatch, tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
parsed = {
|
|
"file_path": SECRET_SOURCE_PATH,
|
|
"source_hash": SOURCE_HASH,
|
|
"page_count": 1,
|
|
"metadata": {"account_name": "생활비", "account_number_masked": "****5704", "holder_name": "홍길동", "final_balance": 123456},
|
|
"validation": {"ok": True, "errors": []},
|
|
"transactions": [
|
|
KbPdfTransaction(
|
|
page=7,
|
|
transaction_datetime="2026-06-18T09:00:00",
|
|
transaction_date="2026-06-18",
|
|
raw_summary="적요-민감메모",
|
|
counterparty="테스트상대",
|
|
withdrawal=42000,
|
|
deposit=0,
|
|
balance_after=123456,
|
|
remittance_memo="숨기고싶은메모",
|
|
branch="강남지점",
|
|
)
|
|
],
|
|
}
|
|
monkeypatch.setattr(service, "parse_kb_bank_pdf", lambda file_path, password: parsed)
|
|
|
|
imported = service.import_kb_bank_pdf("ignored.pdf", "pw", db_path=db_path)
|
|
assert imported["success"] is True
|
|
|
|
denied = service.list_ledger_import_batches(
|
|
source="kb_bank_pdf",
|
|
provenance_exposure=EXPOSURE_LEVEL_DEBUG_GATED,
|
|
db_path=db_path,
|
|
)
|
|
assert denied == {
|
|
"success": False,
|
|
"error_code": "forbidden",
|
|
"error": "debug-gated provenance requires explicit debug_provenance_gate=True",
|
|
}
|
|
|
|
allowed = service.list_ledger_import_batches(
|
|
source="kb_bank_pdf",
|
|
provenance_exposure=EXPOSURE_LEVEL_DEBUG_GATED,
|
|
debug_provenance_gate=True,
|
|
db_path=db_path,
|
|
)
|
|
assert allowed["success"] is True
|
|
assert allowed["import_batches"][0]["provenance"]["filename"] == _expected_filename(SECRET_SOURCE_PATH)
|
|
assert "debug" not in allowed["import_batches"][0]["provenance"]
|
|
|
|
|
|
def test_mcp_list_ledger_import_batches_passes_provenance_exposure(monkeypatch):
|
|
from jarca_ledger_mcp.adapters.mcp import server
|
|
|
|
captured = {}
|
|
|
|
def fake_list_ledger_import_batches(source, include_rolled_back, limit, provenance_exposure="minimal", debug_provenance_gate=False):
|
|
captured.update(
|
|
source=source,
|
|
include_rolled_back=include_rolled_back,
|
|
limit=limit,
|
|
provenance_exposure=provenance_exposure,
|
|
debug_provenance_gate=debug_provenance_gate,
|
|
)
|
|
return {"success": True, "import_batches": [], "count": 0}
|
|
|
|
monkeypatch.setattr(server.contracts, "list_ledger_import_batches", fake_list_ledger_import_batches)
|
|
|
|
result = server.list_ledger_import_batches(source="kb_bank_pdf", include_rolled_back=False, limit=5)
|
|
assert result["success"] is True
|
|
assert captured == {
|
|
"source": "kb_bank_pdf",
|
|
"include_rolled_back": False,
|
|
"limit": 5,
|
|
"provenance_exposure": EXPOSURE_LEVEL_MINIMAL,
|
|
"debug_provenance_gate": False,
|
|
}
|
|
|
|
result = server.list_ledger_import_batches(
|
|
source="seoulpay_xlsx",
|
|
include_rolled_back=True,
|
|
limit=3,
|
|
provenance_exposure=EXPOSURE_LEVEL_OPERATOR_SUMMARY,
|
|
debug_provenance_gate=True,
|
|
)
|
|
assert result["success"] is True
|
|
assert captured == {
|
|
"source": "seoulpay_xlsx",
|
|
"include_rolled_back": True,
|
|
"limit": 3,
|
|
"provenance_exposure": EXPOSURE_LEVEL_OPERATOR_SUMMARY,
|
|
"debug_provenance_gate": True,
|
|
}
|