lllin000_PaperForge/tests/test_memory_restore.py

"""Focused tests for memory restore-backup (Issue #78 repair).

Covers:
- CLI restore-backup subcommand
- Backup validation (integrity_check)
- Corrupt DB preservation
- Atomic replace (os.replace)
- Invalid backup safety (no restore)
- Probe emits restore_backup action
"""

from __future__ import annotations

import json
import sqlite3
import subprocess
import sys
from pathlib import Path

import pytest


def _run_memory_cmd(vault: Path, subcmd: str, extra_args: list[str] | None = None) -> subprocess.CompletedProcess:
    cmd = [sys.executable, "-m", "paperforge", "--vault", str(vault), "memory", subcmd]
    if extra_args:
        cmd.extend(extra_args)
    return subprocess.run(cmd, capture_output=True, text=True, timeout=15)


def _setup_memory_dirs(tmp_path: Path) -> Path:
    """Create paperforge.json and indexes dir, return indexes dir."""
    (tmp_path / "paperforge.json").write_text(
        json.dumps({"system_dir": "99_System"}), encoding="utf-8",
    )
    indexes = tmp_path / "99_System" / "PaperForge" / "indexes"
    indexes.mkdir(parents=True, exist_ok=True)
    return indexes


def _create_valid_db(db_path: Path, paper_count: int = 3) -> None:
    """Create a valid paperforge.db with schema_version and papers."""
    conn = sqlite3.connect(str(db_path))
    conn.execute("CREATE TABLE papers (zotero_key TEXT PRIMARY KEY, title TEXT)")
    conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT)")
    conn.execute("INSERT INTO meta VALUES ('schema_version', '8')")
    for i in range(paper_count):
        conn.execute("INSERT INTO papers VALUES (?, ?)", (f"KEY{i}", f"Paper {i}"))
    conn.commit()
    conn.close()


class TestMemoryRestoreCli:
    """CLI restore-backup subcommand acceptance."""

    def test_restore_backup_no_backup_file(self, tmp_path: Path) -> None:
        """No backup file -> exits nonzero."""
        indexes = _setup_memory_dirs(tmp_path)
        _create_valid_db(indexes / "paperforge.db")
        result = _run_memory_cmd(tmp_path, "restore-backup", ["--json"])
        data = json.loads(result.stdout)
        assert data["ok"] is False
        assert "No backup found" in data["error"]["message"]

    def test_restore_backup_invalid_backup(self, tmp_path: Path) -> None:
        """Invalid backup (not a database) -> exits nonzero, current DB untouched."""
        indexes = _setup_memory_dirs(tmp_path)
        db_path = indexes / "paperforge.db"
        backup_path = indexes / "paperforge.db.backup"
        _create_valid_db(db_path)
        backup_path.write_text("not a database", encoding="utf-8")
        result = _run_memory_cmd(tmp_path, "restore-backup", ["--json"])
        data = json.loads(result.stdout)
        assert data["ok"] is False
        assert "integrity check" in data["error"]["message"].lower() or "validate backup" in data["error"]["message"].lower() or "not a database" in data["error"]["message"].lower()
        # Current DB still intact
        assert db_path.exists()

    def test_restore_backup_success(self, tmp_path: Path) -> None:
        """Valid backup -> restores successfully, old DB preserved."""
        indexes = _setup_memory_dirs(tmp_path)
        db_path = indexes / "paperforge.db"
        backup_path = indexes / "paperforge.db.backup"

        # Create current DB with old content
        _create_valid_db(db_path, paper_count=1)

        # Create backup with more papers
        _create_valid_db(backup_path, paper_count=5)

        result = _run_memory_cmd(tmp_path, "restore-backup", ["--json"])
        assert result.returncode == 0
        data = json.loads(result.stdout)
        assert data["ok"] is True
        assert data["data"]["action"] == "restore_backup"

        # Verify restored DB has 5 papers
        conn = sqlite3.connect(str(db_path))
        row = conn.execute("SELECT COUNT(*) FROM papers").fetchone()
        conn.close()
        assert row[0] == 5

        # Verify backup still exists (not consumed)
        assert backup_path.exists()

        # Verify old DB preserved as corrupt snapshot
        corrupt_files = list(indexes.glob("paperforge.corrupt-*.db"))
        assert len(corrupt_files) == 1

    def test_restore_backup_without_current_db(self, tmp_path: Path) -> None:
        """No current DB, just backup -> restores directly."""
        indexes = _setup_memory_dirs(tmp_path)
        db_path = indexes / "paperforge.db"
        backup_path = indexes / "paperforge.db.backup"
        _create_valid_db(backup_path, paper_count=3)

        result = _run_memory_cmd(tmp_path, "restore-backup", ["--json"])
        assert result.returncode == 0
        data = json.loads(result.stdout)
        assert data["ok"] is True

        # Verify restored
        conn = sqlite3.connect(str(db_path))
        row = conn.execute("SELECT COUNT(*) FROM papers").fetchone()
        conn.close()
        assert row[0] == 3

        # No corrupt snapshot since there was no current DB
        corrupt_files = list(indexes.glob("paperforge.corrupt-*.db"))
        assert len(corrupt_files) == 0


    def test_replace_failure_leaves_db_and_backup_untouched(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
        """When os.replace fails, current DB and backup remain unchanged (direct call)."""
        import sqlite3
        from paperforge.commands.memory import _restore_backup

        indexes = _setup_memory_dirs(tmp_path)
        db_path = indexes / "paperforge.db"
        backup_path = indexes / "paperforge.db.backup"

        # Create current DB with known content
        conn = sqlite3.connect(str(db_path))
        conn.execute("CREATE TABLE papers (zotero_key TEXT PRIMARY KEY, title TEXT)")
        conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT)")
        conn.execute("INSERT INTO meta VALUES ('schema_version', '8')")
        conn.execute("INSERT INTO papers VALUES ('OLD', 'Old Paper')")
        conn.commit()
        conn.close()
        original_bytes = db_path.read_bytes()

        # Create backup with different content
        conn = sqlite3.connect(str(backup_path))
        conn.execute("CREATE TABLE papers (zotero_key TEXT PRIMARY KEY, title TEXT)")
        conn.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT)")
        conn.execute("INSERT INTO meta VALUES ('schema_version', '8')")
        conn.execute("INSERT INTO papers VALUES ('NEW', 'New Paper')")
        conn.commit()
        conn.close()
        backup_bytes = backup_path.read_bytes()

        # Monkeypatch os.replace to fail (direct call, so monkeypatch works)
        import os as _os
        def _failing_replace(src, dst):
            raise OSError("simulated replace failure")
        monkeypatch.setattr(_os, "replace", _failing_replace)

        result = _restore_backup(tmp_path)
        assert result.ok is False
        assert "Failed to restore backup" in result.error.message

        # Original DB still intact (byte-identical)
        assert db_path.read_bytes() == original_bytes

        # Backup still intact
        assert backup_path.exists()
        assert backup_path.read_bytes() == backup_bytes

        # Clean up temp files
        for f in indexes.glob("paperforge.restore-*"):
            f.unlink(missing_ok=True)


class TestMemoryRestoreProbe:
    """Probe emits restore_backup when corrupt DB + backup exists."""

    def test_corrupt_db_with_backup_emits_restore(self, tmp_path: Path) -> None:
        """Corrupt DB + valid backup -> restore_backup action."""
        from paperforge.commands import probe as probe_mod

        (tmp_path / "paperforge.json").write_text(
            json.dumps({"system_dir": "99_System"}), encoding="utf-8",
        )
        indexes = tmp_path / "99_System" / "PaperForge" / "indexes"
        indexes.mkdir(parents=True, exist_ok=True)

        # Create corrupt DB
        (indexes / "paperforge.db").write_text("not a database", encoding="utf-8")
        # Create valid backup
        _create_valid_db(indexes / "paperforge.db.backup", paper_count=3)

        data = probe_mod.probe_memory(tmp_path)
        assert data["capability_state"] == "unavailable"
        assert data["reason"]["code"] == "memory.db_corrupt"
        assert data["action"]["primary"]["verb"] == "restore_backup"
        assert data["action"]["primary"]["command"] == "paperforge memory restore-backup"
        assert data["action"]["primary"]["destructive"] is True
        assert data["action"]["primary"]["confirmation_required"] is True

    def test_corrupt_db_without_backup_emits_rebuild(self, tmp_path: Path) -> None:
        """Corrupt DB without backup -> rebuild action."""
        from paperforge.commands import probe as probe_mod

        (tmp_path / "paperforge.json").write_text(
            json.dumps({"system_dir": "99_System"}), encoding="utf-8",
        )
        indexes = tmp_path / "99_System" / "PaperForge" / "indexes"
        indexes.mkdir(parents=True, exist_ok=True)

        # Create corrupt DB, no backup
        (indexes / "paperforge.db").write_text("not a database", encoding="utf-8")

        data = probe_mod.probe_memory(tmp_path)
        assert data["capability_state"] == "unavailable"
        assert data["reason"]["code"] == "memory.db_corrupt"
        assert data["action"]["primary"]["verb"] == "run"
        assert data["action"]["primary"]["command"] == "paperforge memory build"