chore(jarca-ledger): finalize approved batch docs, dashboard, and CI smoke updates
This commit is contained in:
Executable
+81
@@ -0,0 +1,81 @@
|
||||
#!/usr/bin/env python3
|
||||
"""CI wrapper for Jarca Ledger DB restore/safety smoke.
|
||||
|
||||
The wrapper builds a synthetic fixture DB under a temporary directory, copies it
|
||||
through the existing restore smoke script, and emits only status-level evidence.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(REPO_ROOT))
|
||||
|
||||
from jarca_ledger_mcp.db import DEFAULT_DB_PATH
|
||||
from scripts.ci_fixture_db import create_ci_fixture_db
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description="Run DB safety smoke with synthetic temp DBs only.")
|
||||
parser.add_argument("--timeout", type=int, default=45, help="Subprocess timeout in seconds")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def parse_json_output(output: str) -> dict[str, Any]:
|
||||
try:
|
||||
return json.loads(output)
|
||||
except json.JSONDecodeError:
|
||||
return {"success": False, "error_code": "invalid_json_output"}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
repo = Path(__file__).resolve().parents[1]
|
||||
with tempfile.TemporaryDirectory(prefix="jarca-ledger-ci-db-") as tmp:
|
||||
tmp_path = Path(tmp)
|
||||
source = tmp_path / "fixture.sqlite3"
|
||||
backup = tmp_path / "backup.sqlite3"
|
||||
target = tmp_path / "restored.sqlite3"
|
||||
try:
|
||||
fixture = create_ci_fixture_db(source)
|
||||
backup.write_bytes(source.read_bytes())
|
||||
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
|
||||
env = os.environ.copy()
|
||||
env["JARCA_LEDGER_DB_PATH"] = str(source)
|
||||
cmd = [sys.executable, "scripts/ledger_restore_smoke.py", "--backup", str(backup), "--target", str(target)]
|
||||
run = subprocess.run(cmd, cwd=repo, env=env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=args.timeout, check=False)
|
||||
except subprocess.TimeoutExpired:
|
||||
print(json.dumps({"success": False, "gate": "db-safety-smoke", "error_code": "timeout", "timeout_seconds": args.timeout}, ensure_ascii=False))
|
||||
return 21
|
||||
except Exception as exc: # noqa: BLE001 - deterministic wrapper failure
|
||||
print(json.dumps({"success": False, "gate": "db-safety-smoke", "error_code": "wrapper_error", "error": exc.__class__.__name__}, ensure_ascii=False))
|
||||
return 1
|
||||
|
||||
smoke = parse_json_output(run.stdout)
|
||||
success = run.returncode == 0 and bool(smoke.get("success")) and target.exists()
|
||||
result = {
|
||||
"success": success,
|
||||
"gate": "db-safety-smoke",
|
||||
"fixture": "synthetic",
|
||||
"fixture_integrity": fixture.get("integrity_check"),
|
||||
"restore_smoke_exit_code": run.returncode,
|
||||
"restored_exists": target.exists(),
|
||||
"restored_integrity": smoke.get("target", {}).get("integrity_check"),
|
||||
"health_success": smoke.get("health_check", {}).get("success"),
|
||||
"operating_db_used": False,
|
||||
}
|
||||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||||
return 0 if success else 22
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user