mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-11 20:58:51 +08:00
Compare commits
2 Commits
feat/plugi
...
salvage/pr
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6ea931025e | ||
|
|
6509f9c3ac |
@@ -2473,6 +2473,29 @@ def _normalize_service_definition(text: str) -> str:
|
||||
return "\n".join(line.rstrip() for line in text.strip().splitlines())
|
||||
|
||||
|
||||
# Directives that older systemd versions silently ignore/strip. Normalize
|
||||
# them out of stale-check comparisons so a unit that differs only by these
|
||||
# directives is not perpetually flagged as outdated.
|
||||
_SYSTEMD_OPTIONAL_DIRECTIVES = (
|
||||
"RestartMaxDelaySec",
|
||||
"RestartSteps",
|
||||
)
|
||||
|
||||
|
||||
def _strip_optional_systemd_directives(text: str) -> str:
|
||||
"""Remove systemd directives that older hosts silently drop."""
|
||||
lines = text.splitlines()
|
||||
filtered = []
|
||||
for line in lines:
|
||||
stripped = line.strip()
|
||||
if stripped and not stripped.startswith("#"):
|
||||
key = stripped.split("=", 1)[0].strip()
|
||||
if key in _SYSTEMD_OPTIONAL_DIRECTIVES:
|
||||
continue
|
||||
filtered.append(line)
|
||||
return "\n".join(filtered)
|
||||
|
||||
|
||||
def _normalize_launchd_plist_for_comparison(text: str) -> str:
|
||||
"""Normalize launchd plist text for staleness checks.
|
||||
|
||||
@@ -2500,9 +2523,16 @@ def systemd_unit_is_current(system: bool = False) -> bool:
|
||||
installed = unit_path.read_text(encoding="utf-8")
|
||||
expected_user = _read_systemd_user_from_unit(unit_path) if system else None
|
||||
expected = generate_systemd_unit(system=system, run_as_user=expected_user)
|
||||
return _normalize_service_definition(installed) == _normalize_service_definition(
|
||||
expected
|
||||
# Normalize out directives that older systemd versions silently drop
|
||||
# (RestartMaxDelaySec, RestartSteps) so a unit that differs only by
|
||||
# those directives is not perpetually flagged as outdated.
|
||||
norm_installed = _normalize_service_definition(
|
||||
_strip_optional_systemd_directives(installed)
|
||||
)
|
||||
norm_expected = _normalize_service_definition(
|
||||
_strip_optional_systemd_directives(expected)
|
||||
)
|
||||
return norm_installed == norm_expected
|
||||
|
||||
|
||||
def refresh_systemd_unit_if_needed(system: bool = False) -> bool:
|
||||
|
||||
@@ -58,6 +58,7 @@ AUTHOR_MAP = {
|
||||
"129007007+HeLLGURD@users.noreply.github.com": "HeLLGURD",
|
||||
"290859878+synapsesx@users.noreply.github.com": "synapsesx",
|
||||
"dirtyren@users.noreply.github.com": "dirtyren",
|
||||
"islam666@users.noreply.github.com": "islam666",
|
||||
"zhaolei.vc@bytedance.com": "zhaoleibd",
|
||||
"jeffrobodie@gmail.com": "jeffrobodie-glitch",
|
||||
"kyssta-exe@users.noreply.github.com": "kyssta-exe",
|
||||
|
||||
247
tests/hermes_cli/test_systemd_optional_directives.py
Normal file
247
tests/hermes_cli/test_systemd_optional_directives.py
Normal file
@@ -0,0 +1,247 @@
|
||||
"""Tests for systemd optional-directive normalization (issue #41119).
|
||||
|
||||
On older systemd versions that don't support RestartMaxDelaySec /
|
||||
RestartSteps, the installed unit file has those directives silently
|
||||
dropped. Without normalization, systemd_unit_is_current() would
|
||||
perpetually report the unit as outdated because the strict text
|
||||
comparison sees a difference.
|
||||
|
||||
The fix: _strip_optional_systemd_directives() removes those directives
|
||||
from both the installed and expected text before comparison.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _strip_optional_systemd_directives
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestStripOptionalSystemdDirectives:
|
||||
def test_removes_restart_max_delay_sec(self):
|
||||
from hermes_cli.gateway import _strip_optional_systemd_directives
|
||||
text = """[Service]
|
||||
Restart=always
|
||||
RestartSec=5
|
||||
RestartMaxDelaySec=300
|
||||
RestartSteps=5
|
||||
"""
|
||||
result = _strip_optional_systemd_directives(text)
|
||||
assert "RestartMaxDelaySec" not in result
|
||||
assert "RestartSteps" not in result
|
||||
assert "Restart=always" in result
|
||||
assert "RestartSec=5" in result
|
||||
|
||||
def test_preserves_other_directives(self):
|
||||
from hermes_cli.gateway import _strip_optional_systemd_directives
|
||||
text = """[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python gateway run
|
||||
Restart=always
|
||||
RestartSec=5
|
||||
KillMode=mixed
|
||||
KillSignal=SIGTERM
|
||||
"""
|
||||
result = _strip_optional_systemd_directives(text)
|
||||
assert "Type=simple" in result
|
||||
assert "ExecStart=" in result
|
||||
assert "KillMode=mixed" in result
|
||||
assert "KillSignal=SIGTERM" in result
|
||||
|
||||
def test_handles_empty_string(self):
|
||||
from hermes_cli.gateway import _strip_optional_systemd_directives
|
||||
assert _strip_optional_systemd_directives("") == ""
|
||||
|
||||
def test_handles_no_optional_directives(self):
|
||||
from hermes_cli.gateway import _strip_optional_systemd_directives
|
||||
text = "[Service]\nRestart=always\n"
|
||||
result = _strip_optional_systemd_directives(text)
|
||||
assert "Restart=always" in result
|
||||
assert "RestartMaxDelaySec" not in result
|
||||
|
||||
def test_preserves_comments(self):
|
||||
from hermes_cli.gateway import _strip_optional_systemd_directives
|
||||
text = """[Service]
|
||||
# RestartMaxDelaySec is set below
|
||||
RestartMaxDelaySec=300
|
||||
"""
|
||||
result = _strip_optional_systemd_directives(text)
|
||||
# The comment line should be preserved
|
||||
assert "# RestartMaxDelaySec" in result
|
||||
# The actual directive should be removed
|
||||
assert "RestartMaxDelaySec=300" not in result
|
||||
|
||||
def test_handles_inline_values_with_equals(self):
|
||||
from hermes_cli.gateway import _strip_optional_systemd_directives
|
||||
text = "RestartMaxDelaySec=300\n"
|
||||
result = _strip_optional_systemd_directives(text)
|
||||
assert result == ""
|
||||
|
||||
def test_full_unit_comparison(self):
|
||||
"""Simulate the full stale-check flow with an older systemd unit."""
|
||||
from hermes_cli.gateway import (
|
||||
_normalize_service_definition,
|
||||
_strip_optional_systemd_directives,
|
||||
)
|
||||
# What the installed unit looks like on older systemd (directives stripped)
|
||||
installed = """[Unit]
|
||||
Description=Hermes Gateway
|
||||
After=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python -m hermes_cli.main gateway run
|
||||
Restart=always
|
||||
RestartSec=5
|
||||
KillMode=mixed
|
||||
KillSignal=SIGTERM
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
"""
|
||||
# What generate_systemd_unit produces (with the directives)
|
||||
expected = """[Unit]
|
||||
Description=Hermes Gateway
|
||||
After=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python -m hermes_cli.main gateway run
|
||||
Restart=always
|
||||
RestartSec=5
|
||||
RestartMaxDelaySec=300
|
||||
RestartSteps=5
|
||||
KillMode=mixed
|
||||
KillSignal=SIGTERM
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
"""
|
||||
# Without normalization, they differ
|
||||
assert _normalize_service_definition(installed) != _normalize_service_definition(expected)
|
||||
|
||||
# With optional-directive stripping, they match
|
||||
norm_installed = _normalize_service_definition(
|
||||
_strip_optional_systemd_directives(installed)
|
||||
)
|
||||
norm_expected = _normalize_service_definition(
|
||||
_strip_optional_systemd_directives(expected)
|
||||
)
|
||||
assert norm_installed == norm_expected
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# systemd_unit_is_current integration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestSystemdUnitIsCurrent:
|
||||
def test_unit_without_optional_directives_is_current(self, tmp_path, monkeypatch):
|
||||
"""Installed unit missing RestartMaxDelaySec/RestartSteps should be
|
||||
considered current when the generated unit includes them."""
|
||||
from hermes_cli import gateway as gw
|
||||
|
||||
installed = """[Unit]
|
||||
Description=Hermes Gateway
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python gateway run
|
||||
Restart=always
|
||||
RestartSec=5
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
"""
|
||||
unit_file = tmp_path / "hermes-gateway.service"
|
||||
unit_file.write_text(installed)
|
||||
|
||||
monkeypatch.setattr(gw, "get_systemd_unit_path", lambda system=False: unit_file)
|
||||
monkeypatch.setattr(
|
||||
gw,
|
||||
"generate_systemd_unit",
|
||||
lambda system=False, run_as_user=None: installed + "\nRestartMaxDelaySec=300\nRestartSteps=5\n",
|
||||
)
|
||||
|
||||
assert gw.systemd_unit_is_current(system=False) is True
|
||||
|
||||
def test_unit_with_different_restart_is_not_current(self, tmp_path, monkeypatch):
|
||||
"""A unit with genuinely different config should still be outdated."""
|
||||
from hermes_cli import gateway as gw
|
||||
|
||||
installed = """[Unit]
|
||||
Description=Hermes Gateway
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python gateway run
|
||||
Restart=always
|
||||
RestartSec=10
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
"""
|
||||
expected = """[Unit]
|
||||
Description=Hermes Gateway
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python gateway run
|
||||
Restart=always
|
||||
RestartSec=5
|
||||
RestartMaxDelaySec=300
|
||||
RestartSteps=5
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
"""
|
||||
unit_file = tmp_path / "hermes-gateway.service"
|
||||
unit_file.write_text(installed)
|
||||
|
||||
monkeypatch.setattr(gw, "get_systemd_unit_path", lambda system=False: unit_file)
|
||||
monkeypatch.setattr(
|
||||
gw,
|
||||
"generate_systemd_unit",
|
||||
lambda system=False, run_as_user=None: expected,
|
||||
)
|
||||
|
||||
assert gw.systemd_unit_is_current(system=False) is False
|
||||
|
||||
def test_unit_with_optional_directives_is_current(self, tmp_path, monkeypatch):
|
||||
"""Installed unit WITH the optional directives should also be current."""
|
||||
from hermes_cli import gateway as gw
|
||||
|
||||
unit_text = """[Unit]
|
||||
Description=Hermes Gateway
|
||||
|
||||
[Service]
|
||||
Type=simple
|
||||
ExecStart=/usr/bin/python gateway run
|
||||
Restart=always
|
||||
RestartSec=5
|
||||
RestartMaxDelaySec=300
|
||||
RestartSteps=5
|
||||
|
||||
[Install]
|
||||
WantedBy=default.target
|
||||
"""
|
||||
unit_file = tmp_path / "hermes-gateway.service"
|
||||
unit_file.write_text(unit_text)
|
||||
|
||||
monkeypatch.setattr(gw, "get_systemd_unit_path", lambda system=False: unit_file)
|
||||
monkeypatch.setattr(
|
||||
gw,
|
||||
"generate_systemd_unit",
|
||||
lambda system=False, run_as_user=None: unit_text,
|
||||
)
|
||||
|
||||
assert gw.systemd_unit_is_current(system=False) is True
|
||||
|
||||
def test_nonexistent_unit_is_not_current(self, tmp_path, monkeypatch):
|
||||
from hermes_cli import gateway as gw
|
||||
unit_file = tmp_path / "nonexistent.service"
|
||||
monkeypatch.setattr(gw, "get_systemd_unit_path", lambda system=False: unit_file)
|
||||
assert gw.systemd_unit_is_current(system=False) is False
|
||||
Reference in New Issue
Block a user