| """Tests for ``hermes debug`` CLI command and debug utilities.""" |
|
|
| import os |
| import sys |
| import urllib.error |
| from pathlib import Path |
| from unittest.mock import MagicMock, patch, call |
|
|
| import pytest |
|
|
| |
| |
| |
|
|
| @pytest.fixture |
| def hermes_home(tmp_path, monkeypatch): |
| """Set up an isolated HERMES_HOME with minimal logs.""" |
| home = tmp_path / ".hermes" |
| home.mkdir() |
| monkeypatch.setenv("HERMES_HOME", str(home)) |
|
|
| |
| logs_dir = home / "logs" |
| logs_dir.mkdir() |
| (logs_dir / "agent.log").write_text( |
| "2026-04-12 17:00:00 INFO agent: session started\n" |
| "2026-04-12 17:00:01 INFO tools.terminal: running ls\n" |
| "2026-04-12 17:00:02 WARNING agent: high token usage\n" |
| ) |
| (logs_dir / "errors.log").write_text( |
| "2026-04-12 17:00:05 ERROR gateway.run: connection lost\n" |
| ) |
| (logs_dir / "gateway.log").write_text( |
| "2026-04-12 17:00:10 INFO gateway.run: started\n" |
| ) |
|
|
| return home |
|
|
|
|
| |
| |
| |
|
|
| class TestUploadPasteRs: |
| """Test paste.rs upload path.""" |
|
|
| def test_upload_paste_rs_success(self): |
| from hermes_cli.debug import _upload_paste_rs |
|
|
| mock_resp = MagicMock() |
| mock_resp.read.return_value = b"https://paste.rs/abc123\n" |
| mock_resp.__enter__ = lambda s: s |
| mock_resp.__exit__ = MagicMock(return_value=False) |
|
|
| with patch("hermes_cli.debug.urllib.request.urlopen", return_value=mock_resp): |
| url = _upload_paste_rs("hello world") |
|
|
| assert url == "https://paste.rs/abc123" |
|
|
| def test_upload_paste_rs_bad_response(self): |
| from hermes_cli.debug import _upload_paste_rs |
|
|
| mock_resp = MagicMock() |
| mock_resp.read.return_value = b"<html>error</html>" |
| mock_resp.__enter__ = lambda s: s |
| mock_resp.__exit__ = MagicMock(return_value=False) |
|
|
| with patch("hermes_cli.debug.urllib.request.urlopen", return_value=mock_resp): |
| with pytest.raises(ValueError, match="Unexpected response"): |
| _upload_paste_rs("test") |
|
|
| def test_upload_paste_rs_network_error(self): |
| from hermes_cli.debug import _upload_paste_rs |
|
|
| with patch( |
| "hermes_cli.debug.urllib.request.urlopen", |
| side_effect=urllib.error.URLError("connection refused"), |
| ): |
| with pytest.raises(urllib.error.URLError): |
| _upload_paste_rs("test") |
|
|
|
|
| class TestUploadDpasteCom: |
| """Test dpaste.com fallback upload path.""" |
|
|
| def test_upload_dpaste_com_success(self): |
| from hermes_cli.debug import _upload_dpaste_com |
|
|
| mock_resp = MagicMock() |
| mock_resp.read.return_value = b"https://dpaste.com/ABCDEFG\n" |
| mock_resp.__enter__ = lambda s: s |
| mock_resp.__exit__ = MagicMock(return_value=False) |
|
|
| with patch("hermes_cli.debug.urllib.request.urlopen", return_value=mock_resp): |
| url = _upload_dpaste_com("hello world", expiry_days=7) |
|
|
| assert url == "https://dpaste.com/ABCDEFG" |
|
|
|
|
| class TestUploadToPastebin: |
| """Test the combined upload with fallback.""" |
|
|
| def test_tries_paste_rs_first(self): |
| from hermes_cli.debug import upload_to_pastebin |
|
|
| with patch("hermes_cli.debug._upload_paste_rs", |
| return_value="https://paste.rs/test") as prs: |
| url = upload_to_pastebin("content") |
|
|
| assert url == "https://paste.rs/test" |
| prs.assert_called_once() |
|
|
| def test_falls_back_to_dpaste_com(self): |
| from hermes_cli.debug import upload_to_pastebin |
|
|
| with patch("hermes_cli.debug._upload_paste_rs", |
| side_effect=Exception("down")), \ |
| patch("hermes_cli.debug._upload_dpaste_com", |
| return_value="https://dpaste.com/TEST") as dp: |
| url = upload_to_pastebin("content") |
|
|
| assert url == "https://dpaste.com/TEST" |
| dp.assert_called_once() |
|
|
| def test_raises_when_both_fail(self): |
| from hermes_cli.debug import upload_to_pastebin |
|
|
| with patch("hermes_cli.debug._upload_paste_rs", |
| side_effect=Exception("err1")), \ |
| patch("hermes_cli.debug._upload_dpaste_com", |
| side_effect=Exception("err2")): |
| with pytest.raises(RuntimeError, match="Failed to upload"): |
| upload_to_pastebin("content") |
|
|
|
|
| |
| |
| |
|
|
| class TestCaptureLogSnapshot: |
| """Test _capture_log_snapshot for log reading and truncation.""" |
|
|
| def test_reads_small_file(self, hermes_home): |
| from hermes_cli.debug import _capture_log_snapshot |
|
|
| snap = _capture_log_snapshot("agent", tail_lines=10) |
| assert snap.full_text is not None |
| assert "session started" in snap.full_text |
| assert "session started" in snap.tail_text |
|
|
| def test_returns_none_for_missing(self, tmp_path, monkeypatch): |
| home = tmp_path / ".hermes" |
| home.mkdir() |
| monkeypatch.setenv("HERMES_HOME", str(home)) |
|
|
| from hermes_cli.debug import _capture_log_snapshot |
| snap = _capture_log_snapshot("agent", tail_lines=10) |
| assert snap.full_text is None |
| assert snap.tail_text == "(file not found)" |
|
|
| def test_empty_primary_reports_file_empty(self, hermes_home): |
| """Empty primary (no .1 fallback) surfaces as '(file empty)', not missing.""" |
| (hermes_home / "logs" / "agent.log").write_text("") |
|
|
| from hermes_cli.debug import _capture_log_snapshot |
| snap = _capture_log_snapshot("agent", tail_lines=10) |
| assert snap.full_text is None |
| assert snap.tail_text == "(file empty)" |
|
|
| def test_race_truncate_after_resolve_reports_empty(self, hermes_home, monkeypatch): |
| """If the log is truncated between resolve and stat, say 'empty', not 'missing'.""" |
| log_path = hermes_home / "logs" / "agent.log" |
| from hermes_cli import debug |
|
|
| monkeypatch.setattr(debug, "_resolve_log_path", lambda _name: log_path) |
| log_path.write_text("") |
|
|
| snap = debug._capture_log_snapshot("agent", tail_lines=10) |
| assert snap.path == log_path |
| assert snap.full_text is None |
| assert snap.tail_text == "(file empty)" |
|
|
| def test_truncates_large_file(self, hermes_home): |
| """Files larger than max_bytes get tail-truncated.""" |
| from hermes_cli.debug import _capture_log_snapshot |
|
|
| |
| big_content = "x" * 100 + "\n" |
| (hermes_home / "logs" / "agent.log").write_text(big_content * 200) |
|
|
| snap = _capture_log_snapshot("agent", tail_lines=10, max_bytes=1024) |
| assert snap.full_text is not None |
| assert "truncated" in snap.full_text |
|
|
| def test_keeps_first_line_when_truncation_on_boundary(self, hermes_home): |
| """When truncation lands on a line boundary, keep the first full line.""" |
| from hermes_cli.debug import _capture_log_snapshot |
|
|
| |
| |
| line = "A" * 99 + "\n" |
| num_lines = 200 |
| (hermes_home / "logs" / "agent.log").write_text(line * num_lines) |
|
|
| |
| |
| snap = _capture_log_snapshot("agent", tail_lines=5, max_bytes=1000) |
| assert snap.full_text is not None |
| assert "truncated" in snap.full_text |
| raw = snap.full_text.split("\n", 1)[1] |
| kept = [l for l in raw.strip().splitlines() if l.startswith("A")] |
| assert len(kept) == 10 |
|
|
| def test_drops_partial_when_truncation_mid_line(self, hermes_home): |
| """When truncation lands mid-line, drop the partial fragment.""" |
| from hermes_cli.debug import _capture_log_snapshot |
|
|
| line = "A" * 99 + "\n" |
| num_lines = 200 |
| (hermes_home / "logs" / "agent.log").write_text(line * num_lines) |
|
|
| |
| snap = _capture_log_snapshot("agent", tail_lines=5, max_bytes=950) |
| assert snap.full_text is not None |
| assert "truncated" in snap.full_text |
| raw = snap.full_text.split("\n", 1)[1] |
| kept = [l for l in raw.strip().splitlines() if l.startswith("A")] |
| |
| assert len(kept) == 9 |
|
|
| def test_unknown_log_returns_none(self, hermes_home): |
| from hermes_cli.debug import _capture_log_snapshot |
| snap = _capture_log_snapshot("nonexistent", tail_lines=10) |
| assert snap.full_text is None |
|
|
| def test_falls_back_to_rotated_file(self, hermes_home): |
| """When gateway.log doesn't exist, falls back to gateway.log.1.""" |
| from hermes_cli.debug import _capture_log_snapshot |
|
|
| logs_dir = hermes_home / "logs" |
| |
| (logs_dir / "gateway.log").unlink(missing_ok=True) |
| (logs_dir / "gateway.log.1").write_text( |
| "2026-04-12 10:00:00 INFO gateway.run: rotated content\n" |
| ) |
|
|
| snap = _capture_log_snapshot("gateway", tail_lines=10) |
| assert snap.full_text is not None |
| assert "rotated content" in snap.full_text |
|
|
| def test_prefers_primary_over_rotated(self, hermes_home): |
| """Primary log is used when it exists, even if .1 also exists.""" |
| from hermes_cli.debug import _capture_log_snapshot |
|
|
| logs_dir = hermes_home / "logs" |
| (logs_dir / "gateway.log").write_text("primary content\n") |
| (logs_dir / "gateway.log.1").write_text("rotated content\n") |
|
|
| snap = _capture_log_snapshot("gateway", tail_lines=10) |
| assert "primary content" in snap.full_text |
| assert "rotated" not in snap.full_text |
|
|
| def test_falls_back_when_primary_empty(self, hermes_home): |
| """Empty primary log falls back to .1 rotation.""" |
| from hermes_cli.debug import _capture_log_snapshot |
|
|
| logs_dir = hermes_home / "logs" |
| (logs_dir / "agent.log").write_text("") |
| (logs_dir / "agent.log.1").write_text("rotated agent data\n") |
|
|
| snap = _capture_log_snapshot("agent", tail_lines=10) |
| assert snap.full_text is not None |
| assert "rotated agent data" in snap.full_text |
|
|
|
|
| |
| |
| |
|
|
| class TestCollectDebugReport: |
| """Test the debug report builder.""" |
|
|
| def test_report_includes_dump_output(self, hermes_home): |
| from hermes_cli.debug import collect_debug_report |
|
|
| with patch("hermes_cli.dump.run_dump") as mock_dump: |
| mock_dump.side_effect = lambda args: print( |
| "--- hermes dump ---\nversion: 0.8.0\n--- end dump ---" |
| ) |
| report = collect_debug_report(log_lines=50) |
|
|
| assert "--- hermes dump ---" in report |
| assert "version: 0.8.0" in report |
|
|
| def test_report_includes_agent_log(self, hermes_home): |
| from hermes_cli.debug import collect_debug_report |
|
|
| with patch("hermes_cli.dump.run_dump"): |
| report = collect_debug_report(log_lines=50) |
|
|
| assert "--- agent.log" in report |
| assert "session started" in report |
|
|
| def test_report_includes_errors_log(self, hermes_home): |
| from hermes_cli.debug import collect_debug_report |
|
|
| with patch("hermes_cli.dump.run_dump"): |
| report = collect_debug_report(log_lines=50) |
|
|
| assert "--- errors.log" in report |
| assert "connection lost" in report |
|
|
| def test_report_includes_gateway_log(self, hermes_home): |
| from hermes_cli.debug import collect_debug_report |
|
|
| with patch("hermes_cli.dump.run_dump"): |
| report = collect_debug_report(log_lines=50) |
|
|
| assert "--- gateway.log" in report |
|
|
| def test_missing_logs_handled(self, tmp_path, monkeypatch): |
| home = tmp_path / ".hermes" |
| home.mkdir() |
| monkeypatch.setenv("HERMES_HOME", str(home)) |
|
|
| from hermes_cli.debug import collect_debug_report |
|
|
| with patch("hermes_cli.dump.run_dump"): |
| report = collect_debug_report(log_lines=50) |
|
|
| assert "(file not found)" in report |
|
|
|
|
| |
| |
| |
|
|
| class TestRunDebugShare: |
| """Test the run_debug_share CLI handler.""" |
|
|
| def test_share_sweeps_expired_pastes(self, hermes_home, capsys): |
| """Slash-command path should sweep old pending deletes before uploading.""" |
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = False |
|
|
| with patch("hermes_cli.dump.run_dump"), \ |
| patch("hermes_cli.debug._sweep_expired_pastes", return_value=(0, 0)) as mock_sweep, \ |
| patch("hermes_cli.debug.upload_to_pastebin", |
| return_value="https://paste.rs/test"): |
| run_debug_share(args) |
|
|
| mock_sweep.assert_called_once() |
| assert "Debug report uploaded" in capsys.readouterr().out |
|
|
| def test_share_survives_sweep_failure(self, hermes_home, capsys): |
| """Expired-paste cleanup is best-effort and must not block sharing.""" |
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = False |
|
|
| with patch("hermes_cli.dump.run_dump"), \ |
| patch( |
| "hermes_cli.debug._sweep_expired_pastes", |
| side_effect=RuntimeError("offline"), |
| ), \ |
| patch("hermes_cli.debug.upload_to_pastebin", |
| return_value="https://paste.rs/test"): |
| run_debug_share(args) |
|
|
| assert "https://paste.rs/test" in capsys.readouterr().out |
|
|
| def test_local_flag_prints_full_logs(self, hermes_home, capsys): |
| """--local prints the report plus full log contents.""" |
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = True |
|
|
| with patch("hermes_cli.dump.run_dump"): |
| run_debug_share(args) |
|
|
| out = capsys.readouterr().out |
| assert "--- agent.log" in out |
| assert "FULL agent.log" in out |
| assert "FULL gateway.log" in out |
|
|
| def test_share_uploads_three_pastes(self, hermes_home, capsys): |
| """Successful share uploads report + agent.log + gateway.log.""" |
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = False |
|
|
| call_count = [0] |
| uploaded_content = [] |
| def _mock_upload(content, expiry_days=7): |
| call_count[0] += 1 |
| uploaded_content.append(content) |
| return f"https://paste.rs/paste{call_count[0]}" |
|
|
| with patch("hermes_cli.dump.run_dump") as mock_dump, \ |
| patch("hermes_cli.debug.upload_to_pastebin", |
| side_effect=_mock_upload): |
| mock_dump.side_effect = lambda a: print("--- hermes dump ---\nversion: test\n--- end dump ---") |
| run_debug_share(args) |
|
|
| out = capsys.readouterr().out |
| |
| assert call_count[0] == 3 |
| assert "paste.rs/paste1" in out |
| assert "paste.rs/paste2" in out |
| assert "paste.rs/paste3" in out |
| assert "Report" in out |
| assert "agent.log" in out |
| assert "gateway.log" in out |
|
|
| |
| agent_paste = uploaded_content[1] |
| assert "--- hermes dump ---" in agent_paste |
| assert "--- full agent.log ---" in agent_paste |
| gateway_paste = uploaded_content[2] |
| assert "--- hermes dump ---" in gateway_paste |
| assert "--- full gateway.log ---" in gateway_paste |
|
|
| def test_share_keeps_report_and_full_log_on_same_snapshot(self, hermes_home, capsys): |
| """A mid-run rotation must not make full agent.log older than the report.""" |
| from hermes_cli.debug import run_debug_share, collect_debug_report as real_collect_debug_report |
|
|
| logs_dir = hermes_home / "logs" |
| (logs_dir / "agent.log").write_text( |
| "2026-04-22 12:00:00 INFO agent: newest line\n" |
| ) |
| (logs_dir / "agent.log.1").write_text( |
| "2026-04-10 12:00:00 INFO agent: old rotated line\n" |
| ) |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = False |
|
|
| uploaded_content = [] |
|
|
| def _mock_upload(content, expiry_days=7): |
| uploaded_content.append(content) |
| return f"https://paste.rs/paste{len(uploaded_content)}" |
|
|
| def _wrapped_collect_debug_report(*, log_lines=200, dump_text="", log_snapshots=None): |
| report = real_collect_debug_report( |
| log_lines=log_lines, |
| dump_text=dump_text, |
| log_snapshots=log_snapshots, |
| ) |
| |
| |
| |
| (logs_dir / "agent.log").write_text("") |
| (logs_dir / "agent.log.1").write_text( |
| "2026-04-10 12:00:00 INFO agent: old rotated line\n" |
| ) |
| return report |
|
|
| with patch("hermes_cli.dump.run_dump"), \ |
| patch("hermes_cli.debug.collect_debug_report", side_effect=_wrapped_collect_debug_report), \ |
| patch("hermes_cli.debug.upload_to_pastebin", side_effect=_mock_upload): |
| run_debug_share(args) |
|
|
| report_paste = uploaded_content[0] |
| agent_paste = uploaded_content[1] |
| assert "2026-04-22 12:00:00 INFO agent: newest line" in report_paste |
| assert "2026-04-22 12:00:00 INFO agent: newest line" in agent_paste |
| assert "old rotated line" not in agent_paste |
|
|
| def test_share_skips_missing_logs(self, tmp_path, monkeypatch, capsys): |
| """Only uploads logs that exist.""" |
| home = tmp_path / ".hermes" |
| home.mkdir() |
| monkeypatch.setenv("HERMES_HOME", str(home)) |
|
|
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = False |
|
|
| call_count = [0] |
| def _mock_upload(content, expiry_days=7): |
| call_count[0] += 1 |
| return f"https://paste.rs/paste{call_count[0]}" |
|
|
| with patch("hermes_cli.dump.run_dump"), \ |
| patch("hermes_cli.debug.upload_to_pastebin", |
| side_effect=_mock_upload): |
| run_debug_share(args) |
|
|
| out = capsys.readouterr().out |
| |
| assert call_count[0] == 1 |
| assert "Report" in out |
|
|
| def test_share_continues_on_log_upload_failure(self, hermes_home, capsys): |
| """Log upload failure doesn't stop the report from being shared.""" |
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = False |
|
|
| call_count = [0] |
| def _mock_upload(content, expiry_days=7): |
| call_count[0] += 1 |
| if call_count[0] > 1: |
| raise RuntimeError("upload failed") |
| return "https://paste.rs/report" |
|
|
| with patch("hermes_cli.dump.run_dump"), \ |
| patch("hermes_cli.debug.upload_to_pastebin", |
| side_effect=_mock_upload): |
| run_debug_share(args) |
|
|
| out = capsys.readouterr().out |
| assert "Report" in out |
| assert "paste.rs/report" in out |
| assert "failed to upload" in out |
|
|
| def test_share_exits_on_report_upload_failure(self, hermes_home, capsys): |
| """If the main report fails to upload, exit with code 1.""" |
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = False |
|
|
| with patch("hermes_cli.dump.run_dump"), \ |
| patch("hermes_cli.debug.upload_to_pastebin", |
| side_effect=RuntimeError("all failed")): |
| with pytest.raises(SystemExit) as exc_info: |
| run_debug_share(args) |
|
|
| assert exc_info.value.code == 1 |
| out = capsys.readouterr() |
| assert "all failed" in out.err |
|
|
|
|
| |
| |
| |
|
|
| class TestRunDebug: |
| def test_no_subcommand_shows_usage(self, capsys): |
| from hermes_cli.debug import run_debug |
|
|
| args = MagicMock() |
| args.debug_command = None |
|
|
| run_debug(args) |
|
|
| out = capsys.readouterr().out |
| assert "hermes debug" in out |
| assert "share" in out |
| assert "delete" in out |
|
|
| def test_share_subcommand_routes(self, hermes_home): |
| from hermes_cli.debug import run_debug |
|
|
| args = MagicMock() |
| args.debug_command = "share" |
| args.lines = 200 |
| args.expire = 7 |
| args.local = True |
|
|
| with patch("hermes_cli.dump.run_dump"): |
| run_debug(args) |
|
|
|
|
| |
| |
| |
|
|
| |
| |
| |
|
|
| class TestExtractPasteId: |
| def test_paste_rs_url(self): |
| from hermes_cli.debug import _extract_paste_id |
| assert _extract_paste_id("https://paste.rs/abc123") == "abc123" |
|
|
| def test_paste_rs_trailing_slash(self): |
| from hermes_cli.debug import _extract_paste_id |
| assert _extract_paste_id("https://paste.rs/abc123/") == "abc123" |
|
|
| def test_http_variant(self): |
| from hermes_cli.debug import _extract_paste_id |
| assert _extract_paste_id("http://paste.rs/xyz") == "xyz" |
|
|
| def test_non_paste_rs_returns_none(self): |
| from hermes_cli.debug import _extract_paste_id |
| assert _extract_paste_id("https://dpaste.com/ABCDEF") is None |
|
|
| def test_empty_returns_none(self): |
| from hermes_cli.debug import _extract_paste_id |
| assert _extract_paste_id("") is None |
|
|
|
|
| class TestDeletePaste: |
| def test_delete_sends_delete_request(self): |
| from hermes_cli.debug import delete_paste |
|
|
| mock_resp = MagicMock() |
| mock_resp.status = 200 |
| mock_resp.__enter__ = lambda s: s |
| mock_resp.__exit__ = MagicMock(return_value=False) |
|
|
| with patch("hermes_cli.debug.urllib.request.urlopen", |
| return_value=mock_resp) as mock_open: |
| result = delete_paste("https://paste.rs/abc123") |
|
|
| assert result is True |
| req = mock_open.call_args[0][0] |
| assert req.method == "DELETE" |
| assert "paste.rs/abc123" in req.full_url |
|
|
| def test_delete_rejects_non_paste_rs(self): |
| from hermes_cli.debug import delete_paste |
|
|
| with pytest.raises(ValueError, match="only paste.rs"): |
| delete_paste("https://dpaste.com/something") |
|
|
|
|
| class TestScheduleAutoDelete: |
| """``_schedule_auto_delete`` used to spawn a detached Python subprocess |
| per call (one per paste URL batch). Those subprocesses slept 6 hours |
| and accumulated forever under repeated use — 15+ orphaned interpreters |
| were observed in production. |
| |
| The new implementation is stateless: it records pending deletions to |
| ``~/.hermes/pastes/pending.json`` and lets ``_sweep_expired_pastes`` |
| handle the DELETE requests synchronously on the next ``hermes debug`` |
| invocation. |
| """ |
|
|
| def test_does_not_spawn_subprocess(self, hermes_home): |
| """Regression guard: _schedule_auto_delete must NEVER spawn subprocesses. |
| |
| We assert this structurally rather than by mocking Popen: the new |
| implementation doesn't even import ``subprocess`` at module scope, |
| so a mock patch wouldn't find it. |
| """ |
| import ast |
| import inspect |
| from hermes_cli.debug import _schedule_auto_delete |
|
|
| |
| |
| source = inspect.getsource(_schedule_auto_delete) |
| tree = ast.parse(source) |
| func_node = tree.body[0] |
| if ( |
| func_node.body |
| and isinstance(func_node.body[0], ast.Expr) |
| and isinstance(func_node.body[0].value, ast.Constant) |
| and isinstance(func_node.body[0].value.value, str) |
| ): |
| func_node.body = func_node.body[1:] |
| code_only = ast.unparse(func_node) |
|
|
| assert "Popen" not in code_only, ( |
| "_schedule_auto_delete must not spawn subprocesses — " |
| "use pending.json + _sweep_expired_pastes instead" |
| ) |
| assert "subprocess" not in code_only, ( |
| "_schedule_auto_delete must not reference subprocess at all" |
| ) |
| assert "time.sleep" not in code_only, ( |
| "Regression: sleeping in _schedule_auto_delete is the bug being fixed" |
| ) |
|
|
| |
| |
| import os as _os |
| before = set(_os.listdir("/proc")) if _os.path.exists("/proc") else None |
| _schedule_auto_delete( |
| ["https://paste.rs/abc", "https://paste.rs/def"], |
| delay_seconds=10, |
| ) |
| if before is not None: |
| after = set(_os.listdir("/proc")) |
| new = after - before |
| |
| new_pids = [p for p in new if p.isdigit()] |
| |
| |
| |
| for pid in new_pids: |
| try: |
| with open(f"/proc/{pid}/cmdline", "rb") as f: |
| cmdline = f.read().decode("utf-8", errors="replace") |
| assert "time.sleep" not in cmdline, ( |
| f"Leaked sleeper subprocess PID {pid}: {cmdline}" |
| ) |
| except OSError: |
| pass |
|
|
| def test_records_pending_to_json(self, hermes_home): |
| """Scheduled URLs are persisted to pending.json with expiration.""" |
| from hermes_cli.debug import _schedule_auto_delete, _pending_file |
| import json |
|
|
| _schedule_auto_delete( |
| ["https://paste.rs/abc", "https://paste.rs/def"], |
| delay_seconds=10, |
| ) |
|
|
| pending_path = _pending_file() |
| assert pending_path.exists() |
|
|
| entries = json.loads(pending_path.read_text()) |
| assert len(entries) == 2 |
| urls = {e["url"] for e in entries} |
| assert urls == {"https://paste.rs/abc", "https://paste.rs/def"} |
|
|
| |
| import time |
| for e in entries: |
| assert e["expire_at"] > time.time() |
| assert e["expire_at"] <= time.time() + 15 |
|
|
| def test_skips_non_paste_rs_urls(self, hermes_home): |
| """dpaste.com URLs auto-expire — don't track them.""" |
| from hermes_cli.debug import _schedule_auto_delete, _pending_file |
|
|
| _schedule_auto_delete(["https://dpaste.com/something"]) |
|
|
| |
| assert not _pending_file().exists() |
|
|
| def test_merges_with_existing_pending(self, hermes_home): |
| """Subsequent calls merge into existing pending.json.""" |
| from hermes_cli.debug import _schedule_auto_delete, _load_pending |
|
|
| _schedule_auto_delete(["https://paste.rs/first"], delay_seconds=10) |
| _schedule_auto_delete(["https://paste.rs/second"], delay_seconds=10) |
|
|
| entries = _load_pending() |
| urls = {e["url"] for e in entries} |
| assert urls == {"https://paste.rs/first", "https://paste.rs/second"} |
|
|
| def test_dedupes_same_url(self, hermes_home): |
| """Same URL recorded twice → one entry with the later expire_at.""" |
| from hermes_cli.debug import _schedule_auto_delete, _load_pending |
|
|
| _schedule_auto_delete(["https://paste.rs/dup"], delay_seconds=10) |
| _schedule_auto_delete(["https://paste.rs/dup"], delay_seconds=100) |
|
|
| entries = _load_pending() |
| assert len(entries) == 1 |
| assert entries[0]["url"] == "https://paste.rs/dup" |
|
|
|
|
| class TestSweepExpiredPastes: |
| """Test the opportunistic sweep that replaces the sleeping subprocess.""" |
|
|
| def test_sweep_empty_is_noop(self, hermes_home): |
| from hermes_cli.debug import _sweep_expired_pastes |
|
|
| deleted, remaining = _sweep_expired_pastes() |
| assert deleted == 0 |
| assert remaining == 0 |
|
|
| def test_sweep_deletes_expired_entries(self, hermes_home): |
| from hermes_cli.debug import ( |
| _sweep_expired_pastes, |
| _save_pending, |
| _load_pending, |
| ) |
| import time |
|
|
| |
| _save_pending([ |
| {"url": "https://paste.rs/expired", "expire_at": time.time() - 100}, |
| {"url": "https://paste.rs/future", "expire_at": time.time() + 3600}, |
| ]) |
|
|
| delete_calls = [] |
|
|
| def fake_delete(url): |
| delete_calls.append(url) |
| return True |
|
|
| with patch("hermes_cli.debug.delete_paste", side_effect=fake_delete): |
| deleted, remaining = _sweep_expired_pastes() |
|
|
| assert delete_calls == ["https://paste.rs/expired"] |
| assert deleted == 1 |
| assert remaining == 1 |
|
|
| entries = _load_pending() |
| urls = {e["url"] for e in entries} |
| assert urls == {"https://paste.rs/future"} |
|
|
| def test_sweep_leaves_future_entries_alone(self, hermes_home): |
| from hermes_cli.debug import _sweep_expired_pastes, _save_pending |
| import time |
|
|
| _save_pending([ |
| {"url": "https://paste.rs/future1", "expire_at": time.time() + 3600}, |
| {"url": "https://paste.rs/future2", "expire_at": time.time() + 7200}, |
| ]) |
|
|
| with patch("hermes_cli.debug.delete_paste") as mock_delete: |
| deleted, remaining = _sweep_expired_pastes() |
|
|
| mock_delete.assert_not_called() |
| assert deleted == 0 |
| assert remaining == 2 |
|
|
| def test_sweep_survives_network_failure(self, hermes_home): |
| """Failed DELETEs stay in pending.json until the 24h grace window.""" |
| from hermes_cli.debug import ( |
| _sweep_expired_pastes, |
| _save_pending, |
| _load_pending, |
| ) |
| import time |
|
|
| _save_pending([ |
| {"url": "https://paste.rs/flaky", "expire_at": time.time() - 100}, |
| ]) |
|
|
| with patch( |
| "hermes_cli.debug.delete_paste", |
| side_effect=Exception("network down"), |
| ): |
| deleted, remaining = _sweep_expired_pastes() |
|
|
| |
| assert deleted == 0 |
| assert remaining == 1 |
| assert len(_load_pending()) == 1 |
|
|
| def test_sweep_drops_entries_past_grace_window(self, hermes_home): |
| """After 24h past expiration, give up even on network failures.""" |
| from hermes_cli.debug import ( |
| _sweep_expired_pastes, |
| _save_pending, |
| _load_pending, |
| ) |
| import time |
|
|
| |
| very_old = time.time() - (25 * 3600) |
| _save_pending([ |
| {"url": "https://paste.rs/ancient", "expire_at": very_old}, |
| ]) |
|
|
| with patch( |
| "hermes_cli.debug.delete_paste", |
| side_effect=Exception("network down"), |
| ): |
| deleted, remaining = _sweep_expired_pastes() |
|
|
| assert deleted == 1 |
| assert remaining == 0 |
| assert _load_pending() == [] |
|
|
|
|
| class TestRunDebugSweepsOnInvocation: |
| """``run_debug`` must sweep expired pastes on every invocation.""" |
|
|
| def test_run_debug_calls_sweep(self, hermes_home): |
| from hermes_cli.debug import run_debug |
|
|
| args = MagicMock() |
| args.debug_command = None |
|
|
| with patch("hermes_cli.debug._sweep_expired_pastes") as mock_sweep: |
| run_debug(args) |
|
|
| mock_sweep.assert_called_once() |
|
|
| def test_run_debug_survives_sweep_failure(self, hermes_home, capsys): |
| """If the sweep throws, the subcommand still runs.""" |
| from hermes_cli.debug import run_debug |
|
|
| args = MagicMock() |
| args.debug_command = None |
|
|
| with patch( |
| "hermes_cli.debug._sweep_expired_pastes", |
| side_effect=RuntimeError("boom"), |
| ): |
| run_debug(args) |
|
|
| |
| out = capsys.readouterr().out |
| assert "Usage: hermes debug" in out |
|
|
|
|
| class TestRunDebugDelete: |
| def test_deletes_valid_url(self, capsys): |
| from hermes_cli.debug import run_debug_delete |
|
|
| args = MagicMock() |
| args.urls = ["https://paste.rs/abc"] |
|
|
| with patch("hermes_cli.debug.delete_paste", return_value=True): |
| run_debug_delete(args) |
|
|
| out = capsys.readouterr().out |
| assert "Deleted" in out |
| assert "paste.rs/abc" in out |
|
|
| def test_handles_delete_failure(self, capsys): |
| from hermes_cli.debug import run_debug_delete |
|
|
| args = MagicMock() |
| args.urls = ["https://paste.rs/abc"] |
|
|
| with patch("hermes_cli.debug.delete_paste", |
| side_effect=Exception("network error")): |
| run_debug_delete(args) |
|
|
| out = capsys.readouterr().out |
| assert "Could not delete" in out |
|
|
| def test_no_urls_shows_usage(self, capsys): |
| from hermes_cli.debug import run_debug_delete |
|
|
| args = MagicMock() |
| args.urls = [] |
|
|
| run_debug_delete(args) |
|
|
| out = capsys.readouterr().out |
| assert "Usage" in out |
|
|
|
|
| class TestShareIncludesAutoDelete: |
| """Verify that run_debug_share schedules auto-deletion and prints TTL.""" |
|
|
| def test_share_schedules_auto_delete(self, hermes_home, capsys): |
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = False |
|
|
| with patch("hermes_cli.dump.run_dump"), \ |
| patch("hermes_cli.debug.upload_to_pastebin", |
| return_value="https://paste.rs/test1"), \ |
| patch("hermes_cli.debug._schedule_auto_delete") as mock_sched: |
| run_debug_share(args) |
|
|
| |
| mock_sched.assert_called_once() |
| urls_arg = mock_sched.call_args[0][0] |
| assert "https://paste.rs/test1" in urls_arg |
|
|
| out = capsys.readouterr().out |
| assert "auto-delete" in out |
|
|
| def test_share_shows_privacy_notice(self, hermes_home, capsys): |
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = False |
|
|
| with patch("hermes_cli.dump.run_dump"), \ |
| patch("hermes_cli.debug.upload_to_pastebin", |
| return_value="https://paste.rs/test"), \ |
| patch("hermes_cli.debug._schedule_auto_delete"): |
| run_debug_share(args) |
|
|
| out = capsys.readouterr().out |
| assert "public paste service" in out |
|
|
| def test_local_no_privacy_notice(self, hermes_home, capsys): |
| from hermes_cli.debug import run_debug_share |
|
|
| args = MagicMock() |
| args.lines = 50 |
| args.expire = 7 |
| args.local = True |
|
|
| with patch("hermes_cli.dump.run_dump"): |
| run_debug_share(args) |
|
|
| out = capsys.readouterr().out |
| assert "public paste service" not in out |
|
|