test_update_command.py
1 """Tests for /update gateway slash command. 2 3 Tests both the _handle_update_command handler (spawns update process) and 4 the _send_update_notification startup hook (sends results after restart). 5 """ 6 7 import json 8 import os 9 from pathlib import Path 10 from unittest.mock import patch, MagicMock, AsyncMock 11 12 import pytest 13 14 from gateway.config import Platform 15 from gateway.platforms.base import MessageEvent 16 from gateway.session import SessionSource 17 18 19 def _make_event(text="/update", platform=Platform.TELEGRAM, 20 user_id="12345", chat_id="67890", thread_id=None): 21 """Build a MessageEvent for testing.""" 22 source = SessionSource( 23 platform=platform, 24 user_id=user_id, 25 chat_id=chat_id, 26 user_name="testuser", 27 thread_id=thread_id, 28 ) 29 return MessageEvent(text=text, source=source) 30 31 32 def _make_runner(): 33 """Create a bare GatewayRunner without calling __init__.""" 34 from gateway.run import GatewayRunner 35 runner = object.__new__(GatewayRunner) 36 runner.adapters = {} 37 runner._voice_mode = {} 38 return runner 39 40 41 # --------------------------------------------------------------------------- 42 # _handle_update_command 43 # --------------------------------------------------------------------------- 44 45 46 class TestHandleUpdateCommand: 47 """Tests for GatewayRunner._handle_update_command.""" 48 49 @pytest.mark.asyncio 50 async def test_managed_install_returns_package_manager_guidance(self, monkeypatch): 51 runner = _make_runner() 52 event = _make_event() 53 monkeypatch.setenv("HERMES_MANAGED", "homebrew") 54 55 result = await runner._handle_update_command(event) 56 57 assert "managed by Homebrew" in result 58 assert "brew upgrade hermes-agent" in result 59 60 @pytest.mark.asyncio 61 async def test_no_git_directory(self, tmp_path): 62 """Returns an error when .git does not exist.""" 63 runner = _make_runner() 64 event = _make_event() 65 # Point _hermes_home to tmp_path and project_root to a dir without .git 66 fake_root = tmp_path / "project" 67 fake_root.mkdir() 68 with patch("gateway.run._hermes_home", tmp_path), \ 69 patch("gateway.run.Path") as MockPath: 70 # Path(__file__).parent.parent.resolve() -> fake_root 71 MockPath.return_value = MagicMock() 72 MockPath.__truediv__ = Path.__truediv__ 73 # Easier: just patch the __file__ resolution in the method 74 pass 75 76 # Simpler approach — mock at method level using a wrapper 77 from gateway.run import GatewayRunner 78 runner = _make_runner() 79 80 with patch("gateway.run._hermes_home", tmp_path): 81 # The handler does Path(__file__).parent.parent.resolve() 82 # We need to make project_root / '.git' not exist. 83 # Since Path(__file__) resolves to the real gateway/run.py, 84 # project_root will be the real hermes-agent dir (which HAS .git). 85 # Patch Path to control this. 86 original_path = Path 87 88 class FakePath(type(Path())): 89 pass 90 91 # Actually, simplest: just patch the specific file attr 92 fake_file = str(fake_root / "gateway" / "run.py") 93 (fake_root / "gateway").mkdir(parents=True) 94 (fake_root / "gateway" / "run.py").touch() 95 96 with patch("gateway.run.__file__", fake_file): 97 result = await runner._handle_update_command(event) 98 99 assert "Not a git repository" in result 100 101 @pytest.mark.asyncio 102 async def test_no_hermes_binary(self, tmp_path): 103 """Returns error when hermes is not on PATH and hermes_cli is not importable.""" 104 runner = _make_runner() 105 event = _make_event() 106 107 # Create project dir WITH .git 108 fake_root = tmp_path / "project" 109 fake_root.mkdir() 110 (fake_root / ".git").mkdir() 111 (fake_root / "gateway").mkdir() 112 (fake_root / "gateway" / "run.py").touch() 113 fake_file = str(fake_root / "gateway" / "run.py") 114 115 with patch("gateway.run._hermes_home", tmp_path), \ 116 patch("gateway.run.__file__", fake_file), \ 117 patch("shutil.which", return_value=None), \ 118 patch("importlib.util.find_spec", return_value=None): 119 result = await runner._handle_update_command(event) 120 121 assert "Could not locate" in result 122 assert "hermes update" in result 123 124 @pytest.mark.asyncio 125 async def test_fallback_to_sys_executable(self, tmp_path): 126 """Falls back to sys.executable -m hermes_cli.main when hermes not on PATH.""" 127 runner = _make_runner() 128 event = _make_event() 129 130 fake_root = tmp_path / "project" 131 fake_root.mkdir() 132 (fake_root / ".git").mkdir() 133 (fake_root / "gateway").mkdir() 134 (fake_root / "gateway" / "run.py").touch() 135 fake_file = str(fake_root / "gateway" / "run.py") 136 hermes_home = tmp_path / "hermes" 137 hermes_home.mkdir() 138 139 mock_popen = MagicMock() 140 fake_spec = MagicMock() 141 142 with patch("gateway.run._hermes_home", hermes_home), \ 143 patch("gateway.run.__file__", fake_file), \ 144 patch("shutil.which", return_value=None), \ 145 patch("importlib.util.find_spec", return_value=fake_spec), \ 146 patch("subprocess.Popen", mock_popen): 147 result = await runner._handle_update_command(event) 148 149 assert "Starting Hermes update" in result 150 call_args = mock_popen.call_args[0][0] 151 # The update_cmd uses sys.executable -m hermes_cli.main 152 joined = " ".join(call_args) if isinstance(call_args, list) else call_args 153 assert "hermes_cli.main" in joined or "bash" in call_args[0] 154 155 @pytest.mark.asyncio 156 async def test_resolve_hermes_bin_prefers_which(self, tmp_path): 157 """_resolve_hermes_bin returns argv parts from shutil.which when available.""" 158 from gateway.run import _resolve_hermes_bin 159 160 with patch("shutil.which", return_value="/custom/path/hermes"): 161 result = _resolve_hermes_bin() 162 163 assert result == ["/custom/path/hermes"] 164 165 @pytest.mark.asyncio 166 async def test_resolve_hermes_bin_fallback(self): 167 """_resolve_hermes_bin falls back to sys.executable argv when which fails.""" 168 import sys 169 from gateway.run import _resolve_hermes_bin 170 171 fake_spec = MagicMock() 172 with patch("shutil.which", return_value=None), \ 173 patch("importlib.util.find_spec", return_value=fake_spec): 174 result = _resolve_hermes_bin() 175 176 assert result == [sys.executable, "-m", "hermes_cli.main"] 177 178 @pytest.mark.asyncio 179 async def test_resolve_hermes_bin_returns_none_when_both_fail(self): 180 """_resolve_hermes_bin returns None when both strategies fail.""" 181 from gateway.run import _resolve_hermes_bin 182 183 with patch("shutil.which", return_value=None), \ 184 patch("importlib.util.find_spec", return_value=None): 185 result = _resolve_hermes_bin() 186 187 assert result is None 188 189 @pytest.mark.asyncio 190 async def test_writes_pending_marker(self, tmp_path): 191 """Writes .update_pending.json with correct platform and chat info.""" 192 runner = _make_runner() 193 event = _make_event(platform=Platform.TELEGRAM, chat_id="99999") 194 195 fake_root = tmp_path / "project" 196 fake_root.mkdir() 197 (fake_root / ".git").mkdir() 198 (fake_root / "gateway").mkdir() 199 (fake_root / "gateway" / "run.py").touch() 200 fake_file = str(fake_root / "gateway" / "run.py") 201 hermes_home = tmp_path / "hermes" 202 hermes_home.mkdir() 203 204 with patch("gateway.run._hermes_home", hermes_home), \ 205 patch("gateway.run.__file__", fake_file), \ 206 patch("shutil.which", side_effect=lambda x: "/usr/bin/hermes" if x == "hermes" else "/usr/bin/setsid"), \ 207 patch("subprocess.Popen"): 208 result = await runner._handle_update_command(event) 209 210 pending_path = hermes_home / ".update_pending.json" 211 assert pending_path.exists() 212 data = json.loads(pending_path.read_text()) 213 assert data["platform"] == "telegram" 214 assert data["chat_id"] == "99999" 215 assert "timestamp" in data 216 assert not (hermes_home / ".update_exit_code").exists() 217 218 @pytest.mark.asyncio 219 async def test_writes_pending_marker_with_thread_id(self, tmp_path): 220 """Persists thread_id so update notifications can route back to the thread.""" 221 runner = _make_runner() 222 event = _make_event( 223 platform=Platform.TELEGRAM, 224 chat_id="99999", 225 thread_id="777", 226 ) 227 228 fake_root = tmp_path / "project" 229 fake_root.mkdir() 230 (fake_root / ".git").mkdir() 231 (fake_root / "gateway").mkdir() 232 (fake_root / "gateway" / "run.py").touch() 233 fake_file = str(fake_root / "gateway" / "run.py") 234 hermes_home = tmp_path / "hermes" 235 hermes_home.mkdir() 236 237 with patch("gateway.run._hermes_home", hermes_home), \ 238 patch("gateway.run.__file__", fake_file), \ 239 patch("shutil.which", side_effect=lambda x: "/usr/bin/hermes" if x == "hermes" else "/usr/bin/setsid"), \ 240 patch("subprocess.Popen"): 241 await runner._handle_update_command(event) 242 243 data = json.loads((hermes_home / ".update_pending.json").read_text()) 244 assert data["thread_id"] == "777" 245 246 @pytest.mark.asyncio 247 async def test_spawns_setsid(self, tmp_path): 248 """Uses setsid when available.""" 249 runner = _make_runner() 250 event = _make_event() 251 252 fake_root = tmp_path / "project" 253 fake_root.mkdir() 254 (fake_root / ".git").mkdir() 255 (fake_root / "gateway").mkdir() 256 (fake_root / "gateway" / "run.py").touch() 257 fake_file = str(fake_root / "gateway" / "run.py") 258 hermes_home = tmp_path / "hermes" 259 hermes_home.mkdir() 260 261 mock_popen = MagicMock() 262 with patch("gateway.run._hermes_home", hermes_home), \ 263 patch("gateway.run.__file__", fake_file), \ 264 patch("shutil.which", side_effect=lambda x: f"/usr/bin/{x}"), \ 265 patch("subprocess.Popen", mock_popen): 266 result = await runner._handle_update_command(event) 267 268 # Verify setsid was used 269 call_args = mock_popen.call_args[0][0] 270 assert call_args[0] == "/usr/bin/setsid" 271 assert call_args[1] == "bash" 272 assert ".update_exit_code" in call_args[-1] 273 assert "Starting Hermes update" in result 274 275 @pytest.mark.asyncio 276 async def test_fallback_when_no_setsid(self, tmp_path): 277 """Falls back to start_new_session=True when setsid is not available.""" 278 runner = _make_runner() 279 event = _make_event() 280 281 fake_root = tmp_path / "project" 282 fake_root.mkdir() 283 (fake_root / ".git").mkdir() 284 (fake_root / "gateway").mkdir() 285 (fake_root / "gateway" / "run.py").touch() 286 fake_file = str(fake_root / "gateway" / "run.py") 287 hermes_home = tmp_path / "hermes" 288 hermes_home.mkdir() 289 290 mock_popen = MagicMock() 291 292 def which_no_setsid(x): 293 if x == "hermes": 294 return "/usr/bin/hermes" 295 if x == "setsid": 296 return None 297 return None 298 299 with patch("gateway.run._hermes_home", hermes_home), \ 300 patch("gateway.run.__file__", fake_file), \ 301 patch("shutil.which", side_effect=which_no_setsid), \ 302 patch("subprocess.Popen", mock_popen): 303 result = await runner._handle_update_command(event) 304 305 # Verify plain bash -c fallback (no nohup, no setsid) 306 call_args = mock_popen.call_args[0][0] 307 assert call_args[0] == "bash" 308 assert "nohup" not in call_args[2] 309 assert ".update_exit_code" in call_args[2] 310 # start_new_session=True should be in kwargs 311 call_kwargs = mock_popen.call_args[1] 312 assert call_kwargs.get("start_new_session") is True 313 assert "Starting Hermes update" in result 314 315 @pytest.mark.asyncio 316 async def test_popen_failure_cleans_up(self, tmp_path): 317 """Cleans up pending file and returns error on Popen failure.""" 318 runner = _make_runner() 319 event = _make_event() 320 321 fake_root = tmp_path / "project" 322 fake_root.mkdir() 323 (fake_root / ".git").mkdir() 324 (fake_root / "gateway").mkdir() 325 (fake_root / "gateway" / "run.py").touch() 326 fake_file = str(fake_root / "gateway" / "run.py") 327 hermes_home = tmp_path / "hermes" 328 hermes_home.mkdir() 329 330 with patch("gateway.run._hermes_home", hermes_home), \ 331 patch("gateway.run.__file__", fake_file), \ 332 patch("shutil.which", side_effect=lambda x: f"/usr/bin/{x}"), \ 333 patch("subprocess.Popen", side_effect=OSError("spawn failed")): 334 result = await runner._handle_update_command(event) 335 336 assert "Failed to start update" in result 337 # Pending file should be cleaned up 338 assert not (hermes_home / ".update_pending.json").exists() 339 assert not (hermes_home / ".update_exit_code").exists() 340 341 @pytest.mark.asyncio 342 async def test_returns_user_friendly_message(self, tmp_path): 343 """The success response is user-friendly.""" 344 runner = _make_runner() 345 event = _make_event() 346 347 fake_root = tmp_path / "project" 348 fake_root.mkdir() 349 (fake_root / ".git").mkdir() 350 (fake_root / "gateway").mkdir() 351 (fake_root / "gateway" / "run.py").touch() 352 fake_file = str(fake_root / "gateway" / "run.py") 353 hermes_home = tmp_path / "hermes" 354 hermes_home.mkdir() 355 356 with patch("gateway.run._hermes_home", hermes_home), \ 357 patch("gateway.run.__file__", fake_file), \ 358 patch("shutil.which", side_effect=lambda x: f"/usr/bin/{x}"), \ 359 patch("subprocess.Popen"): 360 result = await runner._handle_update_command(event) 361 362 assert "stream progress" in result 363 364 365 # --------------------------------------------------------------------------- 366 # _send_update_notification 367 # --------------------------------------------------------------------------- 368 369 370 class TestSendUpdateNotification: 371 """Tests for GatewayRunner._send_update_notification.""" 372 373 @pytest.mark.asyncio 374 async def test_no_pending_file_is_noop(self, tmp_path): 375 """Does nothing when no pending file exists.""" 376 runner = _make_runner() 377 hermes_home = tmp_path / "hermes" 378 hermes_home.mkdir() 379 380 with patch("gateway.run._hermes_home", hermes_home): 381 # Should not raise 382 await runner._send_update_notification() 383 384 @pytest.mark.asyncio 385 async def test_defers_notification_while_update_still_running(self, tmp_path): 386 """Returns False and keeps marker files when the update has not exited yet.""" 387 runner = _make_runner() 388 hermes_home = tmp_path / "hermes" 389 hermes_home.mkdir() 390 391 pending_path = hermes_home / ".update_pending.json" 392 pending_path.write_text(json.dumps({ 393 "platform": "telegram", "chat_id": "67890", "user_id": "12345", 394 })) 395 (hermes_home / ".update_output.txt").write_text("still running") 396 397 mock_adapter = AsyncMock() 398 runner.adapters = {Platform.TELEGRAM: mock_adapter} 399 400 with patch("gateway.run._hermes_home", hermes_home): 401 result = await runner._send_update_notification() 402 403 assert result is False 404 mock_adapter.send.assert_not_called() 405 assert pending_path.exists() 406 407 @pytest.mark.asyncio 408 async def test_recovers_from_claimed_pending_file(self, tmp_path): 409 """A claimed pending file from a crashed notifier is still deliverable.""" 410 runner = _make_runner() 411 hermes_home = tmp_path / "hermes" 412 hermes_home.mkdir() 413 414 claimed_path = hermes_home / ".update_pending.claimed.json" 415 claimed_path.write_text(json.dumps({ 416 "platform": "telegram", "chat_id": "67890", "user_id": "12345", 417 })) 418 (hermes_home / ".update_output.txt").write_text("done") 419 (hermes_home / ".update_exit_code").write_text("0") 420 421 mock_adapter = AsyncMock() 422 runner.adapters = {Platform.TELEGRAM: mock_adapter} 423 424 with patch("gateway.run._hermes_home", hermes_home): 425 result = await runner._send_update_notification() 426 427 assert result is True 428 mock_adapter.send.assert_called_once() 429 assert not claimed_path.exists() 430 431 @pytest.mark.asyncio 432 async def test_sends_notification_with_output(self, tmp_path): 433 """Sends update output to the correct platform and chat.""" 434 runner = _make_runner() 435 hermes_home = tmp_path / "hermes" 436 hermes_home.mkdir() 437 438 # Write pending marker 439 pending = { 440 "platform": "telegram", 441 "chat_id": "67890", 442 "user_id": "12345", 443 "timestamp": "2026-03-04T21:00:00", 444 } 445 (hermes_home / ".update_pending.json").write_text(json.dumps(pending)) 446 (hermes_home / ".update_output.txt").write_text( 447 "→ Found 3 new commit(s)\n✓ Code updated!\n✓ Update complete!" 448 ) 449 (hermes_home / ".update_exit_code").write_text("0") 450 451 # Mock the adapter 452 mock_adapter = AsyncMock() 453 mock_adapter.send = AsyncMock() 454 runner.adapters = {Platform.TELEGRAM: mock_adapter} 455 456 with patch("gateway.run._hermes_home", hermes_home): 457 await runner._send_update_notification() 458 459 mock_adapter.send.assert_called_once() 460 call_args = mock_adapter.send.call_args 461 assert call_args[0][0] == "67890" # chat_id 462 assert "Update complete" in call_args[0][1] or "update finished" in call_args[0][1].lower() 463 464 @pytest.mark.asyncio 465 async def test_sends_notification_with_thread_metadata(self, tmp_path): 466 """Final update notification preserves thread metadata when present.""" 467 runner = _make_runner() 468 hermes_home = tmp_path / "hermes" 469 hermes_home.mkdir() 470 471 pending = { 472 "platform": "telegram", 473 "chat_id": "67890", 474 "thread_id": "777", 475 "user_id": "12345", 476 } 477 (hermes_home / ".update_pending.json").write_text(json.dumps(pending)) 478 (hermes_home / ".update_output.txt").write_text("done") 479 (hermes_home / ".update_exit_code").write_text("0") 480 481 mock_adapter = AsyncMock() 482 runner.adapters = {Platform.TELEGRAM: mock_adapter} 483 484 with patch("gateway.run._hermes_home", hermes_home): 485 await runner._send_update_notification() 486 487 assert mock_adapter.send.call_args.kwargs["metadata"] == {"thread_id": "777"} 488 489 @pytest.mark.asyncio 490 async def test_strips_ansi_codes(self, tmp_path): 491 """ANSI escape codes are removed from output.""" 492 runner = _make_runner() 493 hermes_home = tmp_path / "hermes" 494 hermes_home.mkdir() 495 496 pending = {"platform": "telegram", "chat_id": "111", "user_id": "222"} 497 (hermes_home / ".update_pending.json").write_text(json.dumps(pending)) 498 (hermes_home / ".update_output.txt").write_text( 499 "\x1b[32m✓ Code updated!\x1b[0m\n\x1b[1mDone\x1b[0m" 500 ) 501 (hermes_home / ".update_exit_code").write_text("0") 502 503 mock_adapter = AsyncMock() 504 runner.adapters = {Platform.TELEGRAM: mock_adapter} 505 506 with patch("gateway.run._hermes_home", hermes_home): 507 await runner._send_update_notification() 508 509 sent_text = mock_adapter.send.call_args[0][1] 510 assert "\x1b[" not in sent_text 511 assert "Code updated" in sent_text 512 513 @pytest.mark.asyncio 514 async def test_truncates_long_output(self, tmp_path): 515 """Output longer than 3500 chars is truncated.""" 516 runner = _make_runner() 517 hermes_home = tmp_path / "hermes" 518 hermes_home.mkdir() 519 520 pending = {"platform": "telegram", "chat_id": "111", "user_id": "222"} 521 (hermes_home / ".update_pending.json").write_text(json.dumps(pending)) 522 (hermes_home / ".update_output.txt").write_text("x" * 5000) 523 (hermes_home / ".update_exit_code").write_text("0") 524 525 mock_adapter = AsyncMock() 526 runner.adapters = {Platform.TELEGRAM: mock_adapter} 527 528 with patch("gateway.run._hermes_home", hermes_home): 529 await runner._send_update_notification() 530 531 sent_text = mock_adapter.send.call_args[0][1] 532 # Should start with truncation marker 533 assert "…" in sent_text 534 # Total message should not be absurdly long 535 assert len(sent_text) < 4500 536 537 @pytest.mark.asyncio 538 async def test_sends_failure_message_when_update_fails(self, tmp_path): 539 """Non-zero exit codes produce a failure notification with captured output.""" 540 runner = _make_runner() 541 hermes_home = tmp_path / "hermes" 542 hermes_home.mkdir() 543 544 pending = {"platform": "telegram", "chat_id": "111", "user_id": "222"} 545 (hermes_home / ".update_pending.json").write_text(json.dumps(pending)) 546 (hermes_home / ".update_output.txt").write_text("Traceback: boom") 547 (hermes_home / ".update_exit_code").write_text("1") 548 549 mock_adapter = AsyncMock() 550 runner.adapters = {Platform.TELEGRAM: mock_adapter} 551 552 with patch("gateway.run._hermes_home", hermes_home): 553 result = await runner._send_update_notification() 554 555 assert result is True 556 sent_text = mock_adapter.send.call_args[0][1] 557 assert "update failed" in sent_text.lower() 558 assert "Traceback: boom" in sent_text 559 560 @pytest.mark.asyncio 561 async def test_sends_generic_message_when_no_output(self, tmp_path): 562 """Sends a success message even if the output file is missing.""" 563 runner = _make_runner() 564 hermes_home = tmp_path / "hermes" 565 hermes_home.mkdir() 566 567 pending = {"platform": "telegram", "chat_id": "111", "user_id": "222"} 568 (hermes_home / ".update_pending.json").write_text(json.dumps(pending)) 569 # No .update_output.txt created 570 (hermes_home / ".update_exit_code").write_text("0") 571 572 mock_adapter = AsyncMock() 573 runner.adapters = {Platform.TELEGRAM: mock_adapter} 574 575 with patch("gateway.run._hermes_home", hermes_home): 576 await runner._send_update_notification() 577 578 sent_text = mock_adapter.send.call_args[0][1] 579 assert "finished successfully" in sent_text 580 581 @pytest.mark.asyncio 582 async def test_cleans_up_files_after_notification(self, tmp_path): 583 """Both marker and output files are deleted after notification.""" 584 runner = _make_runner() 585 hermes_home = tmp_path / "hermes" 586 hermes_home.mkdir() 587 588 pending_path = hermes_home / ".update_pending.json" 589 output_path = hermes_home / ".update_output.txt" 590 exit_code_path = hermes_home / ".update_exit_code" 591 pending_path.write_text(json.dumps({ 592 "platform": "telegram", "chat_id": "111", "user_id": "222", 593 })) 594 output_path.write_text("✓ Done") 595 exit_code_path.write_text("0") 596 597 mock_adapter = AsyncMock() 598 runner.adapters = {Platform.TELEGRAM: mock_adapter} 599 600 with patch("gateway.run._hermes_home", hermes_home): 601 await runner._send_update_notification() 602 603 assert not pending_path.exists() 604 assert not output_path.exists() 605 assert not exit_code_path.exists() 606 607 @pytest.mark.asyncio 608 async def test_cleans_up_on_error(self, tmp_path): 609 """Files are cleaned up even if notification fails.""" 610 runner = _make_runner() 611 hermes_home = tmp_path / "hermes" 612 hermes_home.mkdir() 613 614 pending_path = hermes_home / ".update_pending.json" 615 output_path = hermes_home / ".update_output.txt" 616 exit_code_path = hermes_home / ".update_exit_code" 617 pending_path.write_text(json.dumps({ 618 "platform": "telegram", "chat_id": "111", "user_id": "222", 619 })) 620 output_path.write_text("✓ Done") 621 exit_code_path.write_text("0") 622 623 # Adapter send raises 624 mock_adapter = AsyncMock() 625 mock_adapter.send.side_effect = RuntimeError("network error") 626 runner.adapters = {Platform.TELEGRAM: mock_adapter} 627 628 with patch("gateway.run._hermes_home", hermes_home): 629 await runner._send_update_notification() 630 631 # Files should still be cleaned up (finally block) 632 assert not pending_path.exists() 633 assert not output_path.exists() 634 assert not exit_code_path.exists() 635 636 @pytest.mark.asyncio 637 async def test_handles_corrupt_pending_file(self, tmp_path): 638 """Gracefully handles a malformed pending JSON file.""" 639 runner = _make_runner() 640 hermes_home = tmp_path / "hermes" 641 hermes_home.mkdir() 642 643 pending_path = hermes_home / ".update_pending.json" 644 pending_path.write_text("{corrupt json!!") 645 646 with patch("gateway.run._hermes_home", hermes_home): 647 # Should not raise 648 await runner._send_update_notification() 649 650 # File should be cleaned up 651 assert not pending_path.exists() 652 653 @pytest.mark.asyncio 654 async def test_no_adapter_for_platform(self, tmp_path): 655 """Does not crash if the platform adapter is not connected.""" 656 runner = _make_runner() 657 hermes_home = tmp_path / "hermes" 658 hermes_home.mkdir() 659 660 pending = {"platform": "discord", "chat_id": "111", "user_id": "222"} 661 pending_path = hermes_home / ".update_pending.json" 662 output_path = hermes_home / ".update_output.txt" 663 exit_code_path = hermes_home / ".update_exit_code" 664 pending_path.write_text(json.dumps(pending)) 665 output_path.write_text("Done") 666 exit_code_path.write_text("0") 667 668 # Only telegram adapter available, but pending says discord 669 mock_adapter = AsyncMock() 670 runner.adapters = {Platform.TELEGRAM: mock_adapter} 671 672 with patch("gateway.run._hermes_home", hermes_home): 673 await runner._send_update_notification() 674 675 # send should not have been called (wrong platform) 676 mock_adapter.send.assert_not_called() 677 # Files should still be cleaned up 678 assert not pending_path.exists() 679 assert not exit_code_path.exists() 680 681 682 # --------------------------------------------------------------------------- 683 # /update in help and known_commands 684 # --------------------------------------------------------------------------- 685 686 687 class TestUpdateInHelp: 688 """Verify /update appears in help text and known commands set.""" 689 690 @pytest.mark.asyncio 691 async def test_update_in_help_output(self): 692 """The /help output includes /update.""" 693 runner = _make_runner() 694 event = _make_event(text="/help") 695 result = await runner._handle_help_command(event) 696 assert "/update" in result 697 698 def test_update_is_known_command(self): 699 """The /update command is in the help text (proxy for _known_commands).""" 700 # _known_commands is local to _handle_message, so we verify by 701 # checking the help output includes it. 702 from gateway.run import GatewayRunner 703 import inspect 704 source = inspect.getsource(GatewayRunner._handle_message) 705 assert '"update"' in source