From 26b815d723f501d4d84e3856c64b875226c66b8c Mon Sep 17 00:00:00 2001 From: Jarca Date: Tue, 30 Jun 2026 16:35:51 +0900 Subject: [PATCH] fix(ci): restore db safety smoke backup and sample-test skip behavior --- scripts/ci_db_safety_smoke.py | 13 ++++++++++++- tests/test_kb_bank_pdf.py | 13 ++++++++++++- tests/test_toss_securities_pdf.py | 11 +++++++++-- 3 files changed, 33 insertions(+), 4 deletions(-) diff --git a/scripts/ci_db_safety_smoke.py b/scripts/ci_db_safety_smoke.py index d387173..a0efb4e 100755 --- a/scripts/ci_db_safety_smoke.py +++ b/scripts/ci_db_safety_smoke.py @@ -9,6 +9,7 @@ from __future__ import annotations import argparse import json import os +import sqlite3 import subprocess import sys import tempfile @@ -35,6 +36,16 @@ def parse_json_output(output: str) -> dict[str, Any]: return {"success": False, "error_code": "invalid_json_output"} +def copy_sqlite_backup(source: Path, backup: Path) -> None: + """Create a standalone SQLite backup, including data still held in WAL.""" + backup.parent.mkdir(parents=True, exist_ok=True) + if backup.exists(): + backup.unlink() + with sqlite3.connect(f"file:{source}?mode=ro", uri=True) as source_conn: + with sqlite3.connect(backup) as backup_conn: + source_conn.backup(backup_conn) + + def main() -> int: args = parse_args() repo = Path(__file__).resolve().parents[1] @@ -45,7 +56,7 @@ def main() -> int: target = tmp_path / "restored.sqlite3" try: fixture = create_ci_fixture_db(source) - backup.write_bytes(source.read_bytes()) + copy_sqlite_backup(source, backup) if any(path.resolve() == DEFAULT_DB_PATH.resolve() for path in [source, backup, target]): print(json.dumps({"success": False, "gate": "db-safety-smoke", "error_code": "operating_db_path_refused"}, ensure_ascii=False)) return 20 diff --git a/tests/test_kb_bank_pdf.py b/tests/test_kb_bank_pdf.py index 5412d54..91ee566 100644 --- a/tests/test_kb_bank_pdf.py +++ b/tests/test_kb_bank_pdf.py @@ -9,7 +9,18 @@ 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", "") -SAMPLE_AVAILABLE = PDF_PATH.exists() and bool(PDF_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") diff --git a/tests/test_toss_securities_pdf.py b/tests/test_toss_securities_pdf.py index 618b057..b60b1f6 100644 --- a/tests/test_toss_securities_pdf.py +++ b/tests/test_toss_securities_pdf.py @@ -10,7 +10,14 @@ from jarca_ledger_mcp.toss_securities_pdf import parse_toss_securities_pdf SAMPLE = Path("/root/.hermes/cache/documents/doc_f4cfe8e0221f_거래내역서_20250618_20260617_1.pdf") -@pytest.mark.skipif(not SAMPLE.exists(), reason="sample Toss Securities PDF is not available") +def _sample_available(path: Path) -> bool: + try: + return path.is_file() + except OSError: + return False + + +@pytest.mark.skipif(not _sample_available(SAMPLE), reason="sample Toss Securities PDF is not available") def test_parse_toss_securities_pdf_sample(): parsed = parse_toss_securities_pdf(str(SAMPLE)) assert parsed["metadata"]["holder_name"] == "고진환" @@ -22,7 +29,7 @@ def test_parse_toss_securities_pdf_sample(): assert any(tx.section == "USD" and tx.action == "외화증권배당금입금" for tx in parsed["transactions"]) -@pytest.mark.skipif(not SAMPLE.exists(), reason="sample Toss Securities PDF is not available") +@pytest.mark.skipif(not _sample_available(SAMPLE), reason="sample Toss Securities PDF is not available") def test_preview_and_import_toss_securities_pdf_sample(tmp_path): db_path = str(tmp_path / "ledger.sqlite3") preview = service.preview_toss_securities_pdf_import(str(SAMPLE), db_path=db_path)