hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

test(cli): trim observational-store coverage to two invariants

Browse Source 
- Keep (a) list/stats/pinned open SessionDB(read_only=True) — one parametrized test — and (b) a missing store prints empty results and is never created.
- Drop the insights read-only test (already covered on main), the status test, the mutating-action/live-writer/doctor-isolation tests, and the two doctor factory tests.
- Replace _EmptyObservationalStore + error-string sniffing with a plain '_default_db_path() does not exist' branch printing each action's empty output; the fake-store tests in test_sessions_pin keep working because the branch only runs when the open fails.
This commit is contained in:
kshitijk4poor
2026-09-15 12:02:45 +05:30
committed by kshitij
parent 4a268c9e4a
commit e30c4ed50d
3 changed files with 24 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
hermes_cli/sessions_cmd.py
View File

@@ -964,27 +964,14 @@ _DB_HANDLERS = {
}
class _EmptyObservationalStore:
"""list/stats/pinned on a profile that has never created state.db."""
def __init__(self, db_path: Path):
self.db_path = db_path
def list_sessions_rich(self, **_kwargs):
return []
def session_count(self, source=None):
return 0
def message_count(self):
return 0
def close(self):
return None
def _is_missing_session_store(exc: BaseException) -> bool:
return "unable to open database file" in str(exc).lower()
def _print_empty_store(action: str, args) -> None:
"""A profile that never created state.db: report empty instead of opening a writer that creates it."""
if action == "stats":
print("Total sessions: 0\nTotal messages: 0")
elif action == "pinned":
print("[]" if getattr(args, "json", False) else "No pinned sessions. Pin one with: hermes sessions pin <session_id>")
else:
print("No sessions found.")
def cmd_sessions(args, sessions_parser=None):
@@ -994,15 +981,14 @@ def cmd_sessions(args, sessions_parser=None):
return pre(args)
observational = action in _OBSERVATIONAL_DB_ACTIONS
try:
from hermes_state import SessionDB
from hermes_state import SessionDB, _default_db_path
db = SessionDB(read_only=observational)
except Exception as e:
if observational and _is_missing_session_store(e):
from hermes_state import _default_db_path
db = _EmptyObservationalStore(_default_db_path())
else:
print(f"Error: Could not open session database: {e}")
return 1
# mode=ro cannot create the store; a reader on a fresh profile reports empty rather than failing.
if observational and not _default_db_path().exists():
return _print_empty_store(action, args)
print(f"Error: Could not open session database: {e}")
return 1
try:
handler = _DB_HANDLERS.get(action)
if handler is None:

13
tests/hermes_cli/test_cli_insights_command.py
View File

@@ -1,4 +1,4 @@
from unittest.mock import MagicMock, call, patch
from unittest.mock import MagicMock, patch
from types import SimpleNamespace
import pytest
@@ -94,14 +94,3 @@ def test_subcommand_insights_closes_database_when_generation_fails(capsys):
db.close.assert_called_once()
assert "Error generating insights: boom" in capsys.readouterr().out
def test_insights_paths_open_a_read_only_store(capsys):
cli_obj = HermesCLI.__new__(HermesCLI)
slash_db, command_db = MagicMock(), MagicMock()
with patch("hermes_state.SessionDB", side_effect=[slash_db, command_db]) as factory, \
patch("agent.insights.InsightsEngine", _InsightsEngineStub):
cli_obj._show_insights("/insights")
cmd_insights(SimpleNamespace(days=30, source=None))
assert factory.call_args_list == [call(read_only=True), call(read_only=True)]

99
tests/hermes_cli/test_observational_sessiondb_modes.py
View File

@@ -1,70 +1,22 @@
"""Regression coverage for #110173: observational CLI database readers stay read-only."""
"""Regression coverage for #110173: observational `hermes sessions` readers stay read-only."""
from argparse import Namespace
from unittest.mock import MagicMock, call
from unittest.mock import MagicMock
import pytest
import hermes_cli.sessions_cmd as sessions_cmd
def test_status_session_summary_opens_a_read_only_store(monkeypatch):
from hermes_cli import status
db = MagicMock()
db.list_gateway_sessions.return_value = []
factory = MagicMock(return_value=db)
@pytest.mark.parametrize("action", ["list", "stats", "pinned"])
def test_observational_sessions_actions_open_a_read_only_store(monkeypatch, action):
factory = MagicMock()
monkeypatch.setattr("hermes_state.SessionDB", factory)
monkeypatch.setitem(sessions_cmd._DB_HANDLERS, action, lambda _db, _args: None)
status._render_sessions(Namespace(config={}))
sessions_cmd.cmd_sessions(Namespace(sessions_action=action))
factory.assert_called_once_with(read_only=True)
db.close.assert_called_once()
def test_sessions_list_stats_and_pinned_open_a_read_only_store(monkeypatch):
factory = MagicMock()
monkeypatch.setattr("hermes_state.SessionDB", factory)
monkeypatch.setitem(sessions_cmd._DB_HANDLERS, "list", lambda _db, _args: None)
monkeypatch.setitem(sessions_cmd._DB_HANDLERS, "stats", lambda _db, _args: None)
monkeypatch.setitem(sessions_cmd._DB_HANDLERS, "pinned", lambda _db, _args: None)
sessions_cmd.cmd_sessions(Namespace(sessions_action="list"))
sessions_cmd.cmd_sessions(Namespace(sessions_action="stats"))
sessions_cmd.cmd_sessions(Namespace(sessions_action="pinned"))
assert factory.call_args_list == [call(read_only=True), call(read_only=True), call(read_only=True)]
def test_mutating_sessions_action_keeps_a_writable_store(monkeypatch):
factory = MagicMock()
monkeypatch.setattr("hermes_state.SessionDB", factory)
monkeypatch.setitem(sessions_cmd._DB_HANDLERS, "delete", lambda _db, _args: None)
sessions_cmd.cmd_sessions(Namespace(sessions_action="delete"))
factory.assert_called_once_with(read_only=False)
def test_sessions_stats_reader_does_not_disrupt_a_live_writer(monkeypatch, tmp_path):
"""The actual command reader leaves a writer's WAL generation untouched and usable."""
from hermes_state import SessionDB
import hermes_state
db_path = tmp_path / "state.db"
writer = SessionDB(db_path=db_path)
try:
writer.create_session("live", source="cli")
wal_path = db_path.with_name("state.db-wal")
before = wal_path.read_bytes() if wal_path.exists() else None
monkeypatch.setattr(hermes_state, "_default_db_path", lambda: db_path)
sessions_cmd.cmd_sessions(Namespace(sessions_action="stats"))
after = wal_path.read_bytes() if wal_path.exists() else None
assert after == before
writer.create_session("still-live", source="cli")
assert writer.get_session("still-live") is not None
finally:
writer.close()
def test_sessions_observational_commands_on_missing_store_stay_empty(monkeypatch, tmp_path, capsys):
@@ -87,36 +39,3 @@ def test_sessions_observational_commands_on_missing_store_stay_empty(monkeypatch
assert sessions_cmd.cmd_sessions(pinned_args) is None
assert "No pinned sessions" in capsys.readouterr().out
assert not db_path.exists()
def test_doctor_without_fix_isolates_write_health_from_live_wal(monkeypatch, tmp_path):
from pathlib import Path
from hermes_cli.doctor_report import Finding
from hermes_cli.doctor_state import _state_db_health
from hermes_state import SessionDB
import hermes_state_repair
db_path = tmp_path / "state.db"
writer = SessionDB(db_path=db_path)
probed = []
real = hermes_state_repair._db_opens_cleanly
def _capture(path):
probed.append(Path(path))
return real(path)
try:
writer.create_session("live", source="cli")
wal_path = db_path.with_name("state.db-wal")
before = wal_path.read_bytes() if wal_path.exists() else b""
monkeypatch.setattr(hermes_state_repair, "_db_opens_cleanly", _capture)
_state_db_health(Finding(), False, db_path, "~/hermes")
assert probed and probed[0] != db_path
assert (wal_path.read_bytes() if wal_path.exists() else b"") == before
writer.create_session("still-live", source="cli")
assert writer.get_session("still-live") is not None
finally:
writer.close()