transcription/tools/run_destructive_tests.py

from __future__ import annotations

import argparse
import ctypes
import os
import shlex
import shutil
import subprocess
import sys
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} ==========")


if sys.platform == "win32":
    # `ctypes.wintypes` raises on import off Windows, and `fcntl` does not exist
    # on Windows, so the two implementations are selected at module level where a
    # type checker can narrow `sys.platform` and analyze only the live branch.
    from ctypes import wintypes

    def test_file_unlocked(path: Path) -> bool:
        if not path.exists():
            return True

        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

else:
    import fcntl

    def test_file_unlocked(path: Path) -> bool:
        if not path.exists():
            return True

        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())