test_debug.py
1 """Tests for ``hermes debug`` CLI command and debug utilities.""" 2 3 import os 4 import sys 5 import urllib.error 6 from pathlib import Path 7 from unittest.mock import MagicMock, patch, call 8 9 import pytest 10 11 # --------------------------------------------------------------------------- 12 # Fixtures 13 # --------------------------------------------------------------------------- 14 15 @pytest.fixture 16 def hermes_home(tmp_path, monkeypatch): 17 """Set up an isolated HERMES_HOME with minimal logs.""" 18 home = tmp_path / ".hermes" 19 home.mkdir() 20 monkeypatch.setenv("HERMES_HOME", str(home)) 21 22 # Create log files 23 logs_dir = home / "logs" 24 logs_dir.mkdir() 25 (logs_dir / "agent.log").write_text( 26 "2026-04-12 17:00:00 INFO agent: session started\n" 27 "2026-04-12 17:00:01 INFO tools.terminal: running ls\n" 28 "2026-04-12 17:00:02 WARNING agent: high token usage\n" 29 ) 30 (logs_dir / "errors.log").write_text( 31 "2026-04-12 17:00:05 ERROR gateway.run: connection lost\n" 32 ) 33 (logs_dir / "gateway.log").write_text( 34 "2026-04-12 17:00:10 INFO gateway.run: started\n" 35 ) 36 37 return home 38 39 40 # --------------------------------------------------------------------------- 41 # Unit tests for upload helpers 42 # --------------------------------------------------------------------------- 43 44 class TestUploadPasteRs: 45 """Test paste.rs upload path.""" 46 47 def test_upload_paste_rs_success(self): 48 from hermes_cli.debug import _upload_paste_rs 49 50 mock_resp = MagicMock() 51 mock_resp.read.return_value = b"https://paste.rs/abc123\n" 52 mock_resp.__enter__ = lambda s: s 53 mock_resp.__exit__ = MagicMock(return_value=False) 54 55 with patch("hermes_cli.debug.urllib.request.urlopen", return_value=mock_resp): 56 url = _upload_paste_rs("hello world") 57 58 assert url == "https://paste.rs/abc123" 59 60 def test_upload_paste_rs_bad_response(self): 61 from hermes_cli.debug import _upload_paste_rs 62 63 mock_resp = MagicMock() 64 mock_resp.read.return_value = b"<html>error</html>" 65 mock_resp.__enter__ = lambda s: s 66 mock_resp.__exit__ = MagicMock(return_value=False) 67 68 with patch("hermes_cli.debug.urllib.request.urlopen", return_value=mock_resp): 69 with pytest.raises(ValueError, match="Unexpected response"): 70 _upload_paste_rs("test") 71 72 def test_upload_paste_rs_network_error(self): 73 from hermes_cli.debug import _upload_paste_rs 74 75 with patch( 76 "hermes_cli.debug.urllib.request.urlopen", 77 side_effect=urllib.error.URLError("connection refused"), 78 ): 79 with pytest.raises(urllib.error.URLError): 80 _upload_paste_rs("test") 81 82 83 class TestUploadDpasteCom: 84 """Test dpaste.com fallback upload path.""" 85 86 def test_upload_dpaste_com_success(self): 87 from hermes_cli.debug import _upload_dpaste_com 88 89 mock_resp = MagicMock() 90 mock_resp.read.return_value = b"https://dpaste.com/ABCDEFG\n" 91 mock_resp.__enter__ = lambda s: s 92 mock_resp.__exit__ = MagicMock(return_value=False) 93 94 with patch("hermes_cli.debug.urllib.request.urlopen", return_value=mock_resp): 95 url = _upload_dpaste_com("hello world", expiry_days=7) 96 97 assert url == "https://dpaste.com/ABCDEFG" 98 99 100 class TestUploadToPastebin: 101 """Test the combined upload with fallback.""" 102 103 def test_tries_paste_rs_first(self): 104 from hermes_cli.debug import upload_to_pastebin 105 106 with patch("hermes_cli.debug._upload_paste_rs", 107 return_value="https://paste.rs/test") as prs: 108 url = upload_to_pastebin("content") 109 110 assert url == "https://paste.rs/test" 111 prs.assert_called_once() 112 113 def test_falls_back_to_dpaste_com(self): 114 from hermes_cli.debug import upload_to_pastebin 115 116 with patch("hermes_cli.debug._upload_paste_rs", 117 side_effect=Exception("down")), \ 118 patch("hermes_cli.debug._upload_dpaste_com", 119 return_value="https://dpaste.com/TEST") as dp: 120 url = upload_to_pastebin("content") 121 122 assert url == "https://dpaste.com/TEST" 123 dp.assert_called_once() 124 125 def test_raises_when_both_fail(self): 126 from hermes_cli.debug import upload_to_pastebin 127 128 with patch("hermes_cli.debug._upload_paste_rs", 129 side_effect=Exception("err1")), \ 130 patch("hermes_cli.debug._upload_dpaste_com", 131 side_effect=Exception("err2")): 132 with pytest.raises(RuntimeError, match="Failed to upload"): 133 upload_to_pastebin("content") 134 135 136 # --------------------------------------------------------------------------- 137 # Log reading 138 # --------------------------------------------------------------------------- 139 140 class TestCaptureLogSnapshot: 141 """Test _capture_log_snapshot for log reading and truncation.""" 142 143 def test_reads_small_file(self, hermes_home): 144 from hermes_cli.debug import _capture_log_snapshot 145 146 snap = _capture_log_snapshot("agent", tail_lines=10) 147 assert snap.full_text is not None 148 assert "session started" in snap.full_text 149 assert "session started" in snap.tail_text 150 151 def test_returns_none_for_missing(self, tmp_path, monkeypatch): 152 home = tmp_path / ".hermes" 153 home.mkdir() 154 monkeypatch.setenv("HERMES_HOME", str(home)) 155 156 from hermes_cli.debug import _capture_log_snapshot 157 snap = _capture_log_snapshot("agent", tail_lines=10) 158 assert snap.full_text is None 159 assert snap.tail_text == "(file not found)" 160 161 def test_empty_primary_reports_file_empty(self, hermes_home): 162 """Empty primary (no .1 fallback) surfaces as '(file empty)', not missing.""" 163 (hermes_home / "logs" / "agent.log").write_text("") 164 165 from hermes_cli.debug import _capture_log_snapshot 166 snap = _capture_log_snapshot("agent", tail_lines=10) 167 assert snap.full_text is None 168 assert snap.tail_text == "(file empty)" 169 170 def test_race_truncate_after_resolve_reports_empty(self, hermes_home, monkeypatch): 171 """If the log is truncated between resolve and stat, say 'empty', not 'missing'.""" 172 log_path = hermes_home / "logs" / "agent.log" 173 from hermes_cli import debug 174 175 monkeypatch.setattr(debug, "_resolve_log_path", lambda _name: log_path) 176 log_path.write_text("") 177 178 snap = debug._capture_log_snapshot("agent", tail_lines=10) 179 assert snap.path == log_path 180 assert snap.full_text is None 181 assert snap.tail_text == "(file empty)" 182 183 def test_truncates_large_file(self, hermes_home): 184 """Files larger than max_bytes get tail-truncated.""" 185 from hermes_cli.debug import _capture_log_snapshot 186 187 # Write a file larger than 1KB 188 big_content = "x" * 100 + "\n" 189 (hermes_home / "logs" / "agent.log").write_text(big_content * 200) 190 191 snap = _capture_log_snapshot("agent", tail_lines=10, max_bytes=1024) 192 assert snap.full_text is not None 193 assert "truncated" in snap.full_text 194 195 def test_keeps_first_line_when_truncation_on_boundary(self, hermes_home): 196 """When truncation lands on a line boundary, keep the first full line.""" 197 from hermes_cli.debug import _capture_log_snapshot 198 199 # File must exceed the initial chunk_size (8192) used by the 200 # backward-reading loop so the truncation path actually fires. 201 line = "A" * 99 + "\n" # 100 bytes per line 202 num_lines = 200 # 20000 bytes 203 (hermes_home / "logs" / "agent.log").write_text(line * num_lines) 204 205 # max_bytes = 1000 = 100 * 10 → cut at byte 20000 - 1000 = 19000, 206 # and byte 19000 - 1 is '\n'. Boundary hit → keep all 10 lines. 207 snap = _capture_log_snapshot("agent", tail_lines=5, max_bytes=1000) 208 assert snap.full_text is not None 209 assert "truncated" in snap.full_text 210 raw = snap.full_text.split("\n", 1)[1] 211 kept = [l for l in raw.strip().splitlines() if l.startswith("A")] 212 assert len(kept) == 10 213 214 def test_drops_partial_when_truncation_mid_line(self, hermes_home): 215 """When truncation lands mid-line, drop the partial fragment.""" 216 from hermes_cli.debug import _capture_log_snapshot 217 218 line = "A" * 99 + "\n" # 100 bytes per line 219 num_lines = 200 # 20000 bytes 220 (hermes_home / "logs" / "agent.log").write_text(line * num_lines) 221 222 # max_bytes = 950 doesn't divide evenly into 100 → mid-line cut. 223 snap = _capture_log_snapshot("agent", tail_lines=5, max_bytes=950) 224 assert snap.full_text is not None 225 assert "truncated" in snap.full_text 226 raw = snap.full_text.split("\n", 1)[1] 227 kept = [l for l in raw.strip().splitlines() if l.startswith("A")] 228 # 950 / 100 = 9.5 → 9 complete lines after dropping partial 229 assert len(kept) == 9 230 231 def test_unknown_log_returns_none(self, hermes_home): 232 from hermes_cli.debug import _capture_log_snapshot 233 snap = _capture_log_snapshot("nonexistent", tail_lines=10) 234 assert snap.full_text is None 235 236 def test_falls_back_to_rotated_file(self, hermes_home): 237 """When gateway.log doesn't exist, falls back to gateway.log.1.""" 238 from hermes_cli.debug import _capture_log_snapshot 239 240 logs_dir = hermes_home / "logs" 241 # Remove the primary (if any) and create a .1 rotation 242 (logs_dir / "gateway.log").unlink(missing_ok=True) 243 (logs_dir / "gateway.log.1").write_text( 244 "2026-04-12 10:00:00 INFO gateway.run: rotated content\n" 245 ) 246 247 snap = _capture_log_snapshot("gateway", tail_lines=10) 248 assert snap.full_text is not None 249 assert "rotated content" in snap.full_text 250 251 def test_prefers_primary_over_rotated(self, hermes_home): 252 """Primary log is used when it exists, even if .1 also exists.""" 253 from hermes_cli.debug import _capture_log_snapshot 254 255 logs_dir = hermes_home / "logs" 256 (logs_dir / "gateway.log").write_text("primary content\n") 257 (logs_dir / "gateway.log.1").write_text("rotated content\n") 258 259 snap = _capture_log_snapshot("gateway", tail_lines=10) 260 assert "primary content" in snap.full_text 261 assert "rotated" not in snap.full_text 262 263 def test_falls_back_when_primary_empty(self, hermes_home): 264 """Empty primary log falls back to .1 rotation.""" 265 from hermes_cli.debug import _capture_log_snapshot 266 267 logs_dir = hermes_home / "logs" 268 (logs_dir / "agent.log").write_text("") 269 (logs_dir / "agent.log.1").write_text("rotated agent data\n") 270 271 snap = _capture_log_snapshot("agent", tail_lines=10) 272 assert snap.full_text is not None 273 assert "rotated agent data" in snap.full_text 274 275 276 # --------------------------------------------------------------------------- 277 # Capture log redaction (force=True applies regardless of HERMES_REDACT_SECRETS) 278 # --------------------------------------------------------------------------- 279 280 # A vendor-prefixed token used across redaction tests. Long enough to clear 281 # the redactor's `floor` parameter so it actually masks rather than fully blanks. 282 _REDACT_FIXTURE_TOKEN = "sk-proj-A1B2C3D4E5F6G7H8I9J0aA" 283 284 285 class TestCaptureLogSnapshotRedaction: 286 """Pin upload-time redaction at the _capture_log_snapshot boundary.""" 287 288 @pytest.fixture 289 def hermes_home_with_secret(self, tmp_path, monkeypatch): 290 """Isolated HERMES_HOME whose agent.log contains a vendor-prefixed token.""" 291 home = tmp_path / ".hermes" 292 home.mkdir() 293 monkeypatch.setenv("HERMES_HOME", str(home)) 294 # Critical: ensure the user has NOT opted in to redaction. The whole 295 # point of this PR is that share-time redaction works for users who 296 # never set this env var. 297 monkeypatch.delenv("HERMES_REDACT_SECRETS", raising=False) 298 299 logs_dir = home / "logs" 300 logs_dir.mkdir() 301 (logs_dir / "agent.log").write_text( 302 f"2026-04-12 17:00:00 INFO config: api_key={_REDACT_FIXTURE_TOKEN} loaded\n" 303 ) 304 (logs_dir / "errors.log").write_text("") 305 (logs_dir / "gateway.log").write_text("") 306 return home 307 308 def test_default_redacts_tail_and_full_text(self, hermes_home_with_secret): 309 from hermes_cli.debug import _capture_log_snapshot 310 311 snap = _capture_log_snapshot("agent", tail_lines=10) 312 313 # Both views the upload uses must be sanitized. 314 assert _REDACT_FIXTURE_TOKEN not in snap.tail_text 315 assert snap.full_text is not None 316 assert _REDACT_FIXTURE_TOKEN not in snap.full_text 317 318 def test_redact_false_passes_through(self, hermes_home_with_secret): 319 from hermes_cli.debug import _capture_log_snapshot 320 321 snap = _capture_log_snapshot("agent", tail_lines=10, redact=False) 322 323 # Original token survives when the caller opts out. 324 assert _REDACT_FIXTURE_TOKEN in snap.tail_text 325 assert _REDACT_FIXTURE_TOKEN in (snap.full_text or "") 326 327 def test_force_true_overrides_unset_env_var(self, hermes_home_with_secret): 328 """Regression test: redact_sensitive_text short-circuits without force=True. 329 330 If a future refactor drops `force=True` from `_redact_log_text`, this 331 test fails immediately. Without `force=True`, the redactor returns the 332 input unchanged when HERMES_REDACT_SECRETS is unset, and the feature 333 ships silently broken for its target audience. 334 """ 335 import os 336 337 from hermes_cli.debug import _capture_log_snapshot 338 339 # Belt-and-suspenders: confirm the env var is genuinely unset for this 340 # test so we know we're exercising the force=True path. 341 assert os.environ.get("HERMES_REDACT_SECRETS", "") == "" 342 343 snap = _capture_log_snapshot("agent", tail_lines=10) 344 345 assert _REDACT_FIXTURE_TOKEN not in snap.tail_text 346 assert snap.full_text is not None 347 assert _REDACT_FIXTURE_TOKEN not in snap.full_text 348 349 def test_capture_default_log_snapshots_threads_redact( 350 self, hermes_home_with_secret 351 ): 352 from hermes_cli.debug import _capture_default_log_snapshots 353 354 snaps = _capture_default_log_snapshots(50) 355 356 # Default threads redact=True to all three captured logs. 357 assert _REDACT_FIXTURE_TOKEN not in snaps["agent"].tail_text 358 assert _REDACT_FIXTURE_TOKEN not in (snaps["agent"].full_text or "") 359 360 def test_capture_default_log_snapshots_no_redact_passes_through( 361 self, hermes_home_with_secret 362 ): 363 from hermes_cli.debug import _capture_default_log_snapshots 364 365 snaps = _capture_default_log_snapshots(50, redact=False) 366 367 assert _REDACT_FIXTURE_TOKEN in snaps["agent"].tail_text 368 assert _REDACT_FIXTURE_TOKEN in (snaps["agent"].full_text or "") 369 370 371 # --------------------------------------------------------------------------- 372 # Debug report collection 373 # --------------------------------------------------------------------------- 374 375 class TestCollectDebugReport: 376 """Test the debug report builder.""" 377 378 def test_report_includes_dump_output(self, hermes_home): 379 from hermes_cli.debug import collect_debug_report 380 381 with patch("hermes_cli.dump.run_dump") as mock_dump: 382 mock_dump.side_effect = lambda args: print( 383 "--- hermes dump ---\nversion: 0.8.0\n--- end dump ---" 384 ) 385 report = collect_debug_report(log_lines=50) 386 387 assert "--- hermes dump ---" in report 388 assert "version: 0.8.0" in report 389 390 def test_report_includes_agent_log(self, hermes_home): 391 from hermes_cli.debug import collect_debug_report 392 393 with patch("hermes_cli.dump.run_dump"): 394 report = collect_debug_report(log_lines=50) 395 396 assert "--- agent.log" in report 397 assert "session started" in report 398 399 def test_report_includes_errors_log(self, hermes_home): 400 from hermes_cli.debug import collect_debug_report 401 402 with patch("hermes_cli.dump.run_dump"): 403 report = collect_debug_report(log_lines=50) 404 405 assert "--- errors.log" in report 406 assert "connection lost" in report 407 408 def test_report_includes_gateway_log(self, hermes_home): 409 from hermes_cli.debug import collect_debug_report 410 411 with patch("hermes_cli.dump.run_dump"): 412 report = collect_debug_report(log_lines=50) 413 414 assert "--- gateway.log" in report 415 416 def test_missing_logs_handled(self, tmp_path, monkeypatch): 417 home = tmp_path / ".hermes" 418 home.mkdir() 419 monkeypatch.setenv("HERMES_HOME", str(home)) 420 421 from hermes_cli.debug import collect_debug_report 422 423 with patch("hermes_cli.dump.run_dump"): 424 report = collect_debug_report(log_lines=50) 425 426 assert "(file not found)" in report 427 428 429 # --------------------------------------------------------------------------- 430 # CLI entry point — run_debug_share 431 # --------------------------------------------------------------------------- 432 433 class TestRunDebugShare: 434 """Test the run_debug_share CLI handler.""" 435 436 def test_share_sweeps_expired_pastes(self, hermes_home, capsys): 437 """Slash-command path should sweep old pending deletes before uploading.""" 438 from hermes_cli.debug import run_debug_share 439 440 args = MagicMock() 441 args.lines = 50 442 args.expire = 7 443 args.local = False 444 445 with patch("hermes_cli.dump.run_dump"), \ 446 patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)) as mock_sweep, \ 447 patch("hermes_cli.debug.upload_to_pastebin", 448 return_value="https://paste.rs/test"): 449 run_debug_share(args) 450 451 mock_sweep.assert_called_once() 452 assert "Debug report uploaded" in capsys.readouterr().out 453 454 def test_share_survives_sweep_failure(self, hermes_home, capsys): 455 """Expired-paste cleanup is best-effort and must not block sharing.""" 456 from hermes_cli.debug import run_debug_share 457 458 args = MagicMock() 459 args.lines = 50 460 args.expire = 7 461 args.local = False 462 463 with patch("hermes_cli.dump.run_dump"), \ 464 patch( 465 "hermes_cli.debug._sweep_expired_pastes", 466 side_effect=RuntimeError("offline"), 467 ), \ 468 patch("hermes_cli.debug.upload_to_pastebin", 469 return_value="https://paste.rs/test"): 470 run_debug_share(args) 471 472 assert "https://paste.rs/test" in capsys.readouterr().out 473 474 def test_local_flag_prints_full_logs(self, hermes_home, capsys): 475 """--local prints the report plus full log contents.""" 476 from hermes_cli.debug import run_debug_share 477 478 args = MagicMock() 479 args.lines = 50 480 args.expire = 7 481 args.local = True 482 483 with patch("hermes_cli.dump.run_dump"): 484 run_debug_share(args) 485 486 out = capsys.readouterr().out 487 assert "--- agent.log" in out 488 assert "FULL agent.log" in out 489 assert "FULL gateway.log" in out 490 491 def test_share_uploads_three_pastes(self, hermes_home, capsys): 492 """Successful share uploads report + agent.log + gateway.log.""" 493 from hermes_cli.debug import run_debug_share 494 495 args = MagicMock() 496 args.lines = 50 497 args.expire = 7 498 args.local = False 499 500 call_count = [0] 501 uploaded_content = [] 502 def _mock_upload(content, expiry_days=7): 503 call_count[0] += 1 504 uploaded_content.append(content) 505 return f"https://paste.rs/paste{call_count[0]}" 506 507 with patch("hermes_cli.dump.run_dump") as mock_dump, \ 508 patch("hermes_cli.debug.upload_to_pastebin", 509 side_effect=_mock_upload): 510 mock_dump.side_effect = lambda a: print("--- hermes dump ---\nversion: test\n--- end dump ---") 511 run_debug_share(args) 512 513 out = capsys.readouterr().out 514 # Should have 3 uploads: report, agent.log, gateway.log 515 assert call_count[0] == 3 516 assert "paste.rs/paste1" in out # Report 517 assert "paste.rs/paste2" in out # agent.log 518 assert "paste.rs/paste3" in out # gateway.log 519 assert "Report" in out 520 assert "agent.log" in out 521 assert "gateway.log" in out 522 523 # Each log paste should start with the dump header 524 agent_paste = uploaded_content[1] 525 assert "--- hermes dump ---" in agent_paste 526 assert "--- full agent.log ---" in agent_paste 527 gateway_paste = uploaded_content[2] 528 assert "--- hermes dump ---" in gateway_paste 529 assert "--- full gateway.log ---" in gateway_paste 530 531 def test_share_keeps_report_and_full_log_on_same_snapshot(self, hermes_home, capsys): 532 """A mid-run rotation must not make full agent.log older than the report.""" 533 from hermes_cli.debug import run_debug_share, collect_debug_report as real_collect_debug_report 534 535 logs_dir = hermes_home / "logs" 536 (logs_dir / "agent.log").write_text( 537 "2026-04-22 12:00:00 INFO agent: newest line\n" 538 ) 539 (logs_dir / "agent.log.1").write_text( 540 "2026-04-10 12:00:00 INFO agent: old rotated line\n" 541 ) 542 543 args = MagicMock() 544 args.lines = 50 545 args.expire = 7 546 args.local = False 547 548 uploaded_content = [] 549 550 def _mock_upload(content, expiry_days=7): 551 uploaded_content.append(content) 552 return f"https://paste.rs/paste{len(uploaded_content)}" 553 554 def _wrapped_collect_debug_report(*, log_lines=200, dump_text="", log_snapshots=None): 555 report = real_collect_debug_report( 556 log_lines=log_lines, 557 dump_text=dump_text, 558 log_snapshots=log_snapshots, 559 ) 560 # Simulate the live log rotating after the report is built but 561 # before the old implementation would have re-read agent.log for 562 # standalone upload. 563 (logs_dir / "agent.log").write_text("") 564 (logs_dir / "agent.log.1").write_text( 565 "2026-04-10 12:00:00 INFO agent: old rotated line\n" 566 ) 567 return report 568 569 with patch("hermes_cli.dump.run_dump"), \ 570 patch("hermes_cli.debug.collect_debug_report", side_effect=_wrapped_collect_debug_report), \ 571 patch("hermes_cli.debug.upload_to_pastebin", side_effect=_mock_upload): 572 run_debug_share(args) 573 574 report_paste = uploaded_content[0] 575 agent_paste = uploaded_content[1] 576 assert "2026-04-22 12:00:00 INFO agent: newest line" in report_paste 577 assert "2026-04-22 12:00:00 INFO agent: newest line" in agent_paste 578 assert "old rotated line" not in agent_paste 579 580 def test_share_skips_missing_logs(self, tmp_path, monkeypatch, capsys): 581 """Only uploads logs that exist.""" 582 home = tmp_path / ".hermes" 583 home.mkdir() 584 monkeypatch.setenv("HERMES_HOME", str(home)) 585 586 from hermes_cli.debug import run_debug_share 587 588 args = MagicMock() 589 args.lines = 50 590 args.expire = 7 591 args.local = False 592 593 call_count = [0] 594 def _mock_upload(content, expiry_days=7): 595 call_count[0] += 1 596 return f"https://paste.rs/paste{call_count[0]}" 597 598 with patch("hermes_cli.dump.run_dump"), \ 599 patch("hermes_cli.debug.upload_to_pastebin", 600 side_effect=_mock_upload): 601 run_debug_share(args) 602 603 out = capsys.readouterr().out 604 # Only the report should be uploaded (no log files exist) 605 assert call_count[0] == 1 606 assert "Report" in out 607 608 def test_share_continues_on_log_upload_failure(self, hermes_home, capsys): 609 """Log upload failure doesn't stop the report from being shared.""" 610 from hermes_cli.debug import run_debug_share 611 612 args = MagicMock() 613 args.lines = 50 614 args.expire = 7 615 args.local = False 616 617 call_count = [0] 618 def _mock_upload(content, expiry_days=7): 619 call_count[0] += 1 620 if call_count[0] > 1: 621 raise RuntimeError("upload failed") 622 return "https://paste.rs/report" 623 624 with patch("hermes_cli.dump.run_dump"), \ 625 patch("hermes_cli.debug.upload_to_pastebin", 626 side_effect=_mock_upload): 627 run_debug_share(args) 628 629 out = capsys.readouterr().out 630 assert "Report" in out 631 assert "paste.rs/report" in out 632 assert "failed to upload" in out 633 634 def test_share_exits_on_report_upload_failure(self, hermes_home, capsys): 635 """If the main report fails to upload, exit with code 1.""" 636 from hermes_cli.debug import run_debug_share 637 638 args = MagicMock() 639 args.lines = 50 640 args.expire = 7 641 args.local = False 642 643 with patch("hermes_cli.dump.run_dump"), \ 644 patch("hermes_cli.debug.upload_to_pastebin", 645 side_effect=RuntimeError("all failed")): 646 with pytest.raises(SystemExit) as exc_info: 647 run_debug_share(args) 648 649 assert exc_info.value.code == 1 650 out = capsys.readouterr() 651 assert "all failed" in out.err 652 653 654 # --------------------------------------------------------------------------- 655 # Share-time redaction wiring + visible banner 656 # --------------------------------------------------------------------------- 657 658 class TestRunDebugShareRedaction: 659 """End-to-end: --no-redact flag, banner injection, default behavior.""" 660 661 @pytest.fixture 662 def hermes_home_with_secret(self, tmp_path, monkeypatch): 663 """Isolated HERMES_HOME whose agent.log contains a vendor-prefixed token.""" 664 home = tmp_path / ".hermes" 665 home.mkdir() 666 monkeypatch.setenv("HERMES_HOME", str(home)) 667 monkeypatch.delenv("HERMES_REDACT_SECRETS", raising=False) 668 669 logs_dir = home / "logs" 670 logs_dir.mkdir() 671 (logs_dir / "agent.log").write_text( 672 f"2026-04-12 17:00:00 INFO config: api_key={_REDACT_FIXTURE_TOKEN} loaded\n" 673 ) 674 (logs_dir / "errors.log").write_text("") 675 (logs_dir / "gateway.log").write_text( 676 f"2026-04-12 17:00:01 INFO gateway.run: token {_REDACT_FIXTURE_TOKEN}\n" 677 ) 678 return home 679 680 def test_default_share_redacts_uploaded_content( 681 self, hermes_home_with_secret, capsys 682 ): 683 """The uploaded report and full-log pastes do not contain the raw token.""" 684 from hermes_cli.debug import run_debug_share 685 686 args = MagicMock() 687 args.lines = 50 688 args.expire = 7 689 args.local = False 690 args.no_redact = False 691 692 captured: list[str] = [] 693 694 def fake_upload(content, expiry_days=7): 695 captured.append(content) 696 return f"https://paste.rs/{len(captured)}" 697 698 with patch("hermes_cli.dump.run_dump"), \ 699 patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)), \ 700 patch("hermes_cli.debug.upload_to_pastebin", side_effect=fake_upload): 701 run_debug_share(args) 702 703 # At least the report plus one full log paste reached the upload path. 704 assert len(captured) >= 2 705 for content in captured: 706 assert _REDACT_FIXTURE_TOKEN not in content, ( 707 "raw token leaked into upload-bound content" 708 ) 709 710 def test_default_share_includes_redaction_banner( 711 self, hermes_home_with_secret, capsys 712 ): 713 """Each upload-bound paste carries the visible redaction banner.""" 714 from hermes_cli.debug import run_debug_share 715 716 args = MagicMock() 717 args.lines = 50 718 args.expire = 7 719 args.local = False 720 args.no_redact = False 721 722 captured: list[str] = [] 723 724 def fake_upload(content, expiry_days=7): 725 captured.append(content) 726 return f"https://paste.rs/{len(captured)}" 727 728 with patch("hermes_cli.dump.run_dump"), \ 729 patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)), \ 730 patch("hermes_cli.debug.upload_to_pastebin", side_effect=fake_upload): 731 run_debug_share(args) 732 733 for content in captured: 734 assert "redacted at upload time" in content, ( 735 "redaction banner missing from upload-bound content" 736 ) 737 738 def test_no_redact_flag_disables_redaction_and_banner( 739 self, hermes_home_with_secret, capsys 740 ): 741 """--no-redact preserves original log content and omits the banner.""" 742 from hermes_cli.debug import run_debug_share 743 744 args = MagicMock() 745 args.lines = 50 746 args.expire = 7 747 args.local = False 748 args.no_redact = True 749 750 captured: list[str] = [] 751 752 def fake_upload(content, expiry_days=7): 753 captured.append(content) 754 return f"https://paste.rs/{len(captured)}" 755 756 with patch("hermes_cli.dump.run_dump"), \ 757 patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)), \ 758 patch("hermes_cli.debug.upload_to_pastebin", side_effect=fake_upload): 759 run_debug_share(args) 760 761 # The agent.log paste should now contain the raw token. 762 assert any(_REDACT_FIXTURE_TOKEN in c for c in captured), ( 763 "expected raw token in --no-redact upload" 764 ) 765 # No banner anywhere when redaction is disabled. 766 for content in captured: 767 assert "redacted at upload time" not in content, ( 768 "banner present with --no-redact" 769 ) 770 771 772 # --------------------------------------------------------------------------- 773 # run_debug router 774 # --------------------------------------------------------------------------- 775 776 class TestRunDebug: 777 def test_no_subcommand_shows_usage(self, capsys): 778 from hermes_cli.debug import run_debug 779 780 args = MagicMock() 781 args.debug_command = None 782 783 run_debug(args) 784 785 out = capsys.readouterr().out 786 assert "hermes debug" in out 787 assert "share" in out 788 assert "delete" in out 789 790 def test_share_subcommand_routes(self, hermes_home): 791 from hermes_cli.debug import run_debug 792 793 args = MagicMock() 794 args.debug_command = "share" 795 args.lines = 200 796 args.expire = 7 797 args.local = True 798 799 with patch("hermes_cli.dump.run_dump"): 800 run_debug(args) 801 802 803 # --------------------------------------------------------------------------- 804 # Argparse integration 805 # --------------------------------------------------------------------------- 806 807 # --------------------------------------------------------------------------- 808 # Delete / auto-delete 809 # --------------------------------------------------------------------------- 810 811 class TestExtractPasteId: 812 def test_paste_rs_url(self): 813 from hermes_cli.debug import _extract_paste_id 814 assert _extract_paste_id("https://paste.rs/abc123") == "abc123" 815 816 def test_paste_rs_trailing_slash(self): 817 from hermes_cli.debug import _extract_paste_id 818 assert _extract_paste_id("https://paste.rs/abc123/") == "abc123" 819 820 def test_http_variant(self): 821 from hermes_cli.debug import _extract_paste_id 822 assert _extract_paste_id("http://paste.rs/xyz") == "xyz" 823 824 def test_non_paste_rs_returns_none(self): 825 from hermes_cli.debug import _extract_paste_id 826 assert _extract_paste_id("https://dpaste.com/ABCDEF") is None 827 828 def test_empty_returns_none(self): 829 from hermes_cli.debug import _extract_paste_id 830 assert _extract_paste_id("") is None 831 832 833 class TestDeletePaste: 834 def test_delete_sends_delete_request(self): 835 from hermes_cli.debug import delete_paste 836 837 mock_resp = MagicMock() 838 mock_resp.status = 200 839 mock_resp.__enter__ = lambda s: s 840 mock_resp.__exit__ = MagicMock(return_value=False) 841 842 with patch("hermes_cli.debug.urllib.request.urlopen", 843 return_value=mock_resp) as mock_open: 844 result = delete_paste("https://paste.rs/abc123") 845 846 assert result is True 847 req = mock_open.call_args[0][0] 848 assert req.method == "DELETE" 849 assert "paste.rs/abc123" in req.full_url 850 851 def test_delete_rejects_non_paste_rs(self): 852 from hermes_cli.debug import delete_paste 853 854 with pytest.raises(ValueError, match="only paste.rs"): 855 delete_paste("https://dpaste.com/something") 856 857 858 class TestScheduleAutoDelete: 859 """``_schedule_auto_delete`` used to spawn a detached Python subprocess 860 per call (one per paste URL batch). Those subprocesses slept 6 hours 861 and accumulated forever under repeated use — 15+ orphaned interpreters 862 were observed in production. 863 864 The new implementation is stateless: it records pending deletions to 865 ``~/.hermes/pastes/pending.json`` and lets ``_sweep_expired_pastes`` 866 handle the DELETE requests synchronously on the next ``hermes debug`` 867 invocation. 868 """ 869 870 def test_does_not_spawn_subprocess(self, hermes_home): 871 """Regression guard: _schedule_auto_delete must NEVER spawn subprocesses. 872 873 We assert this structurally rather than by mocking Popen: the new 874 implementation doesn't even import ``subprocess`` at module scope, 875 so a mock patch wouldn't find it. 876 """ 877 import ast 878 import inspect 879 from hermes_cli.debug import _schedule_auto_delete 880 881 # Strip the docstring before scanning so the regression-rationale 882 # prose inside it doesn't trigger our banned-word checks. 883 source = inspect.getsource(_schedule_auto_delete) 884 tree = ast.parse(source) 885 func_node = tree.body[0] 886 if ( 887 func_node.body 888 and isinstance(func_node.body[0], ast.Expr) 889 and isinstance(func_node.body[0].value, ast.Constant) 890 and isinstance(func_node.body[0].value.value, str) 891 ): 892 func_node.body = func_node.body[1:] 893 code_only = ast.unparse(func_node) 894 895 assert "Popen" not in code_only, ( 896 "_schedule_auto_delete must not spawn subprocesses — " 897 "use pending.json + _sweep_expired_pastes instead" 898 ) 899 assert "subprocess" not in code_only, ( 900 "_schedule_auto_delete must not reference subprocess at all" 901 ) 902 assert "time.sleep" not in code_only, ( 903 "Regression: sleeping in _schedule_auto_delete is the bug being fixed" 904 ) 905 906 # And verify that calling it doesn't produce any orphaned children 907 # (it should just write pending.json synchronously). 908 import os as _os 909 before = set(_os.listdir("/proc")) if _os.path.exists("/proc") else None 910 _schedule_auto_delete( 911 ["https://paste.rs/abc", "https://paste.rs/def"], 912 delay_seconds=10, 913 ) 914 if before is not None: 915 after = set(_os.listdir("/proc")) 916 new = after - before 917 # Filter to only integer-named entries (process PIDs) 918 new_pids = [p for p in new if p.isdigit()] 919 # It's fine if unrelated processes appeared — we just need to make 920 # sure we didn't spawn a long-sleeping one. The old bug spawned 921 # a python interpreter whose cmdline contained "time.sleep". 922 for pid in new_pids: 923 try: 924 with open(f"/proc/{pid}/cmdline", "rb") as f: 925 cmdline = f.read().decode("utf-8", errors="replace") 926 assert "time.sleep" not in cmdline, ( 927 f"Leaked sleeper subprocess PID {pid}: {cmdline}" 928 ) 929 except OSError: 930 pass # process exited already 931 932 def test_records_pending_to_json(self, hermes_home): 933 """Scheduled URLs are persisted to pending.json with expiration.""" 934 from hermes_cli.debug import _schedule_auto_delete, _pending_file 935 import json 936 937 _schedule_auto_delete( 938 ["https://paste.rs/abc", "https://paste.rs/def"], 939 delay_seconds=10, 940 ) 941 942 pending_path = _pending_file() 943 assert pending_path.exists() 944 945 entries = json.loads(pending_path.read_text()) 946 assert len(entries) == 2 947 urls = {e["url"] for e in entries} 948 assert urls == {"https://paste.rs/abc", "https://paste.rs/def"} 949 950 # expire_at is ~now + delay_seconds 951 import time 952 for e in entries: 953 assert e["expire_at"] > time.time() 954 assert e["expire_at"] <= time.time() + 15 955 956 def test_skips_non_paste_rs_urls(self, hermes_home): 957 """dpaste.com URLs auto-expire — don't track them.""" 958 from hermes_cli.debug import _schedule_auto_delete, _pending_file 959 960 _schedule_auto_delete(["https://dpaste.com/something"]) 961 962 # pending.json should not be created for non-paste.rs URLs 963 assert not _pending_file().exists() 964 965 def test_merges_with_existing_pending(self, hermes_home): 966 """Subsequent calls merge into existing pending.json.""" 967 from hermes_cli.debug import _schedule_auto_delete, _load_pending 968 969 _schedule_auto_delete(["https://paste.rs/first"], delay_seconds=10) 970 _schedule_auto_delete(["https://paste.rs/second"], delay_seconds=10) 971 972 entries = _load_pending() 973 urls = {e["url"] for e in entries} 974 assert urls == {"https://paste.rs/first", "https://paste.rs/second"} 975 976 def test_dedupes_same_url(self, hermes_home): 977 """Same URL recorded twice → one entry with the later expire_at.""" 978 from hermes_cli.debug import _schedule_auto_delete, _load_pending 979 980 _schedule_auto_delete(["https://paste.rs/dup"], delay_seconds=10) 981 _schedule_auto_delete(["https://paste.rs/dup"], delay_seconds=100) 982 983 entries = _load_pending() 984 assert len(entries) == 1 985 assert entries[0]["url"] == "https://paste.rs/dup" 986 987 988 class TestSweepExpiredPastes: 989 """Test the opportunistic sweep that replaces the sleeping subprocess.""" 990 991 def test_sweep_empty_is_noop(self, hermes_home): 992 from hermes_cli.debug import _sweep_expired_pastes 993 994 deleted, remaining = _sweep_expired_pastes() 995 assert deleted == 0 996 assert remaining == 0 997 998 def test_sweep_deletes_expired_entries(self, hermes_home): 999 from hermes_cli.debug import ( 1000 _sweep_expired_pastes, 1001 _save_pending, 1002 _load_pending, 1003 ) 1004 import time 1005 1006 # Seed pending.json with one expired + one future entry 1007 _save_pending([ 1008 {"url": "https://paste.rs/expired", "expire_at": time.time() - 100}, 1009 {"url": "https://paste.rs/future", "expire_at": time.time() + 3600}, 1010 ]) 1011 1012 delete_calls = [] 1013 1014 def fake_delete(url): 1015 delete_calls.append(url) 1016 return True 1017 1018 with patch("hermes_cli.debug.delete_paste", side_effect=fake_delete): 1019 deleted, remaining = _sweep_expired_pastes() 1020 1021 assert delete_calls == ["https://paste.rs/expired"] 1022 assert deleted == 1 1023 assert remaining == 1 1024 1025 entries = _load_pending() 1026 urls = {e["url"] for e in entries} 1027 assert urls == {"https://paste.rs/future"} 1028 1029 def test_sweep_leaves_future_entries_alone(self, hermes_home): 1030 from hermes_cli.debug import _sweep_expired_pastes, _save_pending 1031 import time 1032 1033 _save_pending([ 1034 {"url": "https://paste.rs/future1", "expire_at": time.time() + 3600}, 1035 {"url": "https://paste.rs/future2", "expire_at": time.time() + 7200}, 1036 ]) 1037 1038 with patch("hermes_cli.debug.delete_paste") as mock_delete: 1039 deleted, remaining = _sweep_expired_pastes() 1040 1041 mock_delete.assert_not_called() 1042 assert deleted == 0 1043 assert remaining == 2 1044 1045 def test_sweep_survives_network_failure(self, hermes_home): 1046 """Failed DELETEs stay in pending.json until the 24h grace window.""" 1047 from hermes_cli.debug import ( 1048 _sweep_expired_pastes, 1049 _save_pending, 1050 _load_pending, 1051 ) 1052 import time 1053 1054 _save_pending([ 1055 {"url": "https://paste.rs/flaky", "expire_at": time.time() - 100}, 1056 ]) 1057 1058 with patch( 1059 "hermes_cli.debug.delete_paste", 1060 side_effect=Exception("network down"), 1061 ): 1062 deleted, remaining = _sweep_expired_pastes() 1063 1064 # Failure within 24h grace → kept for retry 1065 assert deleted == 0 1066 assert remaining == 1 1067 assert len(_load_pending()) == 1 1068 1069 def test_sweep_drops_entries_past_grace_window(self, hermes_home): 1070 """After 24h past expiration, give up even on network failures.""" 1071 from hermes_cli.debug import ( 1072 _sweep_expired_pastes, 1073 _save_pending, 1074 _load_pending, 1075 ) 1076 import time 1077 1078 # Expired 25 hours ago → past the 24h grace window 1079 very_old = time.time() - (25 * 3600) 1080 _save_pending([ 1081 {"url": "https://paste.rs/ancient", "expire_at": very_old}, 1082 ]) 1083 1084 with patch( 1085 "hermes_cli.debug.delete_paste", 1086 side_effect=Exception("network down"), 1087 ): 1088 deleted, remaining = _sweep_expired_pastes() 1089 1090 assert deleted == 1 1091 assert remaining == 0 1092 assert _load_pending() == [] 1093 1094 1095 class TestRunDebugSweepsOnInvocation: 1096 """``run_debug`` must sweep expired pastes on every invocation.""" 1097 1098 def test_run_debug_calls_sweep(self, hermes_home): 1099 from hermes_cli.debug import run_debug 1100 1101 args = MagicMock() 1102 args.debug_command = None # default → prints help 1103 1104 with patch("hermes_cli.debug._sweep_expired_pastes") as mock_sweep: 1105 run_debug(args) 1106 1107 mock_sweep.assert_called_once() 1108 1109 def test_run_debug_survives_sweep_failure(self, hermes_home, capsys): 1110 """If the sweep throws, the subcommand still runs.""" 1111 from hermes_cli.debug import run_debug 1112 1113 args = MagicMock() 1114 args.debug_command = None 1115 1116 with patch( 1117 "hermes_cli.debug._sweep_expired_pastes", 1118 side_effect=RuntimeError("boom"), 1119 ): 1120 run_debug(args) # must not raise 1121 1122 # Default subcommand still printed help 1123 out = capsys.readouterr().out 1124 assert "Usage: hermes debug" in out 1125 1126 1127 class TestRunDebugDelete: 1128 def test_deletes_valid_url(self, capsys): 1129 from hermes_cli.debug import run_debug_delete 1130 1131 args = MagicMock() 1132 args.urls = ["https://paste.rs/abc"] 1133 1134 with patch("hermes_cli.debug.delete_paste", return_value=True): 1135 run_debug_delete(args) 1136 1137 out = capsys.readouterr().out 1138 assert "Deleted" in out 1139 assert "paste.rs/abc" in out 1140 1141 def test_handles_delete_failure(self, capsys): 1142 from hermes_cli.debug import run_debug_delete 1143 1144 args = MagicMock() 1145 args.urls = ["https://paste.rs/abc"] 1146 1147 with patch("hermes_cli.debug.delete_paste", 1148 side_effect=Exception("network error")): 1149 run_debug_delete(args) 1150 1151 out = capsys.readouterr().out 1152 assert "Could not delete" in out 1153 1154 def test_no_urls_shows_usage(self, capsys): 1155 from hermes_cli.debug import run_debug_delete 1156 1157 args = MagicMock() 1158 args.urls = [] 1159 1160 run_debug_delete(args) 1161 1162 out = capsys.readouterr().out 1163 assert "Usage" in out 1164 1165 1166 class TestShareIncludesAutoDelete: 1167 """Verify that run_debug_share schedules auto-deletion and prints TTL.""" 1168 1169 def test_share_schedules_auto_delete(self, hermes_home, capsys): 1170 from hermes_cli.debug import run_debug_share 1171 1172 args = MagicMock() 1173 args.lines = 50 1174 args.expire = 7 1175 args.local = False 1176 1177 with patch("hermes_cli.dump.run_dump"), \ 1178 patch("hermes_cli.debug.upload_to_pastebin", 1179 return_value="https://paste.rs/test1"), \ 1180 patch("hermes_cli.debug._schedule_auto_delete") as mock_sched: 1181 run_debug_share(args) 1182 1183 # auto-delete was scheduled with the uploaded URLs 1184 mock_sched.assert_called_once() 1185 urls_arg = mock_sched.call_args[0][0] 1186 assert "https://paste.rs/test1" in urls_arg 1187 1188 out = capsys.readouterr().out 1189 assert "auto-delete" in out 1190 1191 def test_share_shows_privacy_notice(self, hermes_home, capsys): 1192 from hermes_cli.debug import run_debug_share 1193 1194 args = MagicMock() 1195 args.lines = 50 1196 args.expire = 7 1197 args.local = False 1198 1199 with patch("hermes_cli.dump.run_dump"), \ 1200 patch("hermes_cli.debug.upload_to_pastebin", 1201 return_value="https://paste.rs/test"), \ 1202 patch("hermes_cli.debug._schedule_auto_delete"): 1203 run_debug_share(args) 1204 1205 out = capsys.readouterr().out 1206 assert "public paste service" in out 1207 1208 def test_local_no_privacy_notice(self, hermes_home, capsys): 1209 from hermes_cli.debug import run_debug_share 1210 1211 args = MagicMock() 1212 args.lines = 50 1213 args.expire = 7 1214 args.local = True 1215 1216 with patch("hermes_cli.dump.run_dump"): 1217 run_debug_share(args) 1218 1219 out = capsys.readouterr().out 1220 assert "public paste service" not in out