diff --git a/tests/hermes_cli/test_kanban_db.py b/tests/hermes_cli/test_kanban_db.py index 67152b2bf49..b97ae87964c 100644 --- a/tests/hermes_cli/test_kanban_db.py +++ b/tests/hermes_cli/test_kanban_db.py @@ -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. diff --git a/tests/test_hermes_state_wal_fallback.py b/tests/test_hermes_state_wal_fallback.py index 2f570b82eea..e0a5eda756f 100644 --- a/tests/test_hermes_state_wal_fallback.py +++ b/tests/test_hermes_state_wal_fallback.py @@ -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" + )