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.
This commit is contained in:
islam666 2026-06-07 09:14:30 +00:00 committed by Teknium
parent b18490b890
commit 18c085b1a4
2 changed files with 279 additions and 2 deletions

View file

@ -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:

View 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