mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-04-25 00:51:20 +00:00
304fb921bf
Both fixes close process leaks observed in production (18+ orphaned
agent-browser node daemons, 15+ orphaned paste.rs sleep interpreters
accumulated over ~3 days, ~2.7 GB RSS).
## agent-browser daemon leak
Previously the orphan reaper (_reap_orphaned_browser_sessions) only ran
from _start_browser_cleanup_thread, which is only invoked on the first
browser tool call in a process. Hermes sessions that never used the
browser never swept orphans, and the cross-process orphan detection
relied on in-process _active_sessions, which doesn't see other hermes
PIDs' sessions (race risk).
- Write <session>.owner_pid alongside the socket dir recording the
hermes PID that owns the daemon (extracted into _write_owner_pid for
direct testability).
- Reaper prefers owner_pid liveness over in-process _active_sessions.
Cross-process safe: concurrent hermes instances won't reap each
other's daemons. Legacy tracked_names fallback kept for daemons
that predate owner_pid.
- atexit handler (_emergency_cleanup_all_sessions) now always runs
the reaper, not just when this process had active sessions —
every clean hermes exit sweeps accumulated orphans.
## paste.rs auto-delete leak
_schedule_auto_delete spawned a detached Python subprocess per call
that slept 6 hours then issued DELETE requests. No dedup, no tracking —
every 'hermes debug share' invocation added ~20 MB of resident Python
interpreters that stuck around until the sleep finished.
- Replaced the spawn with ~/.hermes/pastes/pending.json: records
{url, expire_at} entries.
- _sweep_expired_pastes() synchronously DELETEs past-due entries on
every 'hermes debug' invocation (run_debug() dispatcher).
- Network failures stay in pending.json for up to 24h, then give up
(paste.rs's own retention handles the 'user never runs hermes again'
edge case).
- Zero subprocesses; regression test asserts subprocess/Popen/time.sleep
never appear in the function source (skipping docstrings via AST).
## Validation
| | Before | After |
|------------------------------|---------------|--------------|
| Orphan agent-browser daemons | 18 accumulated| 2 (live) |
| paste.rs sleep interpreters | 15 accumulated| 0 |
| RSS reclaimed | - | ~2.7 GB |
| Targeted tests | - | 2253 pass |
E2E verified: alive-owner daemons NOT reaped; dead-owner daemons
SIGTERM'd and socket dirs cleaned; pending.json sweep deletes expired
entries without spawning subprocesses.
865 lines
30 KiB
Python
865 lines
30 KiB
Python
"""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
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@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))
|
|
|
|
# Create log files
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Unit tests for upload helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
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")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Log reading
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestReadFullLog:
|
|
"""Test _read_full_log for standalone log uploads."""
|
|
|
|
def test_reads_small_file(self, hermes_home):
|
|
from hermes_cli.debug import _read_full_log
|
|
|
|
content = _read_full_log("agent")
|
|
assert content is not None
|
|
assert "session started" in content
|
|
|
|
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 _read_full_log
|
|
assert _read_full_log("agent") is None
|
|
|
|
def test_returns_none_for_empty(self, hermes_home):
|
|
# Truncate agent.log to empty
|
|
(hermes_home / "logs" / "agent.log").write_text("")
|
|
|
|
from hermes_cli.debug import _read_full_log
|
|
assert _read_full_log("agent") is None
|
|
|
|
def test_truncates_large_file(self, hermes_home):
|
|
"""Files larger than max_bytes get tail-truncated."""
|
|
from hermes_cli.debug import _read_full_log
|
|
|
|
# Write a file larger than 1KB
|
|
big_content = "x" * 100 + "\n"
|
|
(hermes_home / "logs" / "agent.log").write_text(big_content * 200)
|
|
|
|
content = _read_full_log("agent", max_bytes=1024)
|
|
assert content is not None
|
|
assert "truncated" in content
|
|
|
|
def test_unknown_log_returns_none(self, hermes_home):
|
|
from hermes_cli.debug import _read_full_log
|
|
assert _read_full_log("nonexistent") 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 _read_full_log
|
|
|
|
logs_dir = hermes_home / "logs"
|
|
# Remove the primary (if any) and create a .1 rotation
|
|
(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"
|
|
)
|
|
|
|
content = _read_full_log("gateway")
|
|
assert content is not None
|
|
assert "rotated content" in content
|
|
|
|
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 _read_full_log
|
|
|
|
logs_dir = hermes_home / "logs"
|
|
(logs_dir / "gateway.log").write_text("primary content\n")
|
|
(logs_dir / "gateway.log.1").write_text("rotated content\n")
|
|
|
|
content = _read_full_log("gateway")
|
|
assert "primary content" in content
|
|
assert "rotated" not in content
|
|
|
|
def test_falls_back_when_primary_empty(self, hermes_home):
|
|
"""Empty primary log falls back to .1 rotation."""
|
|
from hermes_cli.debug import _read_full_log
|
|
|
|
logs_dir = hermes_home / "logs"
|
|
(logs_dir / "agent.log").write_text("")
|
|
(logs_dir / "agent.log.1").write_text("rotated agent data\n")
|
|
|
|
content = _read_full_log("agent")
|
|
assert content is not None
|
|
assert "rotated agent data" in content
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Debug report collection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI entry point — run_debug_share
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRunDebugShare:
|
|
"""Test the run_debug_share CLI handler."""
|
|
|
|
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
|
|
# Should have 3 uploads: report, agent.log, gateway.log
|
|
assert call_count[0] == 3
|
|
assert "paste.rs/paste1" in out # Report
|
|
assert "paste.rs/paste2" in out # agent.log
|
|
assert "paste.rs/paste3" in out # gateway.log
|
|
assert "Report" in out
|
|
assert "agent.log" in out
|
|
assert "gateway.log" in out
|
|
|
|
# Each log paste should start with the dump header
|
|
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_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
|
|
# Only the report should be uploaded (no log files exist)
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_debug router
|
|
# ---------------------------------------------------------------------------
|
|
|
|
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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Argparse integration
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Delete / auto-delete
|
|
# ---------------------------------------------------------------------------
|
|
|
|
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
|
|
|
|
# Strip the docstring before scanning so the regression-rationale
|
|
# prose inside it doesn't trigger our banned-word checks.
|
|
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"
|
|
)
|
|
|
|
# And verify that calling it doesn't produce any orphaned children
|
|
# (it should just write pending.json synchronously).
|
|
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
|
|
# Filter to only integer-named entries (process PIDs)
|
|
new_pids = [p for p in new if p.isdigit()]
|
|
# It's fine if unrelated processes appeared — we just need to make
|
|
# sure we didn't spawn a long-sleeping one. The old bug spawned
|
|
# a python interpreter whose cmdline contained "time.sleep".
|
|
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 # process exited already
|
|
|
|
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"}
|
|
|
|
# expire_at is ~now + delay_seconds
|
|
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"])
|
|
|
|
# pending.json should not be created for non-paste.rs URLs
|
|
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
|
|
|
|
# Seed pending.json with one expired + one future entry
|
|
_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()
|
|
|
|
# Failure within 24h grace → kept for retry
|
|
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
|
|
|
|
# Expired 25 hours ago → past the 24h grace window
|
|
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 # default → prints help
|
|
|
|
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) # must not raise
|
|
|
|
# Default subcommand still printed help
|
|
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)
|
|
|
|
# auto-delete was scheduled with the uploaded URLs
|
|
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
|