From 18c085b1a4297c5024a192e389bc7202ef40e4a9 Mon Sep 17 00:00:00 2001 From: islam666 Date: Sun, 7 Jun 2026 09:14:30 +0000 Subject: [PATCH] fix(gateway): normalize optional systemd directives in stale-check (#41119) On older systemd versions that don't support RestartMaxDelaySec / RestartSteps, the installed unit file has those directives silently dropped. systemd_unit_is_current() did a strict text comparison, so the unit was perpetually flagged as outdated. Fix: _strip_optional_systemd_directives() removes RestartMaxDelaySec and RestartSteps from both the installed and expected text before comparison. Units that differ only by these optional directives are now correctly considered current. --- hermes_cli/gateway.py | 34 ++- .../test_systemd_optional_directives.py | 247 ++++++++++++++++++ 2 files changed, 279 insertions(+), 2 deletions(-) create mode 100644 tests/hermes_cli/test_systemd_optional_directives.py diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index d1339444800..335505a1e1c 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -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: diff --git a/tests/hermes_cli/test_systemd_optional_directives.py b/tests/hermes_cli/test_systemd_optional_directives.py new file mode 100644 index 00000000000..34aa1793281 --- /dev/null +++ b/tests/hermes_cli/test_systemd_optional_directives.py @@ -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