feat: add ledger backup safety workflow
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
This commit is contained in:
@@ -0,0 +1,80 @@
|
||||
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
|
||||
Reference in New Issue
Block a user