from __future__ import annotations import argparse import ctypes import os import shlex import shutil import subprocess import sys from ctypes import wintypes from datetime import datetime from pathlib import Path RETRY_CANCEL_CHOICES = {"n", "no", "c", "cancel", "a", "abort", "q", "quit"} ACTIVE_BACKUP_FILENAME = ".active-backup" def show_phase(title: str) -> None: print() print(f"========== {title} ==========") def test_file_unlocked(path: Path) -> bool: if not path.exists(): return True if os.name == "nt": return _test_file_unlocked_windows(path) return _test_file_unlocked_posix(path) def _test_file_unlocked_windows(path: Path) -> bool: generic_read = 0x80000000 generic_write = 0x40000000 open_existing = 3 file_attribute_normal = 0x80 invalid_handle_value = wintypes.HANDLE(-1).value kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.CreateFileW.argtypes = [ wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE, ] kernel32.CreateFileW.restype = wintypes.HANDLE kernel32.CloseHandle.argtypes = [wintypes.HANDLE] kernel32.CloseHandle.restype = wintypes.BOOL handle = kernel32.CreateFileW( str(path), generic_read | generic_write, 0, None, open_existing, file_attribute_normal, None, ) if handle == invalid_handle_value: return False kernel32.CloseHandle(handle) return True def _test_file_unlocked_posix(path: Path) -> bool: import fcntl fd = os.open(path, os.O_RDWR) try: try: fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: return False else: fcntl.flock(fd, fcntl.LOCK_UN) return True finally: os.close(fd) def wait_for_restore_preflight(db_file_path: Path) -> bool: attempt = 1 while not test_file_unlocked(db_file_path): show_phase("Restore Preflight") print(f"WARNING: Restore preflight blocked: database appears to be in use: {db_file_path}") print( "Close conflicting applications (for example DB Browser for SQLite, uvicorn, or any process using this DB)." ) print("Restore is paused and waiting for your input.") answer = input(f"Attempt {attempt}: type 'retry' to check again, or 'cancel' to skip restore: ").strip() print(f"Input received: '{answer}'") if answer.lower() in RETRY_CANCEL_CHOICES: return False attempt += 1 print("Re-checking database lock now...") if attempt > 1: print("Restore preflight passed: database lock released.") return True def restore_backup(backup_path: Path, data_path: Path) -> None: if not backup_path.exists(): raise FileNotFoundError(f"Backup path not found: {backup_path}") if data_path.exists(): shutil.rmtree(data_path) shutil.copytree(backup_path, data_path) print(f"Restored data from backup: {backup_path}") def active_backup_path(backup_root: Path) -> Path | None: marker_path = backup_root / ACTIVE_BACKUP_FILENAME if not marker_path.exists(): return None backup_name = marker_path.read_text(encoding="utf-8").strip() if not backup_name or Path(backup_name).name != backup_name: raise RuntimeError(f"Invalid active backup marker: {marker_path}") backup_path = backup_root / backup_name if not backup_path.is_dir(): raise FileNotFoundError( f"Active backup is missing: {backup_path}. " f"Restore it or remove {marker_path} only after confirming the original data is safe." ) return backup_path def create_or_reuse_backup(data_path: Path, backup_root: Path) -> Path: current_backup = active_backup_path(backup_root) if current_backup is not None: print(f"Reusing active test-cycle backup: {current_backup}") return current_backup db_path = data_path / "transcription.db" if not test_file_unlocked(db_path): print(f"WARNING: Backup preflight warning: database appears to be in use: {db_path}") print("WARNING: Proceeding with backup, but hot backups can capture an in-flight state.") timestamp = datetime.now().astimezone().strftime("%Y%m%d-%H%M%S") backup_path = backup_root / f"data-backup-{timestamp}" shutil.copytree(data_path, backup_path) marker_path = backup_root / ACTIVE_BACKUP_FILENAME temporary_marker_path = marker_path.with_suffix(".tmp") temporary_marker_path.write_text(backup_path.name, encoding="utf-8") temporary_marker_path.replace(marker_path) print(f"Created test-cycle backup: {backup_path}") return backup_path def close_active_backup_cycle(backup_root: Path, backup_path: Path | None = None) -> Path: current_backup = active_backup_path(backup_root) if current_backup is None: raise RuntimeError("No active test backup cycle exists.") if backup_path is not None and current_backup.resolve() != backup_path.resolve(): return current_backup (backup_root / ACTIVE_BACKUP_FILENAME).unlink() print(f"Closed test backup cycle: {current_backup}") return current_backup def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run potentially destructive tests with backup/restore protection.") parser.add_argument("--auto-restore", action="store_true", help="Restore immediately after successful tests.") parser.add_argument("--keep-backup", action="store_true", help="Keep the backup even after a successful restore.") parser.add_argument( "--skip-restore-prompt", action="store_true", help="Do not prompt after successful tests; keep the backup for a later manual restore.", ) parser.add_argument( "--restore-from", help="Restore from an existing backup name or absolute backup path instead of running tests.", ) parser.add_argument( "--accept-current-data", action="store_true", help="Close the active test cycle without restoring; preserve its backup.", ) parser.add_argument("command", nargs=argparse.REMAINDER, help="Command to run after '--'.") args = parser.parse_args(argv) if args.restore_from and args.accept_current_data: parser.error("--restore-from cannot be combined with --accept-current-data.") if (args.restore_from or args.accept_current_data) and args.command: parser.error("Restore/accept modes cannot be combined with a test command.") if not args.restore_from and not args.accept_current_data and not args.command: parser.error("A test command is required unless --restore-from or --accept-current-data is provided.") return args def normalize_command(command: list[str]) -> list[str]: if command and command[0] == "--": return command[1:] return command def run_command(command: list[str]) -> int: show_phase("Test Phase") print(f"Running command: {shlex.join(command)}") try: completed = subprocess.run(command, check=False) except OSError as exc: print(f"ERROR: {exc}", file=sys.stderr) return 1 return completed.returncode def accept_current_data(backup_root: Path) -> int: show_phase("Complete Test Cycle") backup_path = close_active_backup_cycle(backup_root) print(f"Current data accepted. Backup preserved at: {backup_path}") return 0 def restore_saved_backup(restore_from: str, data_path: Path, backup_root: Path) -> int: restore_path = Path(restore_from) if not restore_path.is_absolute(): restore_path = backup_root / restore_path show_phase("Restore Phase") if not wait_for_restore_preflight(data_path / "transcription.db"): print(f"Restore cancelled. Backup preserved at: {restore_path}") return 1 restore_backup(restore_path, data_path) current_backup = active_backup_path(backup_root) if current_backup is not None: if current_backup.resolve() == restore_path.resolve(): close_active_backup_cycle(backup_root, restore_path) else: print(f"Active test cycle remains unchanged: {current_backup}") return 0 def restore_requested(args: argparse.Namespace) -> bool: if args.auto_restore: return True if args.skip_restore_prompt: return False answer = input("Tests passed. Restore data backup now? [y/N] ").strip().lower() return answer in {"y", "yes"} def run_protected_tests(args: argparse.Namespace, data_path: Path, backup_root: Path) -> int: command = normalize_command(list(args.command)) if not command: raise ValueError("No test command provided.") show_phase("Backup Phase") backup_path = create_or_reuse_backup(data_path, backup_root) backup_name = backup_path.name test_exit_code = run_command(command) if test_exit_code != 0: show_phase("Post-Test") print(f"Tests failed (exit code {test_exit_code}).") print(f"Backup preserved at: {backup_path}") print("Inspect state, then restore manually if needed:") print(f" uv run python tools/run_destructive_tests.py --restore-from {backup_name}") return test_exit_code if not restore_requested(args): print(f"Restore skipped by user. Test cycle remains active with backup: {backup_path}") print("Further test runs will reuse this backup instead of creating another.") print("Restore later with:") print(f" uv run python tools/run_destructive_tests.py --restore-from {backup_name}") print("Or accept the current data and close the cycle with:") print(" uv run python tools/run_destructive_tests.py --accept-current-data") return 0 show_phase("Restore Phase") if not wait_for_restore_preflight(data_path / "transcription.db"): print(f"Restore cancelled. Backup preserved at: {backup_path}") return 1 try: restore_backup(backup_path, data_path) close_active_backup_cycle(backup_root, backup_path) if args.keep_backup: print(f"Kept backup: {backup_path}") else: shutil.rmtree(backup_path) print(f"Deleted backup: {backup_path}") except (OSError, RuntimeError) as exc: print("WARNING: Restore failed. The backup remains available.") print(f"WARNING: {exc}") print("Likely cause: another process has data/transcription.db open.") print("Stop the process and retry restore with:") print(f" uv run python tools/run_destructive_tests.py --restore-from {backup_name}") return 1 return 0 def main(argv: list[str] | None = None) -> int: args = parse_args(argv or sys.argv[1:]) repo_root = Path(__file__).resolve().parent.parent data_path = repo_root / "data" backup_root = repo_root / ".test-backups" if not data_path.exists(): raise FileNotFoundError(f"Data directory not found: {data_path}") backup_root.mkdir(parents=True, exist_ok=True) if args.accept_current_data: return accept_current_data(backup_root) if args.restore_from: return restore_saved_backup(args.restore_from, data_path, backup_root) return run_protected_tests(args, data_path, backup_root) if __name__ == "__main__": raise SystemExit(main())