test_background_command.py
1 """Tests for /background gateway slash command. 2 3 Tests the _handle_background_command handler (run a prompt in a separate 4 background session) across gateway messenger platforms. 5 """ 6 7 import asyncio 8 import os 9 from unittest.mock import AsyncMock, MagicMock, patch 10 11 import pytest 12 13 from gateway.config import Platform 14 from gateway.platforms.base import MessageEvent 15 from gateway.session import SessionSource 16 17 18 def _make_event(text="/background", 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(): 31 """Create a bare GatewayRunner with minimal mocks.""" 32 from gateway.run import GatewayRunner 33 runner = object.__new__(GatewayRunner) 34 runner.adapters = {} 35 runner._voice_mode = {} 36 runner._session_db = None 37 runner._reasoning_config = None 38 runner._provider_routing = {} 39 runner._fallback_model = None 40 runner._running_agents = {} 41 runner._background_tasks = set() 42 43 mock_store = MagicMock() 44 runner.session_store = mock_store 45 46 from gateway.hooks import HookRegistry 47 runner.hooks = HookRegistry() 48 49 return runner 50 51 52 # --------------------------------------------------------------------------- 53 # _handle_background_command 54 # --------------------------------------------------------------------------- 55 56 57 class TestHandleBackgroundCommand: 58 """Tests for GatewayRunner._handle_background_command.""" 59 60 @pytest.mark.asyncio 61 async def test_no_prompt_shows_usage(self): 62 """Running /background with no prompt shows usage.""" 63 runner = _make_runner() 64 event = _make_event(text="/background") 65 result = await runner._handle_background_command(event) 66 assert "Usage:" in result 67 assert "/background" in result 68 69 @pytest.mark.asyncio 70 async def test_bg_alias_no_prompt_shows_usage(self): 71 """Running /bg with no prompt shows usage.""" 72 runner = _make_runner() 73 event = _make_event(text="/bg") 74 result = await runner._handle_background_command(event) 75 assert "Usage:" in result 76 77 @pytest.mark.asyncio 78 async def test_empty_prompt_shows_usage(self): 79 """Running /background with only whitespace shows usage.""" 80 runner = _make_runner() 81 event = _make_event(text="/background ") 82 result = await runner._handle_background_command(event) 83 assert "Usage:" in result 84 85 @pytest.mark.asyncio 86 async def test_valid_prompt_starts_task(self): 87 """Running /background with a prompt returns confirmation and starts task.""" 88 runner = _make_runner() 89 90 # Patch asyncio.create_task to capture the coroutine 91 created_tasks = [] 92 original_create_task = asyncio.create_task 93 94 def capture_task(coro, *args, **kwargs): 95 # Close the coroutine to avoid warnings 96 coro.close() 97 mock_task = MagicMock() 98 created_tasks.append(mock_task) 99 return mock_task 100 101 with patch("gateway.run.asyncio.create_task", side_effect=capture_task): 102 event = _make_event(text="/background Summarize the top HN stories") 103 result = await runner._handle_background_command(event) 104 105 assert "🔄" in result 106 assert "Background task started" in result 107 assert "bg_" in result # task ID starts with bg_ 108 assert "Summarize the top HN stories" in result 109 assert len(created_tasks) == 1 # background task was created 110 111 @pytest.mark.asyncio 112 async def test_prompt_truncated_in_preview(self): 113 """Long prompts are truncated to 60 chars in the confirmation message.""" 114 runner = _make_runner() 115 long_prompt = "A" * 100 116 117 with patch("gateway.run.asyncio.create_task", side_effect=lambda c, **kw: (c.close(), MagicMock())[1]): 118 event = _make_event(text=f"/background {long_prompt}") 119 result = await runner._handle_background_command(event) 120 121 assert "..." in result 122 # Should not contain the full prompt 123 assert long_prompt not in result 124 125 @pytest.mark.asyncio 126 async def test_task_id_is_unique(self): 127 """Each background task gets a unique task ID.""" 128 runner = _make_runner() 129 task_ids = set() 130 131 with patch("gateway.run.asyncio.create_task", side_effect=lambda c, **kw: (c.close(), MagicMock())[1]): 132 for i in range(5): 133 event = _make_event(text=f"/background task {i}") 134 result = await runner._handle_background_command(event) 135 # Extract task ID from result (format: "Task ID: bg_HHMMSS_hex") 136 for line in result.split("\n"): 137 if "Task ID:" in line: 138 tid = line.split("Task ID:")[1].strip() 139 task_ids.add(tid) 140 141 assert len(task_ids) == 5 # all unique 142 143 @pytest.mark.asyncio 144 async def test_works_across_platforms(self): 145 """The /background command works for all platforms.""" 146 for platform in [Platform.TELEGRAM, Platform.DISCORD, Platform.SLACK]: 147 runner = _make_runner() 148 with patch("gateway.run.asyncio.create_task", side_effect=lambda c, **kw: (c.close(), MagicMock())[1]): 149 event = _make_event( 150 text="/background test task", 151 platform=platform, 152 ) 153 result = await runner._handle_background_command(event) 154 assert "Background task started" in result 155 156 157 # --------------------------------------------------------------------------- 158 # _run_background_task 159 # --------------------------------------------------------------------------- 160 161 162 class TestRunBackgroundTask: 163 """Tests for GatewayRunner._run_background_task (the actual execution).""" 164 165 @pytest.mark.asyncio 166 async def test_no_adapter_returns_silently(self): 167 """When no adapter is available, the task returns without error.""" 168 runner = _make_runner() 169 source = SessionSource( 170 platform=Platform.TELEGRAM, 171 user_id="12345", 172 chat_id="67890", 173 user_name="testuser", 174 ) 175 # No adapters set — should not raise 176 await runner._run_background_task("test prompt", source, "bg_test") 177 178 @pytest.mark.asyncio 179 async def test_no_credentials_sends_error(self): 180 """When provider credentials are missing, an error is sent.""" 181 runner = _make_runner() 182 mock_adapter = AsyncMock() 183 mock_adapter.send = AsyncMock() 184 runner.adapters[Platform.TELEGRAM] = mock_adapter 185 186 source = SessionSource( 187 platform=Platform.TELEGRAM, 188 user_id="12345", 189 chat_id="67890", 190 user_name="testuser", 191 ) 192 193 with patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": None}): 194 await runner._run_background_task("test prompt", source, "bg_test") 195 196 # Should have sent an error message 197 mock_adapter.send.assert_called_once() 198 call_args = mock_adapter.send.call_args 199 assert "failed" in call_args[1].get("content", call_args[0][1] if len(call_args[0]) > 1 else "").lower() 200 201 @pytest.mark.asyncio 202 async def test_successful_task_sends_result(self): 203 """When the agent completes successfully, the result is sent.""" 204 runner = _make_runner() 205 mock_adapter = AsyncMock() 206 mock_adapter.send = AsyncMock() 207 mock_adapter.extract_media = MagicMock(return_value=([], "Hello from background!")) 208 mock_adapter.extract_images = MagicMock(return_value=([], "Hello from background!")) 209 runner.adapters[Platform.TELEGRAM] = mock_adapter 210 211 source = SessionSource( 212 platform=Platform.TELEGRAM, 213 user_id="12345", 214 chat_id="67890", 215 user_name="testuser", 216 ) 217 218 mock_result = {"final_response": "Hello from background!", "messages": []} 219 220 with patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}), \ 221 patch("run_agent.AIAgent") as MockAgent: 222 mock_agent_instance = MagicMock() 223 mock_agent_instance.shutdown_memory_provider = MagicMock() 224 mock_agent_instance.close = MagicMock() 225 mock_agent_instance.run_conversation.return_value = mock_result 226 MockAgent.return_value = mock_agent_instance 227 228 await runner._run_background_task("say hello", source, "bg_test") 229 230 # Should have sent the result 231 mock_adapter.send.assert_called_once() 232 call_args = mock_adapter.send.call_args 233 content = call_args[1].get("content", call_args[0][1] if len(call_args[0]) > 1 else "") 234 assert "Background task complete" in content 235 assert "Hello from background!" in content 236 mock_agent_instance.shutdown_memory_provider.assert_called_once() 237 mock_agent_instance.close.assert_called_once() 238 239 @pytest.mark.asyncio 240 async def test_agent_cleanup_runs_when_background_agent_raises(self): 241 """Temporary background agents must be cleaned up on error paths too.""" 242 runner = _make_runner() 243 mock_adapter = AsyncMock() 244 mock_adapter.send = AsyncMock() 245 runner.adapters[Platform.TELEGRAM] = mock_adapter 246 247 source = SessionSource( 248 platform=Platform.TELEGRAM, 249 user_id="12345", 250 chat_id="67890", 251 user_name="testuser", 252 ) 253 254 with patch("gateway.run._resolve_runtime_agent_kwargs", return_value={"api_key": "test-key"}), \ 255 patch("run_agent.AIAgent") as MockAgent: 256 mock_agent_instance = MagicMock() 257 mock_agent_instance.shutdown_memory_provider = MagicMock() 258 mock_agent_instance.close = MagicMock() 259 mock_agent_instance.run_conversation.side_effect = RuntimeError("boom") 260 MockAgent.return_value = mock_agent_instance 261 262 await runner._run_background_task("say hello", source, "bg_test") 263 264 mock_adapter.send.assert_called_once() 265 mock_agent_instance.shutdown_memory_provider.assert_called_once() 266 mock_agent_instance.close.assert_called_once() 267 268 @pytest.mark.asyncio 269 async def test_exception_sends_error_message(self): 270 """When the agent raises an exception, an error message is sent.""" 271 runner = _make_runner() 272 mock_adapter = AsyncMock() 273 mock_adapter.send = AsyncMock() 274 runner.adapters[Platform.TELEGRAM] = mock_adapter 275 276 source = SessionSource( 277 platform=Platform.TELEGRAM, 278 user_id="12345", 279 chat_id="67890", 280 user_name="testuser", 281 ) 282 283 with patch("gateway.run._resolve_runtime_agent_kwargs", side_effect=RuntimeError("boom")): 284 await runner._run_background_task("test prompt", source, "bg_test") 285 286 mock_adapter.send.assert_called_once() 287 call_args = mock_adapter.send.call_args 288 content = call_args[1].get("content", call_args[0][1] if len(call_args[0]) > 1 else "") 289 assert "failed" in content.lower() 290 291 292 # --------------------------------------------------------------------------- 293 # /background in help and known_commands 294 # --------------------------------------------------------------------------- 295 296 297 class TestBackgroundInHelp: 298 """Verify /background appears in help text and known commands.""" 299 300 @pytest.mark.asyncio 301 async def test_background_in_help_output(self): 302 """The /help output includes /background.""" 303 runner = _make_runner() 304 event = _make_event(text="/help") 305 result = await runner._handle_help_command(event) 306 assert "/background" in result 307 308 def test_background_is_known_command(self): 309 """The /background command is in GATEWAY_KNOWN_COMMANDS.""" 310 from hermes_cli.commands import GATEWAY_KNOWN_COMMANDS 311 assert "background" in GATEWAY_KNOWN_COMMANDS 312 313 def test_bg_alias_is_known_command(self): 314 """The /bg alias is in GATEWAY_KNOWN_COMMANDS.""" 315 from hermes_cli.commands import GATEWAY_KNOWN_COMMANDS 316 assert "bg" in GATEWAY_KNOWN_COMMANDS 317 318 319 # --------------------------------------------------------------------------- 320 # CLI /background command definition 321 # --------------------------------------------------------------------------- 322 323 324 class TestBackgroundInCLICommands: 325 """Verify /background is registered in the CLI command system.""" 326 327 def test_background_in_commands_dict(self): 328 """The /background command is in the COMMANDS dict.""" 329 from hermes_cli.commands import COMMANDS 330 assert "/background" in COMMANDS 331 332 def test_bg_alias_in_commands_dict(self): 333 """The /bg alias is in the COMMANDS dict.""" 334 from hermes_cli.commands import COMMANDS 335 assert "/bg" in COMMANDS 336 337 def test_background_in_session_category(self): 338 """The /background command is in the Session category.""" 339 from hermes_cli.commands import COMMANDS_BY_CATEGORY 340 assert "/background" in COMMANDS_BY_CATEGORY["Session"] 341 342 def test_background_autocompletes(self): 343 """The /background command appears in autocomplete results.""" 344 pytest.importorskip("prompt_toolkit") 345 from hermes_cli.commands import SlashCommandCompleter 346 from prompt_toolkit.document import Document 347 348 completer = SlashCommandCompleter() 349 doc = Document("backgro") # Partial match 350 completions = list(completer.get_completions(doc, None)) 351 # Text doesn't start with / so no completions 352 assert len(completions) == 0 353 354 doc = Document("/backgro") # With slash prefix 355 completions = list(completer.get_completions(doc, None)) 356 cmd_displays = [str(c.display) for c in completions] 357 assert any("/background" in d for d in cmd_displays)