119 lines
4.0 KiB
Python
119 lines
4.0 KiB
Python
"""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
|
|
);
|
|
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);
|
|
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.*"))
|