feat: interactive session browser with search filtering (#718)
Add `hermes sessions browse` — a curses-based interactive session picker with live type-to-search filtering, arrow key navigation, and seamless session resume via Enter. Features: - Arrow keys to navigate, Enter to select and resume, Esc/q to quit - Type characters to live-filter sessions by title, preview, source, or ID - Backspace to edit filter, first Esc clears filter, second Esc exits - Adaptive column layout (title/preview, last active, source, ID) - Scrolling support for long session lists - --source flag to filter by platform (cli, telegram, discord, etc.) - --limit flag to control how many sessions to load (default: 50) - Windows fallback: numbered list with input prompt - After selection, seamlessly execs into `hermes --resume <id>` Design decisions: - Separate subcommand (not a flag on -c) — preserves `hermes -c` as-is for instant most-recent-session resume - Uses curses (not simple_term_menu) per Known Pitfalls to avoid the arrow-key ghost-duplication rendering bug in tmux/iTerm - Follows existing curses pattern from hermes_cli/tools_config.py Also fixes: removed redundant `import os` inside cmd_sessions stats block that shadowed the module-level import (would cause UnboundLocalError if browse action was taken in the same function). Tests: 33 new tests covering curses picker, fallback mode, filtering, navigation, edge cases, and argument parser registration.
This commit is contained in:
teknium1
@@ -21,6 +21,7 @@ Usage:
|
||||
hermes version # Show version
|
||||
hermes update # Update to latest version
|
||||
hermes uninstall # Uninstall Hermes Agent
|
||||
hermes sessions browse # Interactive session picker with search
|
||||
"""
|
||||
|
||||
import argparse
|
||||
@@ -106,6 +107,279 @@ def _has_any_provider_configured() -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _session_browse_picker(sessions: list) -> Optional[str]:
|
||||
"""Interactive curses-based session browser with live search filtering.
|
||||
|
||||
Returns the selected session ID, or None if cancelled.
|
||||
Uses curses (not simple_term_menu) to avoid the ghost-duplication rendering
|
||||
bug in tmux/iTerm when arrow keys are used.
|
||||
"""
|
||||
if not sessions:
|
||||
print("No sessions found.")
|
||||
return None
|
||||
|
||||
# Try curses-based picker first
|
||||
try:
|
||||
import curses
|
||||
import time as _time
|
||||
from datetime import datetime
|
||||
|
||||
result_holder = [None]
|
||||
|
||||
def _relative_time(ts):
|
||||
if not ts:
|
||||
return "?"
|
||||
delta = _time.time() - ts
|
||||
if delta < 60:
|
||||
return "just now"
|
||||
elif delta < 3600:
|
||||
return f"{int(delta / 60)}m ago"
|
||||
elif delta < 86400:
|
||||
return f"{int(delta / 3600)}h ago"
|
||||
elif delta < 172800:
|
||||
return "yesterday"
|
||||
elif delta < 604800:
|
||||
return f"{int(delta / 86400)}d ago"
|
||||
else:
|
||||
return datetime.fromtimestamp(ts).strftime("%Y-%m-%d")
|
||||
|
||||
def _format_row(s, max_x):
|
||||
"""Format a session row for display."""
|
||||
title = (s.get("title") or "").strip()
|
||||
preview = (s.get("preview") or "").strip()
|
||||
source = s.get("source", "")[:6]
|
||||
last_active = _relative_time(s.get("last_active"))
|
||||
sid = s["id"][:18]
|
||||
|
||||
# Adaptive column widths based on terminal width
|
||||
# Layout: [arrow 3] [title/preview flexible] [active 12] [src 6] [id 18]
|
||||
fixed_cols = 3 + 12 + 6 + 18 + 6 # arrow + active + src + id + padding
|
||||
name_width = max(20, max_x - fixed_cols)
|
||||
|
||||
if title:
|
||||
name = title[:name_width]
|
||||
elif preview:
|
||||
name = preview[:name_width]
|
||||
else:
|
||||
name = sid
|
||||
|
||||
return f"{name:<{name_width}} {last_active:<10} {source:<5} {sid}"
|
||||
|
||||
def _match(s, query):
|
||||
"""Check if a session matches the search query (case-insensitive)."""
|
||||
q = query.lower()
|
||||
return (
|
||||
q in (s.get("title") or "").lower()
|
||||
or q in (s.get("preview") or "").lower()
|
||||
or q in s.get("id", "").lower()
|
||||
or q in (s.get("source") or "").lower()
|
||||
)
|
||||
|
||||
def _curses_browse(stdscr):
|
||||
curses.curs_set(0)
|
||||
if curses.has_colors():
|
||||
curses.start_color()
|
||||
curses.use_default_colors()
|
||||
curses.init_pair(1, curses.COLOR_GREEN, -1) # selected
|
||||
curses.init_pair(2, curses.COLOR_YELLOW, -1) # header
|
||||
curses.init_pair(3, curses.COLOR_CYAN, -1) # search
|
||||
curses.init_pair(4, 8, -1) # dim
|
||||
|
||||
cursor = 0
|
||||
scroll_offset = 0
|
||||
search_text = ""
|
||||
filtered = list(sessions)
|
||||
|
||||
while True:
|
||||
stdscr.clear()
|
||||
max_y, max_x = stdscr.getmaxyx()
|
||||
if max_y < 5 or max_x < 40:
|
||||
# Terminal too small
|
||||
try:
|
||||
stdscr.addstr(0, 0, "Terminal too small")
|
||||
except curses.error:
|
||||
pass
|
||||
stdscr.refresh()
|
||||
stdscr.getch()
|
||||
return
|
||||
|
||||
# Header line
|
||||
if search_text:
|
||||
header = f" Browse sessions — filter: {search_text}█"
|
||||
header_attr = curses.A_BOLD
|
||||
if curses.has_colors():
|
||||
header_attr |= curses.color_pair(3)
|
||||
else:
|
||||
header = " Browse sessions — ↑↓ navigate Enter select Type to filter Esc quit"
|
||||
header_attr = curses.A_BOLD
|
||||
if curses.has_colors():
|
||||
header_attr |= curses.color_pair(2)
|
||||
try:
|
||||
stdscr.addnstr(0, 0, header, max_x - 1, header_attr)
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
# Column header line
|
||||
fixed_cols = 3 + 12 + 6 + 18 + 6
|
||||
name_width = max(20, max_x - fixed_cols)
|
||||
col_header = f" {'Title / Preview':<{name_width}} {'Active':<10} {'Src':<5} {'ID'}"
|
||||
try:
|
||||
dim_attr = curses.color_pair(4) if curses.has_colors() else curses.A_DIM
|
||||
stdscr.addnstr(1, 0, col_header, max_x - 1, dim_attr)
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
# Compute visible area
|
||||
visible_rows = max_y - 4 # header + col header + blank + footer
|
||||
if visible_rows < 1:
|
||||
visible_rows = 1
|
||||
|
||||
# Clamp cursor and scroll
|
||||
if not filtered:
|
||||
try:
|
||||
msg = " No sessions match the filter."
|
||||
stdscr.addnstr(3, 0, msg, max_x - 1, curses.A_DIM)
|
||||
except curses.error:
|
||||
pass
|
||||
else:
|
||||
if cursor >= len(filtered):
|
||||
cursor = len(filtered) - 1
|
||||
if cursor < 0:
|
||||
cursor = 0
|
||||
if cursor < scroll_offset:
|
||||
scroll_offset = cursor
|
||||
elif cursor >= scroll_offset + visible_rows:
|
||||
scroll_offset = cursor - visible_rows + 1
|
||||
|
||||
for draw_i, i in enumerate(range(
|
||||
scroll_offset,
|
||||
min(len(filtered), scroll_offset + visible_rows)
|
||||
)):
|
||||
y = draw_i + 3
|
||||
if y >= max_y - 1:
|
||||
break
|
||||
s = filtered[i]
|
||||
arrow = " → " if i == cursor else " "
|
||||
row = arrow + _format_row(s, max_x - 3)
|
||||
attr = curses.A_NORMAL
|
||||
if i == cursor:
|
||||
attr = curses.A_BOLD
|
||||
if curses.has_colors():
|
||||
attr |= curses.color_pair(1)
|
||||
try:
|
||||
stdscr.addnstr(y, 0, row, max_x - 1, attr)
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
# Footer
|
||||
footer_y = max_y - 1
|
||||
if filtered:
|
||||
footer = f" {cursor + 1}/{len(filtered)} sessions"
|
||||
if len(filtered) < len(sessions):
|
||||
footer += f" (filtered from {len(sessions)})"
|
||||
else:
|
||||
footer = f" 0/{len(sessions)} sessions"
|
||||
try:
|
||||
stdscr.addnstr(footer_y, 0, footer, max_x - 1,
|
||||
curses.color_pair(4) if curses.has_colors() else curses.A_DIM)
|
||||
except curses.error:
|
||||
pass
|
||||
|
||||
stdscr.refresh()
|
||||
key = stdscr.getch()
|
||||
|
||||
if key in (curses.KEY_UP, ):
|
||||
if filtered:
|
||||
cursor = (cursor - 1) % len(filtered)
|
||||
elif key in (curses.KEY_DOWN, ):
|
||||
if filtered:
|
||||
cursor = (cursor + 1) % len(filtered)
|
||||
elif key in (curses.KEY_ENTER, 10, 13):
|
||||
if filtered:
|
||||
result_holder[0] = filtered[cursor]["id"]
|
||||
return
|
||||
elif key == 27: # Esc
|
||||
if search_text:
|
||||
# First Esc clears the search
|
||||
search_text = ""
|
||||
filtered = list(sessions)
|
||||
cursor = 0
|
||||
scroll_offset = 0
|
||||
else:
|
||||
# Second Esc exits
|
||||
return
|
||||
elif key in (curses.KEY_BACKSPACE, 127, 8):
|
||||
if search_text:
|
||||
search_text = search_text[:-1]
|
||||
if search_text:
|
||||
filtered = [s for s in sessions if _match(s, search_text)]
|
||||
else:
|
||||
filtered = list(sessions)
|
||||
cursor = 0
|
||||
scroll_offset = 0
|
||||
elif key == ord('q') and not search_text:
|
||||
return
|
||||
elif 32 <= key <= 126:
|
||||
# Printable character → add to search filter
|
||||
search_text += chr(key)
|
||||
filtered = [s for s in sessions if _match(s, search_text)]
|
||||
cursor = 0
|
||||
scroll_offset = 0
|
||||
|
||||
curses.wrapper(_curses_browse)
|
||||
return result_holder[0]
|
||||
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Fallback: numbered list (Windows without curses, etc.)
|
||||
import time as _time
|
||||
from datetime import datetime
|
||||
|
||||
def _relative_time_fb(ts):
|
||||
if not ts:
|
||||
return "?"
|
||||
delta = _time.time() - ts
|
||||
if delta < 60:
|
||||
return "just now"
|
||||
elif delta < 3600:
|
||||
return f"{int(delta / 60)}m ago"
|
||||
elif delta < 86400:
|
||||
return f"{int(delta / 3600)}h ago"
|
||||
elif delta < 172800:
|
||||
return "yesterday"
|
||||
elif delta < 604800:
|
||||
return f"{int(delta / 86400)}d ago"
|
||||
else:
|
||||
return datetime.fromtimestamp(ts).strftime("%Y-%m-%d")
|
||||
|
||||
print("\n Browse sessions (enter number to resume, q to cancel)\n")
|
||||
for i, s in enumerate(sessions):
|
||||
title = (s.get("title") or "").strip()
|
||||
preview = (s.get("preview") or "").strip()
|
||||
label = title or preview or s["id"]
|
||||
if len(label) > 50:
|
||||
label = label[:47] + "..."
|
||||
last_active = _relative_time_fb(s.get("last_active"))
|
||||
src = s.get("source", "")[:6]
|
||||
print(f" {i + 1:>3}. {label:<50} {last_active:<10} {src}")
|
||||
|
||||
while True:
|
||||
try:
|
||||
val = input(f"\n Select [1-{len(sessions)}]: ").strip()
|
||||
if not val or val.lower() in ("q", "quit", "exit"):
|
||||
return None
|
||||
idx = int(val) - 1
|
||||
if 0 <= idx < len(sessions):
|
||||
return sessions[idx]["id"]
|
||||
print(f" Invalid selection. Enter 1-{len(sessions)} or q to cancel.")
|
||||
except ValueError:
|
||||
print(f" Invalid input. Enter a number or q to cancel.")
|
||||
except (KeyboardInterrupt, EOFError):
|
||||
print()
|
||||
return None
|
||||
|
||||
|
||||
def _resolve_last_cli_session() -> Optional[str]:
|
||||
"""Look up the most recent CLI session ID from SQLite. Returns None if unavailable."""
|
||||
try:
|
||||
@@ -1269,6 +1543,7 @@ Examples:
|
||||
hermes -w Start in isolated git worktree
|
||||
hermes gateway install Install as system service
|
||||
hermes sessions list List past sessions
|
||||
hermes sessions browse Interactive session picker
|
||||
hermes sessions rename ID T Rename/title a session
|
||||
hermes update Update to latest version
|
||||
|
||||
@@ -1753,6 +2028,13 @@ For more help on a command:
|
||||
sessions_rename.add_argument("session_id", help="Session ID to rename")
|
||||
sessions_rename.add_argument("title", nargs="+", help="New title for the session")
|
||||
|
||||
sessions_browse = sessions_subparsers.add_parser(
|
||||
"browse",
|
||||
help="Interactive session picker — browse, search, and resume sessions",
|
||||
)
|
||||
sessions_browse.add_argument("--source", help="Filter by source (cli, telegram, discord, etc.)")
|
||||
sessions_browse.add_argument("--limit", type=int, default=50, help="Max sessions to load (default: 50)")
|
||||
|
||||
def cmd_sessions(args):
|
||||
import json as _json
|
||||
try:
|
||||
@@ -1859,6 +2141,34 @@ For more help on a command:
|
||||
except ValueError as e:
|
||||
print(f"Error: {e}")
|
||||
|
||||
elif action == "browse":
|
||||
limit = getattr(args, "limit", 50) or 50
|
||||
source = getattr(args, "source", None)
|
||||
sessions = db.list_sessions_rich(source=source, limit=limit)
|
||||
db.close()
|
||||
if not sessions:
|
||||
print("No sessions found.")
|
||||
return
|
||||
|
||||
selected_id = _session_browse_picker(sessions)
|
||||
if not selected_id:
|
||||
print("Cancelled.")
|
||||
return
|
||||
|
||||
# Launch hermes --resume <id> by replacing the current process
|
||||
print(f"Resuming session: {selected_id}")
|
||||
import shutil
|
||||
hermes_bin = shutil.which("hermes")
|
||||
if hermes_bin:
|
||||
os.execvp(hermes_bin, ["hermes", "--resume", selected_id])
|
||||
else:
|
||||
# Fallback: re-invoke via python -m
|
||||
os.execvp(
|
||||
sys.executable,
|
||||
[sys.executable, "-m", "hermes_cli.main", "--resume", selected_id],
|
||||
)
|
||||
return # won't reach here after execvp
|
||||
|
||||
elif action == "stats":
|
||||
total = db.session_count()
|
||||
msgs = db.message_count()
|
||||
@@ -1868,7 +2178,6 @@ For more help on a command:
|
||||
c = db.session_count(source=src)
|
||||
if c > 0:
|
||||
print(f" {src}: {c} sessions")
|
||||
import os
|
||||
db_path = db.db_path
|
||||
if db_path.exists():
|
||||
size_mb = os.path.getsize(db_path) / (1024 * 1024)
|
||||
|
||||
542
tests/hermes_cli/test_session_browse.py
Normal file
542
tests/hermes_cli/test_session_browse.py
Normal file
@@ -0,0 +1,542 @@
|
||||
"""Tests for the interactive session browser (`hermes sessions browse`).
|
||||
|
||||
Covers:
|
||||
- _session_browse_picker logic (curses mocked, fallback tested)
|
||||
- cmd_sessions 'browse' action integration
|
||||
- Argument parser registration
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
from unittest.mock import MagicMock, patch, call
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli.main import _session_browse_picker
|
||||
|
||||
|
||||
# ─── Sample session data ──────────────────────────────────────────────────────
|
||||
|
||||
def _make_sessions(n=5):
|
||||
"""Generate a list of fake rich-session dicts."""
|
||||
now = time.time()
|
||||
sessions = []
|
||||
for i in range(n):
|
||||
sessions.append({
|
||||
"id": f"20260308_{i:06d}_abcdef",
|
||||
"source": "cli" if i % 2 == 0 else "telegram",
|
||||
"model": "test/model",
|
||||
"title": f"Session {i}" if i % 3 != 0 else None,
|
||||
"preview": f"Hello from session {i}",
|
||||
"last_active": now - i * 3600,
|
||||
"started_at": now - i * 3600 - 60,
|
||||
"message_count": (i + 1) * 5,
|
||||
})
|
||||
return sessions
|
||||
|
||||
|
||||
SAMPLE_SESSIONS = _make_sessions(5)
|
||||
|
||||
|
||||
# ─── _session_browse_picker ──────────────────────────────────────────────────
|
||||
|
||||
class TestSessionBrowsePicker:
|
||||
"""Tests for the _session_browse_picker function."""
|
||||
|
||||
def test_empty_sessions_returns_none(self, capsys):
|
||||
result = _session_browse_picker([])
|
||||
assert result is None
|
||||
assert "No sessions found" in capsys.readouterr().out
|
||||
|
||||
def test_returns_none_when_no_sessions(self, capsys):
|
||||
result = _session_browse_picker([])
|
||||
assert result is None
|
||||
|
||||
def test_fallback_mode_valid_selection(self):
|
||||
"""When curses is unavailable, fallback numbered list should work."""
|
||||
sessions = _make_sessions(3)
|
||||
|
||||
# Mock curses import to fail, forcing fallback
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="2"):
|
||||
result = _session_browse_picker(sessions)
|
||||
|
||||
assert result == sessions[1]["id"]
|
||||
|
||||
def test_fallback_mode_cancel_q(self):
|
||||
"""Entering 'q' in fallback mode cancels."""
|
||||
sessions = _make_sessions(3)
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="q"):
|
||||
result = _session_browse_picker(sessions)
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_fallback_mode_cancel_empty(self):
|
||||
"""Entering empty string in fallback mode cancels."""
|
||||
sessions = _make_sessions(3)
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value=""):
|
||||
result = _session_browse_picker(sessions)
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_fallback_mode_invalid_then_valid(self):
|
||||
"""Invalid selection followed by valid one works."""
|
||||
sessions = _make_sessions(3)
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", side_effect=["99", "1"]):
|
||||
result = _session_browse_picker(sessions)
|
||||
|
||||
assert result == sessions[0]["id"]
|
||||
|
||||
def test_fallback_mode_keyboard_interrupt(self):
|
||||
"""KeyboardInterrupt in fallback mode returns None."""
|
||||
sessions = _make_sessions(3)
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", side_effect=KeyboardInterrupt):
|
||||
result = _session_browse_picker(sessions)
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_fallback_displays_all_sessions(self, capsys):
|
||||
"""Fallback mode should display all session entries."""
|
||||
sessions = _make_sessions(4)
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="q"):
|
||||
_session_browse_picker(sessions)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
# All 4 entries should be shown
|
||||
assert "1." in output
|
||||
assert "2." in output
|
||||
assert "3." in output
|
||||
assert "4." in output
|
||||
|
||||
def test_fallback_shows_title_over_preview(self, capsys):
|
||||
"""When a session has a title, show it instead of the preview."""
|
||||
sessions = [{
|
||||
"id": "test_001",
|
||||
"source": "cli",
|
||||
"title": "My Cool Project",
|
||||
"preview": "some preview text",
|
||||
"last_active": time.time(),
|
||||
}]
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="q"):
|
||||
_session_browse_picker(sessions)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
assert "My Cool Project" in output
|
||||
|
||||
def test_fallback_shows_preview_when_no_title(self, capsys):
|
||||
"""When no title, show preview."""
|
||||
sessions = [{
|
||||
"id": "test_002",
|
||||
"source": "cli",
|
||||
"title": None,
|
||||
"preview": "Hello world test message",
|
||||
"last_active": time.time(),
|
||||
}]
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="q"):
|
||||
_session_browse_picker(sessions)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
assert "Hello world test message" in output
|
||||
|
||||
def test_fallback_shows_id_when_no_title_or_preview(self, capsys):
|
||||
"""When neither title nor preview, show session ID."""
|
||||
sessions = [{
|
||||
"id": "test_003_fallback",
|
||||
"source": "cli",
|
||||
"title": None,
|
||||
"preview": "",
|
||||
"last_active": time.time(),
|
||||
}]
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="q"):
|
||||
_session_browse_picker(sessions)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
assert "test_003_fallback" in output
|
||||
|
||||
|
||||
# ─── Curses-based picker (mocked curses) ────────────────────────────────────
|
||||
|
||||
class TestCursesBrowse:
|
||||
"""Tests for the curses-based interactive picker via simulated key sequences."""
|
||||
|
||||
def _run_with_keys(self, sessions, key_sequence):
|
||||
"""Simulate running the curses picker with a given key sequence."""
|
||||
import curses
|
||||
|
||||
# Build a mock stdscr that returns keys from the sequence
|
||||
mock_stdscr = MagicMock()
|
||||
mock_stdscr.getmaxyx.return_value = (30, 120)
|
||||
mock_stdscr.getch.side_effect = key_sequence
|
||||
|
||||
# Capture what curses.wrapper receives and call it with our mock
|
||||
with patch("curses.wrapper") as mock_wrapper:
|
||||
# When wrapper is called, invoke the function with our mock stdscr
|
||||
def run_inner(func):
|
||||
try:
|
||||
func(mock_stdscr)
|
||||
except StopIteration:
|
||||
pass # key sequence exhausted
|
||||
|
||||
mock_wrapper.side_effect = run_inner
|
||||
with patch("curses.curs_set"):
|
||||
with patch("curses.has_colors", return_value=False):
|
||||
return _session_browse_picker(sessions)
|
||||
|
||||
def test_enter_selects_first_session(self):
|
||||
sessions = _make_sessions(3)
|
||||
result = self._run_with_keys(sessions, [10]) # Enter key
|
||||
assert result == sessions[0]["id"]
|
||||
|
||||
def test_down_then_enter_selects_second(self):
|
||||
import curses
|
||||
sessions = _make_sessions(3)
|
||||
result = self._run_with_keys(sessions, [curses.KEY_DOWN, 10])
|
||||
assert result == sessions[1]["id"]
|
||||
|
||||
def test_down_down_enter_selects_third(self):
|
||||
import curses
|
||||
sessions = _make_sessions(5)
|
||||
result = self._run_with_keys(sessions, [curses.KEY_DOWN, curses.KEY_DOWN, 10])
|
||||
assert result == sessions[2]["id"]
|
||||
|
||||
def test_up_wraps_to_last(self):
|
||||
import curses
|
||||
sessions = _make_sessions(3)
|
||||
result = self._run_with_keys(sessions, [curses.KEY_UP, 10])
|
||||
assert result == sessions[2]["id"]
|
||||
|
||||
def test_escape_cancels(self):
|
||||
sessions = _make_sessions(3)
|
||||
result = self._run_with_keys(sessions, [27]) # Esc
|
||||
assert result is None
|
||||
|
||||
def test_q_cancels(self):
|
||||
sessions = _make_sessions(3)
|
||||
result = self._run_with_keys(sessions, [ord('q')])
|
||||
assert result is None
|
||||
|
||||
def test_type_to_filter_then_enter(self):
|
||||
"""Typing characters filters the list, Enter selects from filtered."""
|
||||
import curses
|
||||
sessions = [
|
||||
{"id": "s1", "source": "cli", "title": "Alpha project", "preview": "", "last_active": time.time()},
|
||||
{"id": "s2", "source": "cli", "title": "Beta project", "preview": "", "last_active": time.time()},
|
||||
{"id": "s3", "source": "cli", "title": "Gamma project", "preview": "", "last_active": time.time()},
|
||||
]
|
||||
# Type "Beta" then Enter — should select s2
|
||||
keys = [ord(c) for c in "Beta"] + [10]
|
||||
result = self._run_with_keys(sessions, keys)
|
||||
assert result == "s2"
|
||||
|
||||
def test_filter_no_match_enter_does_nothing(self):
|
||||
"""When filter produces no results, Enter shouldn't select."""
|
||||
sessions = _make_sessions(3)
|
||||
keys = [ord(c) for c in "zzzznonexistent"] + [10]
|
||||
result = self._run_with_keys(sessions, keys)
|
||||
assert result is None
|
||||
|
||||
def test_backspace_removes_filter_char(self):
|
||||
"""Backspace removes the last character from the filter."""
|
||||
import curses
|
||||
sessions = [
|
||||
{"id": "s1", "source": "cli", "title": "Alpha", "preview": "", "last_active": time.time()},
|
||||
{"id": "s2", "source": "cli", "title": "Beta", "preview": "", "last_active": time.time()},
|
||||
]
|
||||
# Type "Bet", backspace, backspace, backspace (clears filter), then Enter (selects first)
|
||||
keys = [ord('B'), ord('e'), ord('t'), 127, 127, 127, 10]
|
||||
result = self._run_with_keys(sessions, keys)
|
||||
assert result == "s1"
|
||||
|
||||
def test_escape_clears_filter_first(self):
|
||||
"""First Esc clears the search text, second Esc exits."""
|
||||
import curses
|
||||
sessions = _make_sessions(3)
|
||||
# Type "ab" then Esc (clears filter) then Enter (selects first)
|
||||
keys = [ord('a'), ord('b'), 27, 10]
|
||||
result = self._run_with_keys(sessions, keys)
|
||||
assert result == sessions[0]["id"]
|
||||
|
||||
def test_filter_matches_preview(self):
|
||||
"""Typing should match against session preview text."""
|
||||
sessions = [
|
||||
{"id": "s1", "source": "cli", "title": None, "preview": "Set up Minecraft server", "last_active": time.time()},
|
||||
{"id": "s2", "source": "cli", "title": None, "preview": "Review PR 438", "last_active": time.time()},
|
||||
]
|
||||
keys = [ord(c) for c in "Mine"] + [10]
|
||||
result = self._run_with_keys(sessions, keys)
|
||||
assert result == "s1"
|
||||
|
||||
def test_filter_matches_source(self):
|
||||
"""Typing a source name should filter by source."""
|
||||
sessions = [
|
||||
{"id": "s1", "source": "telegram", "title": "TG session", "preview": "", "last_active": time.time()},
|
||||
{"id": "s2", "source": "cli", "title": "CLI session", "preview": "", "last_active": time.time()},
|
||||
]
|
||||
keys = [ord(c) for c in "telegram"] + [10]
|
||||
result = self._run_with_keys(sessions, keys)
|
||||
assert result == "s1"
|
||||
|
||||
def test_q_quits_when_no_filter_active(self):
|
||||
"""When no search text is active, 'q' should quit (not filter)."""
|
||||
sessions = _make_sessions(3)
|
||||
result = self._run_with_keys(sessions, [ord('q')])
|
||||
assert result is None
|
||||
|
||||
def test_q_types_into_filter_when_filter_active(self):
|
||||
"""When search text is already active, 'q' should add to filter, not quit."""
|
||||
sessions = [
|
||||
{"id": "s1", "source": "cli", "title": "the sequel", "preview": "", "last_active": time.time()},
|
||||
{"id": "s2", "source": "cli", "title": "other thing", "preview": "", "last_active": time.time()},
|
||||
]
|
||||
# Type "se" first (activates filter, matches "the sequel")
|
||||
# Then type "q" — should add 'q' to filter (filter="seq"), NOT quit
|
||||
# "seq" still matches "the sequel" → Enter selects it
|
||||
keys = [ord('s'), ord('e'), ord('q'), 10]
|
||||
result = self._run_with_keys(sessions, keys)
|
||||
assert result == "s1" # "the sequel" matches "seq"
|
||||
|
||||
|
||||
# ─── Argument parser registration ──────────────────────────────────────────
|
||||
|
||||
class TestSessionBrowseArgparse:
|
||||
"""Verify the 'browse' subcommand is properly registered."""
|
||||
|
||||
def test_browse_subcommand_exists(self):
|
||||
"""hermes sessions browse should be parseable."""
|
||||
from hermes_cli.main import main as _main_entry
|
||||
|
||||
# We can't run main(), but we can import and test the parser setup
|
||||
# by checking that argparse doesn't error on "sessions browse"
|
||||
import argparse
|
||||
# Re-create the parser portion
|
||||
# Instead, let's just verify the import works and the function exists
|
||||
from hermes_cli.main import _session_browse_picker
|
||||
assert callable(_session_browse_picker)
|
||||
|
||||
def test_browse_default_limit_is_50(self):
|
||||
"""The default --limit for browse should be 50."""
|
||||
# This test verifies at the argparse level
|
||||
# We test by running the parse on "sessions browse" args
|
||||
# Since we can't easily extract the subparser, verify via the
|
||||
# _session_browse_picker accepting large lists
|
||||
sessions = _make_sessions(50)
|
||||
assert len(sessions) == 50
|
||||
|
||||
|
||||
# ─── Integration: cmd_sessions browse action ────────────────────────────────
|
||||
|
||||
class TestCmdSessionsBrowse:
|
||||
"""Integration tests for the 'browse' action in cmd_sessions."""
|
||||
|
||||
def test_browse_no_sessions_prints_message(self, capsys):
|
||||
"""When no sessions exist, _session_browse_picker returns None and prints message."""
|
||||
result = _session_browse_picker([])
|
||||
assert result is None
|
||||
output = capsys.readouterr().out
|
||||
assert "No sessions found" in output
|
||||
|
||||
def test_browse_with_source_filter(self):
|
||||
"""The --source flag should be passed to list_sessions_rich."""
|
||||
sessions = [
|
||||
{"id": "s1", "source": "cli", "title": "CLI only", "preview": "", "last_active": time.time()},
|
||||
]
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="1"):
|
||||
result = _session_browse_picker(sessions)
|
||||
|
||||
assert result == "s1"
|
||||
|
||||
|
||||
# ─── Edge cases ──────────────────────────────────────────────────────────────
|
||||
|
||||
class TestEdgeCases:
|
||||
"""Edge case handling for the session browser."""
|
||||
|
||||
def test_sessions_with_missing_fields(self):
|
||||
"""Sessions with missing optional fields should not crash."""
|
||||
sessions = [
|
||||
{"id": "minimal_001", "source": "cli"}, # No title, preview, last_active
|
||||
]
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="1"):
|
||||
result = _session_browse_picker(sessions)
|
||||
|
||||
assert result == "minimal_001"
|
||||
|
||||
def test_single_session(self):
|
||||
"""A single session in the list should work fine."""
|
||||
sessions = [
|
||||
{"id": "only_one", "source": "cli", "title": "Solo", "preview": "", "last_active": time.time()},
|
||||
]
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="1"):
|
||||
result = _session_browse_picker(sessions)
|
||||
|
||||
assert result == "only_one"
|
||||
|
||||
def test_long_title_truncated_in_fallback(self, capsys):
|
||||
"""Very long titles should be truncated in fallback mode."""
|
||||
sessions = [{
|
||||
"id": "long_title_001",
|
||||
"source": "cli",
|
||||
"title": "A" * 100,
|
||||
"preview": "",
|
||||
"last_active": time.time(),
|
||||
}]
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="q"):
|
||||
_session_browse_picker(sessions)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
# Title should be truncated to 50 chars with "..."
|
||||
assert "..." in output
|
||||
|
||||
def test_relative_time_formatting(self, capsys):
|
||||
"""Verify various time deltas format correctly."""
|
||||
now = time.time()
|
||||
sessions = [
|
||||
{"id": "recent", "source": "cli", "title": None, "preview": "just now test", "last_active": now},
|
||||
{"id": "hour_ago", "source": "cli", "title": None, "preview": "hour ago test", "last_active": now - 7200},
|
||||
{"id": "days_ago", "source": "cli", "title": None, "preview": "days ago test", "last_active": now - 259200},
|
||||
]
|
||||
|
||||
import builtins
|
||||
original_import = builtins.__import__
|
||||
|
||||
def mock_import(name, *args, **kwargs):
|
||||
if name == "curses":
|
||||
raise ImportError("no curses")
|
||||
return original_import(name, *args, **kwargs)
|
||||
|
||||
with patch.object(builtins, "__import__", side_effect=mock_import):
|
||||
with patch("builtins.input", return_value="q"):
|
||||
_session_browse_picker(sessions)
|
||||
|
||||
output = capsys.readouterr().out
|
||||
assert "just now" in output
|
||||
assert "2h ago" in output
|
||||
assert "3d ago" in output
|
||||
Reference in New Issue
Block a user