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
72 lines
2.9 KiB
Python
72 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Restore/smoke helper for Jarca Ledger backups.
|
|
|
|
This script is intentionally not exposed as an MCP tool. It restores a backup file
|
|
into an explicitly provided target DB path, then runs SQLite integrity and Jarca
|
|
ledger health/status checks. Use a synthetic fixture or repo-external validation
|
|
DB first. Do not point --target at the live operating DB unless the user has
|
|
explicitly approved the restore.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from jarca_ledger_mcp import service
|
|
|
|
APPROVAL_TEXT = "RESTORE_APPROVED"
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Restore a ledger backup into a target DB and run smoke checks.")
|
|
parser.add_argument("--backup", required=True, help="Backup sqlite3 file to restore from")
|
|
parser.add_argument("--target", required=True, help="Target sqlite3 path to create/replace")
|
|
parser.add_argument(
|
|
"--approval-token",
|
|
default="",
|
|
help=f"Required token for overwrite/live restore. Must be {APPROVAL_TEXT!r}.",
|
|
)
|
|
parser.add_argument("--allow-overwrite", action="store_true", help="Allow replacing an existing target DB")
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
backup = Path(args.backup).expanduser().resolve()
|
|
target = Path(args.target).expanduser().resolve()
|
|
if not backup.exists():
|
|
print(json.dumps({"success": False, "error_code": "not_found", "error": f"backup not found: {backup}"}, ensure_ascii=False))
|
|
return 1
|
|
if target.exists() and not args.allow_overwrite:
|
|
print(json.dumps({"success": False, "error_code": "target_exists", "error": f"target exists: {target}"}, ensure_ascii=False))
|
|
return 2
|
|
if target.exists() and args.approval_token != APPROVAL_TEXT:
|
|
print(json.dumps({"success": False, "error_code": "approval_required", "error": f"overwrite requires --approval-token {APPROVAL_TEXT}"}, ensure_ascii=False))
|
|
return 3
|
|
|
|
before = service.get_ledger_backup_status(str(backup))
|
|
if not before.get("success") or before.get("backup", {}).get("integrity_check") != "ok":
|
|
print(json.dumps({"success": False, "error_code": "invalid_backup", "backup_status": before}, ensure_ascii=False))
|
|
return 4
|
|
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(backup, target)
|
|
health = service.health_check(str(target))
|
|
status = service.get_ledger_db_status(str(target))
|
|
result = {
|
|
"success": bool(health.get("success") and status.get("success") and status.get("integrity_check") == "ok"),
|
|
"backup": before.get("backup", {}),
|
|
"target": status,
|
|
"health_check": health,
|
|
"restore_policy": service.BACKUP_RESTORE_POLICY,
|
|
}
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0 if result["success"] else 5
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|