fix(ci): restore db safety smoke backup and sample-test skip behavior

This commit is contained in:
Jarca
2026-06-30 16:35:51 +09:00
parent 4929e52189
commit 26b815d723
3 changed files with 33 additions and 4 deletions
+12 -1
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import argparse import argparse
import json import json
import os import os
import sqlite3
import subprocess import subprocess
import sys import sys
import tempfile import tempfile
@@ -35,6 +36,16 @@ def parse_json_output(output: str) -> dict[str, Any]:
return {"success": False, "error_code": "invalid_json_output"} 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: def main() -> int:
args = parse_args() args = parse_args()
repo = Path(__file__).resolve().parents[1] repo = Path(__file__).resolve().parents[1]
@@ -45,7 +56,7 @@ def main() -> int:
target = tmp_path / "restored.sqlite3" target = tmp_path / "restored.sqlite3"
try: try:
fixture = create_ci_fixture_db(source) 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]): 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)) print(json.dumps({"success": False, "gate": "db-safety-smoke", "error_code": "operating_db_path_refused"}, ensure_ascii=False))
return 20 return 20
+12 -1
View File
@@ -9,7 +9,18 @@ from jarca_ledger_mcp import service
PDF_PATH = Path("/root/.hermes/cache/documents/doc_5a12af727ae0_KB거래내역조회_5704_2606171043.pdf") PDF_PATH = Path("/root/.hermes/cache/documents/doc_5a12af727ae0_KB거래내역조회_5704_2606171043.pdf")
PDF_PASSWORD = os.environ.get("KB_BANK_PDF_TEST_PASSWORD", "") 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") @pytest.mark.skipif(not SAMPLE_AVAILABLE, reason="sample KB Bank PDF/password is not available")
+9 -2
View File
@@ -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") 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(): def test_parse_toss_securities_pdf_sample():
parsed = parse_toss_securities_pdf(str(SAMPLE)) parsed = parse_toss_securities_pdf(str(SAMPLE))
assert parsed["metadata"]["holder_name"] == "고진환" 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"]) 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): def test_preview_and_import_toss_securities_pdf_sample(tmp_path):
db_path = str(tmp_path / "ledger.sqlite3") db_path = str(tmp_path / "ledger.sqlite3")
preview = service.preview_toss_securities_pdf_import(str(SAMPLE), db_path=db_path) preview = service.preview_toss_securities_pdf_import(str(SAMPLE), db_path=db_path)