atlas-iac/testing/tests/test_hermes_kanban_recovery.py

"""Focused tests for the fail-closed Cassandra Kanban recovery."""

from __future__ import annotations

import importlib.util
import sqlite3
from pathlib import Path

import pytest


ROOT = Path(__file__).resolve().parents[2]
SCRIPT = ROOT / "services/hermes/scripts/repair_cassandra_kanban.py"
SPEC = importlib.util.spec_from_file_location("repair_cassandra_kanban", SCRIPT)
assert SPEC and SPEC.loader
recovery = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(recovery)


def _database(path: Path) -> Path:
    connection = sqlite3.connect(path)
    connection.executescript(
        """
        CREATE TABLE tasks (id TEXT PRIMARY KEY, title TEXT NOT NULL);
        CREATE TABLE task_links (
          parent_id TEXT NOT NULL, child_id TEXT NOT NULL,
          PRIMARY KEY (parent_id, child_id)
        );
        CREATE TABLE task_comments (
          id INTEGER PRIMARY KEY AUTOINCREMENT,
          task_id TEXT NOT NULL,
          author TEXT NOT NULL,
          body TEXT NOT NULL,
          created_at INTEGER NOT NULL
        );
        CREATE TABLE task_events (
          id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL,
          run_id INTEGER, created_at INTEGER NOT NULL DEFAULT 1
        );
        CREATE TABLE task_runs (
          id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL
        );
        CREATE TABLE task_attachments (
          id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL
        );
        CREATE TABLE kanban_notify_subs (
          task_id TEXT NOT NULL, platform TEXT NOT NULL,
          PRIMARY KEY (task_id, platform)
        );
        CREATE INDEX idx_comments_task ON task_comments(task_id, created_at);
        CREATE INDEX idx_events_task ON task_events(task_id, created_at);
        CREATE INDEX idx_events_run ON task_events(run_id, id);
        INSERT INTO tasks VALUES ('task-1', 'Preserve me');
        INSERT INTO task_comments(task_id, author, body, created_at)
          VALUES ('task-1', 'worker', 'evidence', 1);
        INSERT INTO task_events(task_id) VALUES ('task-1');
        INSERT INTO task_runs(task_id) VALUES ('task-1');
        INSERT INTO kanban_notify_subs VALUES ('task-1', 'telegram');
        """
    )
    connection.commit()
    connection.close()
    return path


def _known_errors(database: Path) -> list[str]:
    with sqlite3.connect(database) as connection:
        root = connection.execute(
            "SELECT rootpage FROM sqlite_master WHERE name = 'task_comments'"
        ).fetchone()[0]
    return [
        "*** in database main ***\n"
        f"Tree {root} page {root} cell 0: 2nd reference to page 41\n"
        f"Tree {root} page 37 cell 4: Rowid 11 out of order",
        "NUMERIC value in task_comments.author",
        "row 2 missing from index idx_comments_task",
    ]


def test_healthy_board_is_untouched(tmp_path: Path) -> None:
    database = _database(tmp_path / "kanban.db")
    before = database.read_bytes()

    result = recovery.repair_if_needed(database)

    assert result["state"] == "healthy"
    assert database.read_bytes() == before
    assert not list(tmp_path.glob("*.corrupt.*"))


def test_known_corruption_rebuilds_rows_and_retains_backup(tmp_path: Path) -> None:
    database = _database(tmp_path / "kanban.db")
    before = database.read_bytes()

    result = recovery.recover_database(database, _known_errors(database))

    assert result["state"] == "recovered"
    backup = Path(result["backup"])
    assert backup.read_bytes() == before
    with sqlite3.connect(database) as connection:
        assert connection.execute("PRAGMA integrity_check").fetchone() == ("ok",)
        assert connection.execute("SELECT title FROM tasks").fetchone() == (
            "Preserve me",
        )
        assert connection.execute(
            "SELECT author, body FROM task_comments"
        ).fetchone() == (
            "worker",
            "evidence",
        )


def test_unknown_corruption_is_preserved_and_refused(tmp_path: Path) -> None:
    database = _database(tmp_path / "kanban.db")
    before = database.read_bytes()

    with pytest.raises(recovery.RecoveryRefused, match="does not match"):
        recovery.recover_database(database, ["freelist leaf count is too big"])

    assert database.read_bytes() == before
    assert not list(tmp_path.glob("*.corrupt.*"))


def test_known_event_index_mismatch_rebuilds_rows(tmp_path: Path) -> None:
    database = _database(tmp_path / "kanban.db")
    before = database.read_bytes()

    result = recovery.recover_database(
        database,
        [
            "wrong # of entries in index idx_events_task",
            "wrong # of entries in index idx_events_run",
        ],
    )

    assert result["state"] == "recovered"
    assert Path(result["backup"]).read_bytes() == before
    with sqlite3.connect(database) as connection:
        assert connection.execute("PRAGMA integrity_check").fetchone() == ("ok",)
        assert connection.execute("SELECT task_id FROM task_events").fetchone() == (
            "task-1",
        )


@pytest.mark.parametrize(
    "errors",
    [
        ["wrong # of entries in index idx_comments_task"],
        [
            "wrong # of entries in index idx_events_task",
            "freelist leaf count is too big",
        ],
    ],
)
def test_unreviewed_index_failures_remain_refused(
    tmp_path: Path, errors: list[str]
) -> None:
    database = _database(tmp_path / "kanban.db")

    with pytest.raises(recovery.RecoveryRefused, match="does not match"):
        recovery.recover_database(database, errors)