Compare commits

...

2 Commits

Author SHA1 Message Date
teknium1
6ea931025e chore(release): map islam666 for salvaged PR #41131 2026-06-07 06:34:06 -07:00
islam666
6509f9c3ac 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.
2026-06-07 06:34:06 -07:00
3 changed files with 280 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

@@ -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",

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