Jarca
67d8d4871b
confirm dcy-002 and add REQ/BAT evidence add ledger backup/status metadata workflow document restore as runbook/script-only guarded operation add synthetic DB backup/restore smoke tests
81 lines
3.2 KiB
Python
81 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from jarca_ledger_mcp import service
|
|
|
|
|
|
def test_db_status_reports_metadata_without_sensitive_rows(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_ledger_transaction(
|
|
"2026-06-01",
|
|
"expense",
|
|
12_000,
|
|
category="식비",
|
|
subcategory="카페",
|
|
merchant="SENTINEL_MERCHANT",
|
|
counterparty="SENTINEL_COUNTERPARTY",
|
|
raw_payload="SENTINEL_RAW_PAYLOAD",
|
|
db_path=db_path,
|
|
)
|
|
|
|
status = service.get_ledger_db_status(db_path=db_path)
|
|
|
|
assert status["success"] is True
|
|
assert status["db_path"] == str(Path(db_path).resolve())
|
|
assert status["exists"] is True
|
|
assert status["sha256"]
|
|
assert status["integrity_check"] == "ok"
|
|
assert status["table_counts"]["transactions"] == 1
|
|
dumped = repr(status)
|
|
assert "SENTINEL_MERCHANT" not in dumped
|
|
assert "SENTINEL_COUNTERPARTY" not in dumped
|
|
assert "SENTINEL_RAW_PAYLOAD" not in dumped
|
|
|
|
|
|
def test_create_list_and_status_ledger_backup_with_retention(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
backup_dir = str(tmp_path / "backups")
|
|
service.create_ledger_account("테스트 계좌", db_path=db_path)
|
|
before = service.get_ledger_db_status(db_path=db_path)
|
|
|
|
created = service.create_ledger_backup(db_path=db_path, backup_dir=backup_dir, retention_days=30, retention_count=20)
|
|
|
|
assert created["success"] is True
|
|
assert created["source_db"]["sha256"] == before["sha256"]
|
|
assert created["backup"]["path"].startswith(str(Path(backup_dir).resolve()))
|
|
assert Path(created["backup"]["path"]).exists()
|
|
assert created["backup"]["sha256"]
|
|
assert created["backup"]["integrity_check"] == "ok"
|
|
assert created["backup"]["table_counts"] == created["source_db"]["table_counts"]
|
|
assert created["restore_policy"]["mcp_restore_exposed"] is False
|
|
|
|
listed = service.list_ledger_backups(backup_dir=backup_dir, include_hash=True)
|
|
assert listed["success"] is True
|
|
assert listed["count"] == 1
|
|
assert listed["backups"][0]["sha256"] == created["backup"]["sha256"]
|
|
|
|
backup_status = service.get_ledger_backup_status(created["backup"]["path"])
|
|
assert backup_status["success"] is True
|
|
assert backup_status["backup"]["integrity_check"] == "ok"
|
|
assert backup_status["backup"]["table_counts"]["accounts"] == 1
|
|
|
|
|
|
def test_backup_retention_keeps_recent_count_and_prunes_older_extras(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
backup_dir = tmp_path / "backups"
|
|
service.create_ledger_transaction("2026-06-01", "income", 1000, category="수입", db_path=db_path)
|
|
|
|
for i in range(3):
|
|
old = backup_dir / f"ledger_20200101000{i}000000.sqlite3"
|
|
old.parent.mkdir(parents=True, exist_ok=True)
|
|
old.write_text("old", encoding="utf-8")
|
|
|
|
created = service.create_ledger_backup(db_path=db_path, backup_dir=str(backup_dir), retention_days=0, retention_count=2)
|
|
|
|
assert created["success"] is True
|
|
remaining = sorted(p.name for p in backup_dir.glob("*.sqlite3"))
|
|
assert len(remaining) == 2
|
|
assert Path(created["backup"]["path"]).name in remaining
|
|
assert len(created["retention"]["pruned_paths"]) == 2
|