324 lines
12 KiB
Python
324 lines
12 KiB
Python
"""Focused tests for the fail-closed Cassandra Kanban recovery."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
SCRIPT = ROOT / "services/hermes/scripts/repair_cassandra_kanban.py"
|
|
sys.path.insert(0, str(SCRIPT.parent))
|
|
status_recovery = importlib.import_module("kanban_status_recovery")
|
|
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 _task_status_alias_database(path: Path) -> Path:
|
|
"""Create the production-shaped rows needed for status reconciliation."""
|
|
connection = sqlite3.connect(path)
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE tasks (
|
|
id TEXT PRIMARY KEY, title TEXT NOT NULL, assignee TEXT,
|
|
status TEXT NOT NULL, result TEXT, completed_at INTEGER,
|
|
claim_lock TEXT, claim_expires INTEGER, worker_pid INTEGER,
|
|
current_run_id INTEGER, block_kind TEXT,
|
|
block_recurrences INTEGER NOT NULL DEFAULT 0,
|
|
tenant TEXT, idempotency_key TEXT, session_id TEXT
|
|
);
|
|
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, kind TEXT NOT NULL, payload TEXT, created_at INTEGER NOT NULL
|
|
);
|
|
CREATE TABLE task_runs (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL,
|
|
status TEXT NOT NULL, outcome TEXT, ended_at INTEGER,
|
|
claim_lock TEXT, claim_expires INTEGER, worker_pid INTEGER,
|
|
summary TEXT, metadata TEXT, error TEXT, started_at INTEGER 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_tasks_status ON tasks(status);
|
|
CREATE INDEX idx_tasks_assignee_status ON tasks(assignee, status);
|
|
CREATE INDEX idx_tasks_idempotency ON tasks(idempotency_key);
|
|
CREATE INDEX idx_tasks_session_id ON tasks(session_id);
|
|
CREATE INDEX idx_tasks_tenant ON tasks(tenant);
|
|
CREATE INDEX idx_runs_status ON task_runs(status);
|
|
CREATE INDEX idx_runs_task ON task_runs(task_id, started_at);
|
|
INSERT INTO tasks(
|
|
id, title, assignee, status, claim_lock, claim_expires,
|
|
current_run_id, block_kind, block_recurrences
|
|
) VALUES (
|
|
'task-1', 'Recover completion', 'worker', 'running',
|
|
'stale-claim', 99, 1, 'capability', 2
|
|
);
|
|
INSERT INTO task_runs(
|
|
task_id, status, claim_lock, claim_expires, started_at
|
|
) VALUES ('task-1', 'running', 'stale-claim', 99, 1);
|
|
INSERT INTO task_events(task_id, run_id, kind, payload, created_at)
|
|
VALUES (
|
|
'task-1', 1, 'completed',
|
|
'{"result_len": 18, "summary": "verified completion"}', 42
|
|
);
|
|
"""
|
|
)
|
|
connection.commit()
|
|
connection.close()
|
|
return path
|
|
|
|
|
|
def _task_status_alias_errors(database: Path) -> list[str]:
|
|
with sqlite3.connect(database) as connection:
|
|
root = connection.execute(
|
|
"SELECT rootpage FROM sqlite_master WHERE name = 'tasks'"
|
|
).fetchone()[0]
|
|
return [
|
|
"*** in database main ***\n"
|
|
f"Tree {root} page 16 cell 1: Rowid 1 out of order\n"
|
|
f"Tree {root} page 16 cell 1: 2nd reference to page 17",
|
|
"wrong # of entries in index idx_tasks_status",
|
|
"row 1 missing from index idx_tasks_assignee_status",
|
|
"row 1 missing from index idx_runs_status",
|
|
]
|
|
|
|
|
|
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",
|
|
)
|
|
|
|
|
|
def test_known_task_status_alias_uses_independent_completion_evidence(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
database = _task_status_alias_database(tmp_path / "kanban.db")
|
|
original_indexed_values = status_recovery._indexed_values
|
|
|
|
def completed_index_values(connection, table, index, columns):
|
|
values = original_indexed_values(connection, table, index, columns)
|
|
if columns == ("status",) and index in {
|
|
*status_recovery.TASK_STATUS_INDEXES,
|
|
status_recovery.RUN_STATUS_INDEX,
|
|
}:
|
|
values[1] = ("done",)
|
|
return values
|
|
|
|
monkeypatch.setattr(
|
|
status_recovery, "_indexed_values", completed_index_values
|
|
)
|
|
|
|
result = recovery.recover_database(
|
|
database, _task_status_alias_errors(database)
|
|
)
|
|
|
|
assert result["state"] == "recovered"
|
|
with sqlite3.connect(database) as connection:
|
|
assert connection.execute("PRAGMA integrity_check").fetchone() == ("ok",)
|
|
assert connection.execute(
|
|
"SELECT status, result, completed_at, claim_lock, current_run_id, "
|
|
"block_kind, block_recurrences FROM tasks WHERE id = 'task-1'"
|
|
).fetchone() == (
|
|
"done",
|
|
"verified completion",
|
|
42,
|
|
None,
|
|
None,
|
|
None,
|
|
0,
|
|
)
|
|
assert connection.execute(
|
|
"SELECT status, outcome, ended_at, claim_lock, summary "
|
|
"FROM task_runs WHERE id = 1"
|
|
).fetchone() == (
|
|
"done",
|
|
"completed",
|
|
42,
|
|
None,
|
|
"verified completion",
|
|
)
|
|
|
|
|
|
def test_task_status_alias_refuses_disagreeing_status_indexes(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
database = _task_status_alias_database(tmp_path / "kanban.db")
|
|
original_indexed_values = status_recovery._indexed_values
|
|
|
|
def disagreeing_index_values(connection, table, index, columns):
|
|
values = original_indexed_values(connection, table, index, columns)
|
|
if (
|
|
columns == ("status",)
|
|
and index == status_recovery.TASK_STATUS_INDEXES[0]
|
|
):
|
|
values[1] = ("done",)
|
|
return values
|
|
|
|
monkeypatch.setattr(
|
|
status_recovery, "_indexed_values", disagreeing_index_values
|
|
)
|
|
|
|
with pytest.raises(
|
|
recovery.RecoveryRefused,
|
|
match="independent task status indexes disagree",
|
|
):
|
|
recovery.recover_database(database, _task_status_alias_errors(database))
|
|
|
|
|
|
@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)
|