test(state): prove silent WAL fallback in callers

This commit is contained in:
Connor Black 2026-06-19 12:08:14 -04:00 committed by Teknium
parent 0f60cdac27
commit 144e563cdd
2 changed files with 84 additions and 0 deletions

View file

@ -14,6 +14,7 @@ from pathlib import Path
import pytest
import hermes_state
from hermes_cli import kanban_db as kb
@ -886,6 +887,7 @@ def test_connect_falls_back_to_delete_on_locking_protocol(tmp_path, monkeypatch,
# Clear module cache so a fresh connect() is attempted
kb._INITIALIZED_PATHS.clear()
hermes_state._wal_fallback_warned_paths.clear()
real_connect = _sqlite3.connect
@ -921,6 +923,55 @@ def test_connect_falls_back_to_delete_on_locking_protocol(tmp_path, monkeypatch,
conn.close()
def test_connect_works_when_wal_is_silently_refused(tmp_path, monkeypatch, caplog):
"""kanban_db.connect() must stay usable when WAL silently no-ops to DELETE."""
import sqlite3 as _sqlite3
from unittest.mock import patch as _patch
home = tmp_path / ".hermes"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setattr(Path, "home", lambda: tmp_path)
kb._INITIALIZED_PATHS.clear()
hermes_state._wal_fallback_warned_paths.clear()
real_connect = _sqlite3.connect
class _WalSilentNoOpConnection(_sqlite3.Connection):
def execute(self, sql, *args, **kwargs): # type: ignore[override]
if "journal_mode=wal" in sql.lower().replace(" ", ""):
return super().execute("PRAGMA journal_mode=delete", *args, **kwargs)
return super().execute(sql, *args, **kwargs)
def wal_silent_noop_connect(*args, **kwargs):
return real_connect(
*args, factory=_WalSilentNoOpConnection, **kwargs
)
with _patch(
"hermes_cli.kanban_db.sqlite3.connect",
side_effect=wal_silent_noop_connect,
):
with caplog.at_level("ERROR", logger="hermes_state"):
conn = kb.connect()
assert conn.execute("PRAGMA journal_mode").fetchone()[0].lower() == "delete"
t = kb.create_task(conn, title="post-silent-fallback task")
tasks = kb.list_tasks(conn)
assert any(row.id == t for row in tasks)
conn.close()
errors = [
r
for r in caplog.records
if r.levelname == "ERROR" and "kanban.db" in r.getMessage()
]
assert len(errors) >= 1, (
f"Expected a kanban.db ERROR, got: {[r.getMessage() for r in caplog.records]}"
)
def test_unlink_tasks_triggers_recompute_ready(kanban_home):
"""Regression test for issue #22459.

View file

@ -438,3 +438,36 @@ class TestSessionDbUsesWalFallback:
assert get_last_init_error() is None
finally:
db.close()
def test_sessiondb_works_when_wal_is_silently_refused(self, tmp_path, caplog):
"""E2E: SessionDB stays usable when WAL silently no-ops to DELETE."""
target = tmp_path / "macnfs_style.db"
real_connect = sqlite3.connect
factory = _make_silent_noop_factory("delete")
def gated_connect(*args, **kwargs):
return real_connect(str(target), factory=factory, **kwargs)
with patch("hermes_state.sqlite3.connect", side_effect=gated_connect):
with caplog.at_level("ERROR", logger="hermes_state"):
db = SessionDB(db_path=target)
try:
assert db._conn.execute("PRAGMA journal_mode").fetchone()[0].lower() == "delete"
db.create_session(session_id="s2", source="cli", model="test")
sess = db.get_session("s2")
assert sess is not None
assert sess["source"] == "cli"
assert get_last_init_error() is None
finally:
db.close()
errors = [
r
for r in caplog.records
if r.levelname == "ERROR" and "state.db" in r.getMessage()
]
assert len(errors) == 1, (
"silent WAL refusal should emit exactly one state.db error"
)