ethernet
30da5d0a89
many tests patched sys.platform or a module's _IS_WINDOWS flag, then ran on linux ci. the patch selects the branch under test, but the host does not have the behavior the branch exists for. the test proves the patch, not the platform. some gated assertions never ran on any host. this commit adds three markers: linux_only, macos_only, windows_only. a conftest hook skips a marked test on the other hosts, with a clear reason. no test fakes a host now. two documented fakes remain (android/termux, freebsd) because no ci runner exists for them. each fake site got one of four treatments: - gate it: the real host supplies the platform; mocks cover real dependencies only, never host identity - patch the module's own probe when the subject is the probe's consumer - assert against the real host when the fake stood in for any non-x host - delete the patch when it set the value the host already has bare skipif(sys.platform != ...) guards became markers too. the lane model skips these on linux and never imports them on windows, so they ran on no host. platform parametrize tables are now one marked test per os. running on real hosts found real errors: a chrome-sandbox failure in test_gui_command that main hides, and two windows failures fixed here. the agents.md testing section now documents the policy.
219 lines
8.2 KiB
Python
219 lines
8.2 KiB
Python
"""Tests for the long-lived gateway heap-trim helper."""
|
|
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
import hermes_cli.mem_trim as mem_trim
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_trim_state(monkeypatch):
|
|
monkeypatch.setattr(mem_trim, "_last_trim_monotonic", 0.0)
|
|
monkeypatch.setattr(mem_trim, "_probe_done", True)
|
|
monkeypatch.setattr(mem_trim, "_malloc_trim", None)
|
|
monkeypatch.setattr(mem_trim, "_trim_call_count", 0)
|
|
|
|
|
|
def test_unsupported_allocator_is_noop_without_gc(monkeypatch):
|
|
collect = Mock()
|
|
monkeypatch.setattr(mem_trim.gc, "collect", collect)
|
|
|
|
assert mem_trim.trim_memory(force=True, reason="test") is False
|
|
collect.assert_not_called()
|
|
|
|
|
|
def test_config_kill_switch_overrides_force_from_config_file(monkeypatch, tmp_path):
|
|
from hermes_constants import reset_hermes_home_override, set_hermes_home_override
|
|
|
|
hermes_home = tmp_path / "hermes"
|
|
hermes_home.mkdir()
|
|
(hermes_home / "config.yaml").write_text(
|
|
"context:\n memory_trim:\n enabled: false\n",
|
|
encoding="utf-8",
|
|
)
|
|
trim = Mock(return_value=1)
|
|
monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
|
|
token = set_hermes_home_override(hermes_home)
|
|
|
|
try:
|
|
assert mem_trim.trim_memory(force=True) is False
|
|
trim.assert_not_called()
|
|
finally:
|
|
reset_hermes_home_override(token)
|
|
|
|
|
|
def test_default_config_declares_memory_trim_controls():
|
|
from hermes_cli.config import DEFAULT_CONFIG
|
|
|
|
context = DEFAULT_CONFIG["context"]
|
|
assert isinstance(context, dict)
|
|
settings = context["memory_trim"]
|
|
assert isinstance(settings, dict)
|
|
assert isinstance(settings["enabled"], bool)
|
|
assert isinstance(settings["cooldown_seconds"], float)
|
|
|
|
|
|
def test_collect_memory_snapshot_parses_linux_proc_status(monkeypatch):
|
|
# No ``sys.platform`` pin: the only platform check lives inside
|
|
# ``_read_proc_status``, which is replaced below — the subject here is
|
|
# the /proc/self/status parser, which is host-independent.
|
|
monkeypatch.setattr(
|
|
mem_trim,
|
|
"_read_proc_status",
|
|
lambda: "Name:\tpython\nVmRSS:\t1234 kB\nRssAnon:\t567 kB\n",
|
|
)
|
|
monkeypatch.setattr(mem_trim.threading, "active_count", lambda: 9)
|
|
|
|
assert mem_trim.collect_memory_snapshot(history_bytes=42) == {
|
|
"rss_kib": 1234,
|
|
"rss_anon_kib": 567,
|
|
"thread_count": 9,
|
|
"history_bytes": 42,
|
|
}
|
|
|
|
|
|
def test_success_collects_then_trims(monkeypatch):
|
|
calls = []
|
|
monkeypatch.setattr(mem_trim.gc, "collect", lambda: calls.append("gc"))
|
|
monkeypatch.setattr(
|
|
mem_trim, "_malloc_trim", lambda pad: calls.append(("trim", pad)) or 1
|
|
)
|
|
monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
|
|
|
|
assert mem_trim.trim_memory(reason="turn", cooldown_seconds=60) is True
|
|
assert calls == ["gc", ("trim", 0)]
|
|
assert mem_trim._last_trim_monotonic == 100.0
|
|
|
|
|
|
def test_success_logs_memory_snapshot_and_trim_result(monkeypatch, caplog):
|
|
monkeypatch.setattr(mem_trim.gc, "collect", lambda: None)
|
|
monkeypatch.setattr(mem_trim, "_malloc_trim", lambda _pad: 1)
|
|
monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
|
|
snapshots = iter(
|
|
(
|
|
{"rss_kib": 4096, "rss_anon_kib": 3072, "thread_count": 3},
|
|
{"rss_kib": 2048, "rss_anon_kib": 1024, "thread_count": 3},
|
|
)
|
|
)
|
|
monkeypatch.setattr(mem_trim, "collect_memory_snapshot", lambda: next(snapshots))
|
|
|
|
with caplog.at_level("INFO", logger="hermes_cli.mem_trim"):
|
|
assert mem_trim.trim_memory(reason="test turn") is True
|
|
|
|
assert "reason=test turn" in caplog.text
|
|
assert "malloc_trim=1" in caplog.text
|
|
assert "rss_kib=4096->2048" in caplog.text
|
|
|
|
|
|
def test_force_logs_even_when_periodic_log_sampling_skips(monkeypatch, caplog):
|
|
monkeypatch.setattr(mem_trim.gc, "collect", lambda: None)
|
|
monkeypatch.setattr(mem_trim, "_malloc_trim", lambda _pad: 1)
|
|
monkeypatch.setattr(mem_trim, "_config_settings", lambda: (True, 0.0, 99, 1.0))
|
|
# Two ticks: the forced call comes after the 5s force floor so it runs
|
|
# (the floor exists to coalesce burst closes, not to mute logging).
|
|
_ticks = iter([100.0, 110.0])
|
|
monkeypatch.setattr(mem_trim.time, "monotonic", lambda: next(_ticks, 110.0))
|
|
monkeypatch.setattr(
|
|
mem_trim,
|
|
"collect_memory_snapshot",
|
|
lambda: {"rss_kib": 4096, "rss_anon_kib": 3072, "thread_count": 3},
|
|
)
|
|
|
|
with caplog.at_level("INFO", logger="hermes_cli.mem_trim"):
|
|
assert mem_trim.trim_memory(reason="periodic") is True
|
|
assert mem_trim.trim_memory(force=True, reason="close") is True
|
|
|
|
messages = [record.getMessage() for record in caplog.records]
|
|
assert not any("reason=periodic" in message for message in messages)
|
|
assert any("reason=close" in message for message in messages)
|
|
|
|
|
|
def test_cooldown_suppresses_repeated_collection(monkeypatch):
|
|
collect = Mock()
|
|
trim = Mock(return_value=1)
|
|
monkeypatch.setattr(mem_trim.gc, "collect", collect)
|
|
monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
|
|
monkeypatch.setattr(mem_trim, "_last_trim_monotonic", 95.0)
|
|
monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
|
|
|
|
assert mem_trim.trim_memory(cooldown_seconds=60) is False
|
|
collect.assert_not_called()
|
|
trim.assert_not_called()
|
|
assert mem_trim.trim_memory(force=True, cooldown_seconds=60) is True
|
|
|
|
|
|
def test_config_cooldown_controls_rate_limit(monkeypatch):
|
|
trim = Mock(return_value=1)
|
|
monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
|
|
monkeypatch.setattr(mem_trim, "_last_trim_monotonic", 1.0)
|
|
monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config_readonly",
|
|
lambda: {
|
|
"context": {
|
|
"memory_trim": {"enabled": True, "cooldown_seconds": 120.0}
|
|
}
|
|
},
|
|
)
|
|
|
|
assert mem_trim.trim_memory() is False
|
|
trim.assert_not_called()
|
|
|
|
|
|
def test_legacy_environment_switch_does_not_control_behavior(monkeypatch):
|
|
trim = Mock(return_value=1)
|
|
monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
|
|
monkeypatch.setenv("HERMES_DISABLE_MEMORY_TRIM", "1")
|
|
monkeypatch.setattr(
|
|
"hermes_cli.config.load_config_readonly",
|
|
lambda: {"context": {"memory_trim": {"enabled": True}}},
|
|
)
|
|
|
|
assert mem_trim.trim_memory(force=True) is True
|
|
trim.assert_called_once_with(0)
|
|
|
|
|
|
def test_libc_failure_is_fail_open_and_rate_limited(monkeypatch):
|
|
trim = Mock(side_effect=RuntimeError("boom"))
|
|
monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
|
|
monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
|
|
|
|
assert mem_trim.trim_memory(reason="test", cooldown_seconds=60) is False
|
|
assert mem_trim._last_trim_monotonic == 100.0
|
|
assert mem_trim.trim_memory(cooldown_seconds=60) is False
|
|
assert trim.call_count == 1
|
|
|
|
|
|
def test_force_floor_coalesces_burst_closes(monkeypatch):
|
|
"""A delegate batch closes N child agents back-to-back, each forcing a
|
|
trim — the short force floor must coalesce the burst instead of stacking
|
|
N uncooled full gc.collect() passes in the same process."""
|
|
collect = Mock()
|
|
trim = Mock(return_value=1)
|
|
monkeypatch.setattr(mem_trim.gc, "collect", collect)
|
|
monkeypatch.setattr(mem_trim, "_malloc_trim", trim)
|
|
monkeypatch.setattr(mem_trim, "_config_settings", lambda: (True, 0.0, 1, 0.0))
|
|
monkeypatch.setattr(
|
|
mem_trim,
|
|
"collect_memory_snapshot",
|
|
lambda: {"rss_kib": 4096, "rss_anon_kib": 3072, "thread_count": 3},
|
|
)
|
|
monkeypatch.setattr(mem_trim, "_last_trim_monotonic", 0.0)
|
|
|
|
# t=100: first forced close runs.
|
|
monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 100.0)
|
|
assert mem_trim.trim_memory(force=True, reason="agent close") is True
|
|
assert trim.call_count == 1
|
|
|
|
# t=101..103: three more child closes inside the floor — all coalesced.
|
|
for t in (101.0, 102.0, 103.0):
|
|
monkeypatch.setattr(mem_trim.time, "monotonic", lambda t=t: t)
|
|
assert mem_trim.trim_memory(force=True, reason="agent close") is False
|
|
assert trim.call_count == 1, "burst closes must not stack forced trims"
|
|
|
|
# t=106: past the floor — the parent's final close-trim still fires.
|
|
monkeypatch.setattr(mem_trim.time, "monotonic", lambda: 106.0)
|
|
assert mem_trim.trim_memory(force=True, reason="agent close") is True
|
|
assert trim.call_count == 2
|