mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-05-25 05:52:34 +00:00
fix(test_gateway): stop run_gateway() tests from rewriting the dev's installed systemd unit (#22900)
run_gateway() calls refresh_systemd_unit_if_needed() on every invocation so restart settings stay current after exit-code-75 respawns. The user-scope unit path resolves under Path.home() (NOT sandboxed by conftest, only HERMES_HOME is), and generate_systemd_unit() bakes the current HERMES_HOME into the unit's Environment= line. Result: any test that exercises run_gateway() end-to-end on a real Linux dev box silently rewrites the developer's installed ~/.config/systemd/user/hermes-gateway.service with a polluted HERMES_HOME pointing at /tmp/pytest-of-<user>/.../hermes_test. On the next reboot, systemd loads that unit, the gateway starts looking at an empty tmp dir, and Telegram/Discord/etc. all show as 'No messaging platforms enabled' even though the user's real config is fine. Three tests in tests/hermes_cli/test_gateway.py hit this path: test_run_gateway_exits_cleanly_on_keyboard_interrupt, test_run_gateway_exits_nonzero_when_start_gateway_reports_failure, and test_run_gateway_root_guard_has_escape_hatch. Two-layer fix: 1. _install_fake_gateway_run helper (covers all four run_gateway() call sites in test_gateway.py and any future ones) now also stubs supports_systemd_services and refresh_systemd_unit_if_needed. 2. refresh_systemd_unit_if_needed() itself sniffs the generated unit body for /pytest-of- and /hermes_test markers and refuses to write when present. Defense in depth so a future test that bypasses the helper still can't corrupt the dev's gateway. Tests that legitimately exercise the refresh flow (test_run_gateway_refreshes_outdated_unit_on_boot) patch generate_systemd_unit to return synthetic content that doesn't carry those markers, so they keep working. Adds test_refresh_refuses_to_bake_pytest_tmpdir_into_real_user_unit as a regression test for the source-side guard.
This commit is contained in:
parent
4f8d8ad912
commit
2ffef15675
3 changed files with 93 additions and 1 deletions
|
|
@ -2230,7 +2230,30 @@ def refresh_systemd_unit_if_needed(system: bool = False) -> bool:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
expected_user = _read_systemd_user_from_unit(unit_path) if system else None
|
expected_user = _read_systemd_user_from_unit(unit_path) if system else None
|
||||||
unit_path.write_text(generate_systemd_unit(system=system, run_as_user=expected_user), encoding="utf-8")
|
new_unit = generate_systemd_unit(system=system, run_as_user=expected_user)
|
||||||
|
|
||||||
|
# ── Test-environment safety belt ─────────────────────────────────────
|
||||||
|
# The user-scope unit path resolves under ``Path.home()``, which is NOT
|
||||||
|
# sandboxed by the test conftest (only HERMES_HOME is). If a test
|
||||||
|
# exercises ``run_gateway()`` with a pytest-tmp HERMES_HOME, the freshly
|
||||||
|
# generated unit bakes that ``/tmp/pytest-of-.../hermes_test`` path into
|
||||||
|
# ``Environment="HERMES_HOME=..."``. Writing that to the developer's
|
||||||
|
# real user systemd unit file silently breaks their gateway on the next
|
||||||
|
# reboot (systemd loads the polluted env, the gateway looks at an empty
|
||||||
|
# tmp dir, and Telegram/Discord/etc. all show as "not configured").
|
||||||
|
# Refuse to write when the generated unit references a pytest tmpdir.
|
||||||
|
# Detection sniffs the unit body — tests that legitimately exercise the
|
||||||
|
# refresh flow patch ``generate_systemd_unit`` to return synthetic
|
||||||
|
# content (``"new unit\n"``) which doesn't contain these markers and
|
||||||
|
# still works.
|
||||||
|
if not system and (
|
||||||
|
"/pytest-of-" in new_unit
|
||||||
|
or "/hermes_test\"" in new_unit
|
||||||
|
or "/hermes_test/" in new_unit
|
||||||
|
):
|
||||||
|
return False
|
||||||
|
|
||||||
|
unit_path.write_text(new_unit, encoding="utf-8")
|
||||||
_run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
|
_run_systemctl(["daemon-reload"], system=system, check=True, timeout=30)
|
||||||
print(f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install")
|
print(f"↻ Updated gateway {_service_scope_label(system)} service definition to match the current Hermes install")
|
||||||
return True
|
return True
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,21 @@ def _install_fake_gateway_run(monkeypatch, start_gateway):
|
||||||
module = ModuleType("gateway.run")
|
module = ModuleType("gateway.run")
|
||||||
module.start_gateway = start_gateway
|
module.start_gateway = start_gateway
|
||||||
monkeypatch.setitem(sys.modules, "gateway.run", module)
|
monkeypatch.setitem(sys.modules, "gateway.run", module)
|
||||||
|
# ``run_gateway()`` calls ``refresh_systemd_unit_if_needed()`` on every
|
||||||
|
# invocation so that restart settings stay current after exit-code-75
|
||||||
|
# respawns. That helper writes to ``Path.home() / ".config/systemd/user
|
||||||
|
# /hermes-gateway.service"`` and runs ``systemctl --user daemon-reload``
|
||||||
|
# — both target the *real* user environment because the conftest only
|
||||||
|
# sandboxes ``HERMES_HOME``, not ``HOME``. Tests that drive
|
||||||
|
# ``run_gateway()`` end-to-end with a fake ``start_gateway`` MUST stub
|
||||||
|
# the refresh call too, or every run rewrites the developer's installed
|
||||||
|
# unit (baking in the test's pytest-tmp ``HERMES_HOME`` value, which
|
||||||
|
# systemd then uses on the next boot — silently breaking the gateway
|
||||||
|
# for the developer).
|
||||||
|
monkeypatch.setattr(gateway, "supports_systemd_services", lambda: False)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
gateway, "refresh_systemd_unit_if_needed", lambda system=False: False
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def test_run_gateway_exits_cleanly_on_keyboard_interrupt(monkeypatch, capsys):
|
def test_run_gateway_exits_cleanly_on_keyboard_interrupt(monkeypatch, capsys):
|
||||||
|
|
|
||||||
|
|
@ -234,6 +234,60 @@ class TestSystemdServiceRefresh:
|
||||||
assert unit_path.read_text(encoding="utf-8") == "new unit\n"
|
assert unit_path.read_text(encoding="utf-8") == "new unit\n"
|
||||||
assert ["systemctl", "--user", "daemon-reload"] in calls
|
assert ["systemctl", "--user", "daemon-reload"] in calls
|
||||||
|
|
||||||
|
def test_refresh_refuses_to_bake_pytest_tmpdir_into_real_user_unit(
|
||||||
|
self, tmp_path, monkeypatch
|
||||||
|
):
|
||||||
|
"""Defense in depth: ``refresh_systemd_unit_if_needed()`` runs every
|
||||||
|
time ``run_gateway()`` starts. The user-scope unit path resolves
|
||||||
|
under ``Path.home()`` (NOT sandboxed by conftest), and
|
||||||
|
``generate_systemd_unit()`` bakes ``HERMES_HOME`` into the unit's
|
||||||
|
``Environment=`` line. Without this guard, any test that drives
|
||||||
|
``run_gateway()`` end-to-end on a real Linux dev box silently
|
||||||
|
rewrites the developer's installed gateway unit with a
|
||||||
|
``/tmp/pytest-of-.../hermes_test`` HERMES_HOME — silently breaking
|
||||||
|
their gateway on the next boot. The guard sniffs the generated
|
||||||
|
unit body for tmpdir markers and refuses the write. Tests that
|
||||||
|
legitimately exercise the refresh flow patch
|
||||||
|
``generate_systemd_unit`` to return synthetic content that doesn't
|
||||||
|
carry those markers.
|
||||||
|
"""
|
||||||
|
unit_path = tmp_path / "hermes-gateway.service"
|
||||||
|
unit_path.write_text("old unit\n", encoding="utf-8")
|
||||||
|
|
||||||
|
monkeypatch.setattr(
|
||||||
|
gateway_cli, "get_systemd_unit_path", lambda system=False: unit_path
|
||||||
|
)
|
||||||
|
# Realistic generated unit referencing a pytest tmpdir HERMES_HOME
|
||||||
|
polluted_unit = (
|
||||||
|
"[Service]\n"
|
||||||
|
'Environment="HERMES_HOME=/tmp/pytest-of-alice/pytest-42/'
|
||||||
|
'popen-gw0/test_x/hermes_test"\n'
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
gateway_cli,
|
||||||
|
"generate_systemd_unit",
|
||||||
|
lambda system=False, run_as_user=None: polluted_unit,
|
||||||
|
)
|
||||||
|
|
||||||
|
# If the guard fails, daemon-reload would be called — record it.
|
||||||
|
ran = []
|
||||||
|
|
||||||
|
def fake_run(cmd, check=True, **kwargs):
|
||||||
|
ran.append(cmd)
|
||||||
|
return SimpleNamespace(returncode=0, stdout="", stderr="")
|
||||||
|
|
||||||
|
monkeypatch.setattr(gateway_cli.subprocess, "run", fake_run)
|
||||||
|
|
||||||
|
result = gateway_cli.refresh_systemd_unit_if_needed(system=False)
|
||||||
|
|
||||||
|
assert result is False, "refresh should refuse to write a polluted unit"
|
||||||
|
assert (
|
||||||
|
unit_path.read_text(encoding="utf-8") == "old unit\n"
|
||||||
|
), "installed unit must be left untouched"
|
||||||
|
assert not any(
|
||||||
|
"daemon-reload" in str(c) for c in ran
|
||||||
|
), "daemon-reload must not run when write was refused"
|
||||||
|
|
||||||
|
|
||||||
class TestRequireServiceInstalled:
|
class TestRequireServiceInstalled:
|
||||||
def test_exits_with_install_hint_when_unit_missing(self, tmp_path, monkeypatch, capsys):
|
def test_exits_with_install_hint_when_unit_missing(self, tmp_path, monkeypatch, capsys):
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue