test_title_command.py
1 """Tests for /title gateway slash command. 2 3 Tests the _handle_title_command handler (set/show session titles) 4 across all gateway messenger platforms. 5 """ 6 7 import os 8 from types import SimpleNamespace 9 from unittest.mock import AsyncMock, MagicMock, patch 10 11 import pytest 12 13 from gateway.config import GatewayConfig, Platform, PlatformConfig 14 from gateway.platforms.base import MessageEvent 15 from gateway.session import SessionSource 16 17 18 def _make_event(text="/title", platform=Platform.TELEGRAM, 19 user_id="12345", chat_id="67890"): 20 """Build a MessageEvent for testing.""" 21 source = SessionSource( 22 platform=platform, 23 user_id=user_id, 24 chat_id=chat_id, 25 user_name="testuser", 26 ) 27 return MessageEvent(text=text, source=source) 28 29 30 def _make_runner(session_db=None): 31 """Create a bare GatewayRunner with a mock session_store and optional session_db.""" 32 from gateway.run import GatewayRunner 33 runner = object.__new__(GatewayRunner) 34 runner.adapters = {} 35 runner._voice_mode = {} 36 runner._session_db = session_db 37 38 # Mock session_store that returns a session entry with a known session_id 39 mock_session_entry = MagicMock() 40 mock_session_entry.session_id = "test_session_123" 41 mock_session_entry.session_key = "telegram:12345:67890" 42 mock_store = MagicMock() 43 mock_store.get_or_create_session.return_value = mock_session_entry 44 runner.session_store = mock_store 45 46 return runner 47 48 49 # --------------------------------------------------------------------------- 50 # _handle_title_command 51 # --------------------------------------------------------------------------- 52 53 54 class TestHandleTitleCommand: 55 """Tests for GatewayRunner._handle_title_command.""" 56 57 @pytest.mark.asyncio 58 async def test_set_title(self, tmp_path): 59 """Setting a title returns confirmation.""" 60 from hermes_state import SessionDB 61 db = SessionDB(db_path=tmp_path / "state.db") 62 db.create_session("test_session_123", "telegram") 63 64 runner = _make_runner(session_db=db) 65 event = _make_event(text="/title My Research Project") 66 result = await runner._handle_title_command(event) 67 assert "My Research Project" in result 68 assert "✏️" in result 69 70 # Verify in DB 71 assert db.get_session_title("test_session_123") == "My Research Project" 72 db.close() 73 74 @pytest.mark.asyncio 75 async def test_show_title_when_set(self, tmp_path): 76 """Showing title when one is set returns the title.""" 77 from hermes_state import SessionDB 78 db = SessionDB(db_path=tmp_path / "state.db") 79 db.create_session("test_session_123", "telegram") 80 db.set_session_title("test_session_123", "Existing Title") 81 82 runner = _make_runner(session_db=db) 83 event = _make_event(text="/title") 84 result = await runner._handle_title_command(event) 85 assert "Existing Title" in result 86 assert "📌" in result 87 db.close() 88 89 @pytest.mark.asyncio 90 async def test_show_title_when_not_set(self, tmp_path): 91 """Showing title when none is set returns usage hint.""" 92 from hermes_state import SessionDB 93 db = SessionDB(db_path=tmp_path / "state.db") 94 db.create_session("test_session_123", "telegram") 95 96 runner = _make_runner(session_db=db) 97 event = _make_event(text="/title") 98 result = await runner._handle_title_command(event) 99 assert "No title set" in result 100 assert "/title" in result 101 db.close() 102 103 @pytest.mark.asyncio 104 async def test_title_conflict(self, tmp_path): 105 """Setting a title already used by another session returns error.""" 106 from hermes_state import SessionDB 107 db = SessionDB(db_path=tmp_path / "state.db") 108 db.create_session("other_session", "telegram") 109 db.set_session_title("other_session", "Taken Title") 110 db.create_session("test_session_123", "telegram") 111 112 runner = _make_runner(session_db=db) 113 event = _make_event(text="/title Taken Title") 114 result = await runner._handle_title_command(event) 115 assert "already in use" in result 116 assert "⚠️" in result 117 db.close() 118 119 @pytest.mark.asyncio 120 async def test_no_session_db(self): 121 """Returns error when session database is not available.""" 122 runner = _make_runner(session_db=None) 123 event = _make_event(text="/title My Title") 124 result = await runner._handle_title_command(event) 125 assert "not available" in result 126 127 @pytest.mark.asyncio 128 async def test_title_too_long(self, tmp_path): 129 """Setting a title that exceeds max length returns error.""" 130 from hermes_state import SessionDB 131 db = SessionDB(db_path=tmp_path / "state.db") 132 db.create_session("test_session_123", "telegram") 133 134 runner = _make_runner(session_db=db) 135 long_title = "A" * 150 136 event = _make_event(text=f"/title {long_title}") 137 result = await runner._handle_title_command(event) 138 assert "too long" in result 139 assert "⚠️" in result 140 db.close() 141 142 @pytest.mark.asyncio 143 async def test_title_control_chars_sanitized(self, tmp_path): 144 """Control characters are stripped and sanitized title is stored.""" 145 from hermes_state import SessionDB 146 db = SessionDB(db_path=tmp_path / "state.db") 147 db.create_session("test_session_123", "telegram") 148 149 runner = _make_runner(session_db=db) 150 event = _make_event(text="/title hello\x00world") 151 result = await runner._handle_title_command(event) 152 assert "helloworld" in result 153 assert db.get_session_title("test_session_123") == "helloworld" 154 db.close() 155 156 @pytest.mark.asyncio 157 async def test_title_only_control_chars(self, tmp_path): 158 """Title with only control chars returns empty error.""" 159 from hermes_state import SessionDB 160 db = SessionDB(db_path=tmp_path / "state.db") 161 db.create_session("test_session_123", "telegram") 162 163 runner = _make_runner(session_db=db) 164 event = _make_event(text="/title \x00\x01\x02") 165 result = await runner._handle_title_command(event) 166 assert "empty after cleanup" in result 167 db.close() 168 169 @pytest.mark.asyncio 170 async def test_works_across_platforms(self, tmp_path): 171 """The /title command works for Discord, Slack, and WhatsApp too.""" 172 from hermes_state import SessionDB 173 for platform in [Platform.DISCORD, Platform.TELEGRAM]: 174 db = SessionDB(db_path=tmp_path / f"state_{platform.value}.db") 175 db.create_session("test_session_123", platform.value) 176 177 runner = _make_runner(session_db=db) 178 event = _make_event(text="/title Cross-Platform Test", platform=platform) 179 result = await runner._handle_title_command(event) 180 assert "Cross-Platform Test" in result 181 assert db.get_session_title("test_session_123") == "Cross-Platform Test" 182 db.close() 183 184 185 # --------------------------------------------------------------------------- 186 # /title in help and known_commands 187 # --------------------------------------------------------------------------- 188 189 190 class TestTitleInHelp: 191 """Verify /title appears in help text and known commands.""" 192 193 @pytest.mark.asyncio 194 async def test_title_in_help_output(self): 195 """The /help output includes /title.""" 196 runner = _make_runner() 197 event = _make_event(text="/help") 198 # Need hooks for help command 199 from gateway.hooks import HookRegistry 200 runner.hooks = HookRegistry() 201 result = await runner._handle_help_command(event) 202 assert "/title" in result 203 204 def test_title_is_known_command(self): 205 """The /title command is in the _known_commands set.""" 206 from gateway.run import GatewayRunner 207 import inspect 208 source = inspect.getsource(GatewayRunner._handle_message) 209 assert '"title"' in source 210 211 212 # --------------------------------------------------------------------------- 213 # /new with title 214 # --------------------------------------------------------------------------- 215 216 217 class TestResetCommandWithTitle: 218 """Tests for GatewayRunner._handle_reset_command with a title argument.""" 219 220 @pytest.mark.asyncio 221 async def test_reset_command_with_title(self): 222 """Sending /new <title> resets session and sets the title.""" 223 from datetime import datetime 224 225 from gateway.run import GatewayRunner 226 from gateway.session import SessionEntry, SessionSource, build_session_key 227 228 runner = object.__new__(GatewayRunner) 229 runner.config = GatewayConfig( 230 platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")} 231 ) 232 adapter = MagicMock() 233 adapter.send = AsyncMock() 234 runner.adapters = {Platform.TELEGRAM: adapter} 235 runner._voice_mode = {} 236 runner.hooks = SimpleNamespace(emit=AsyncMock(), loaded_hooks=False) 237 runner._session_model_overrides = {} 238 runner._pending_model_notes = {} 239 runner._background_tasks = set() 240 241 source = SessionSource( 242 platform=Platform.TELEGRAM, 243 user_id="12345", 244 chat_id="67890", 245 user_name="testuser", 246 ) 247 session_key = build_session_key(source) 248 new_session_entry = SessionEntry( 249 session_key=session_key, 250 session_id="sess-new", 251 created_at=datetime.now(), 252 updated_at=datetime.now(), 253 platform=Platform.TELEGRAM, 254 chat_type="dm", 255 ) 256 runner.session_store = MagicMock() 257 runner.session_store.get_or_create_session.return_value = new_session_entry 258 runner.session_store.reset_session.return_value = new_session_entry 259 runner.session_store._entries = {session_key: new_session_entry} 260 runner.session_store._generate_session_key.return_value = session_key 261 runner._running_agents = {} 262 runner._pending_messages = {} 263 runner._pending_approvals = {} 264 runner._session_db = MagicMock() 265 runner._agent_cache = {} 266 runner._agent_cache_lock = None 267 runner._is_user_authorized = lambda _source: True 268 runner._format_session_info = lambda: "" 269 270 event = _make_event(text="/new Custom Name") 271 result = await runner._handle_reset_command(event) 272 273 runner.session_store.reset_session.assert_called_once() 274 runner._session_db.set_session_title.assert_called_once_with( 275 "sess-new", "Custom Name" 276 ) 277 # Header reflects the applied title 278 assert "Custom Name" in str(result) 279 280 @pytest.mark.asyncio 281 async def test_reset_command_duplicate_title_surfaces_warning(self): 282 """/new <title> with an already-in-use title returns a warning in the reply.""" 283 from datetime import datetime 284 285 from gateway.run import GatewayRunner 286 from gateway.session import SessionEntry, SessionSource, build_session_key 287 288 runner = object.__new__(GatewayRunner) 289 runner.config = GatewayConfig( 290 platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")} 291 ) 292 adapter = MagicMock() 293 adapter.send = AsyncMock() 294 runner.adapters = {Platform.TELEGRAM: adapter} 295 runner._voice_mode = {} 296 runner.hooks = SimpleNamespace(emit=AsyncMock(), loaded_hooks=False) 297 runner._session_model_overrides = {} 298 runner._pending_model_notes = {} 299 runner._background_tasks = set() 300 301 source = SessionSource( 302 platform=Platform.TELEGRAM, 303 user_id="12345", 304 chat_id="67890", 305 user_name="testuser", 306 ) 307 session_key = build_session_key(source) 308 new_session_entry = SessionEntry( 309 session_key=session_key, 310 session_id="sess-new", 311 created_at=datetime.now(), 312 updated_at=datetime.now(), 313 platform=Platform.TELEGRAM, 314 chat_type="dm", 315 ) 316 runner.session_store = MagicMock() 317 runner.session_store.get_or_create_session.return_value = new_session_entry 318 runner.session_store.reset_session.return_value = new_session_entry 319 runner.session_store._entries = {session_key: new_session_entry} 320 runner.session_store._generate_session_key.return_value = session_key 321 runner._running_agents = {} 322 runner._pending_messages = {} 323 runner._pending_approvals = {} 324 runner._session_db = MagicMock() 325 runner._session_db.set_session_title.side_effect = ValueError( 326 "Title 'Dup' is already in use by session abc-123" 327 ) 328 runner._agent_cache = {} 329 runner._agent_cache_lock = None 330 runner._is_user_authorized = lambda _source: True 331 runner._format_session_info = lambda: "" 332 333 event = _make_event(text="/new Dup") 334 result = await runner._handle_reset_command(event) 335 336 runner._session_db.set_session_title.assert_called_once() 337 reply = str(result) 338 assert "already in use" in reply 339 assert "session started untitled" in reply 340 # Header must NOT claim the rejected title as the session name 341 assert "New session started: Dup" not in reply 342 343 344 # --------------------------------------------------------------------------- 345 # /new in help output 346 # --------------------------------------------------------------------------- 347 348 349 class TestNewInHelp: 350 """Verify /new appears in help text with the [name] args hint.""" 351 352 def test_new_command_in_help_output(self): 353 """The gateway help output includes /new with the [name] hint.""" 354 from hermes_cli.commands import gateway_help_lines 355 lines = gateway_help_lines() 356 new_line = next((line for line in lines if line.startswith("`/new ")), None) 357 assert new_line is not None 358 assert "[name]" in new_line