Compare commits

..
Author SHA1 Message Date
ChuckBuilds 14abea2d24 perf(health): stop rewriting a health record on every healthy cycle
Every successful plugin update called record_success(), which persisted the
record unconditionally. In steady state the only fields that had changed were
total_successes and last_success_time -- a counter and a timestamp that
health_monitor surfaces for display and that nothing reads back after a
restart. Nothing alerts on the age of last_successful_update; it is carried in
the metrics dataclass and shown.

Measured on a rig running 24 plugins, all steady-state (0 consecutive
failures, circuit closed): a five-minute sample caught 22 health-file
rewrites, about 4.4 a minute or 6,300 a day. Each write is ~400 bytes through
cache_manager.set(), which writes a file per call, so each one costs a
filesystem block plus an ext4 journal write.

That lands on an SD card, where the unit of cost is an erase-block cycle
rather than the bytes involved, and where wear is what eventually kills the
card. Two cards have already failed on the other rig with the same
signature -- unreadable block device, EIO on exec, sshd unable to read its
host keys.

The circuit breaker still has to survive a restart, so the write is kept for
exactly the fields it is rebuilt from: consecutive_failures, circuit_state,
circuit_opened_time, half_open_start_time. A failure, a circuit opening and a
recovery are all still written the moment they happen. In-memory state is
updated every time either way, so the health API and web UI show what they
always did.

Tested: 100 healthy cycles now perform zero writes after the first, the
counters remain accurate in memory, and a failure, a recovery and a
half-open-to-closed transition each still reach disk. One test kills and
rebuilds the tracker from the cache to prove the breaker's state genuinely
survives what is no longer written.

Mutation-checked both ways: persisting unconditionally again fails the
steady-state test, and widening _DURABLE_FIELDS to include last_success_time
fails it too. The 46 existing health tests pass.
2026-08-19 20:35:02 -04:00
8 changed files with 135 additions and 466 deletions
@@ -37,10 +37,6 @@ echo " systemctl: $SYSTEMCTL_PATH"
echo ""
echo "Step 1: Configuring sudo permissions for nmcli..."
SUDOERS_FILE="/etc/sudoers.d/ledmatrix_wifi"
SYSCTL_PATH=$(command -v sysctl || echo /usr/sbin/sysctl)
NFT_PATH=$(command -v nft || echo /usr/sbin/nft)
RFKILL_PATH=$(command -v rfkill || echo /usr/sbin/rfkill)
MKDIR_PATH=$(command -v mkdir || echo /usr/bin/mkdir)
# Create a temporary sudoers file using mktemp (handles permissions better)
TEMP_SUDOERS=$(mktemp) || {
@@ -66,36 +62,6 @@ $WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH start dnsmasq
$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH stop dnsmasq
$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart dnsmasq
$WEB_USER ALL=(ALL) NOPASSWD: $SYSTEMCTL_PATH restart NetworkManager
# The captive portal turns IP forwarding on while the access point is up and
# restores the previous value when it comes down (wifi_manager._setup_iptables_
# redirect / _teardown_iptables_redirect). Without this rule that sudo call
# needs a password, so forwarding stays off and clients associate to the AP but
# cannot route. It goes unnoticed on a stock Raspberry Pi image, where
# /etc/sudoers.d/010_pi-nopasswd grants the default user blanket NOPASSWD and
# masks every gap in this file -- it only bites once that blanket rule is
# removed.
$WEB_USER ALL=(ALL) NOPASSWD: $SYSCTL_PATH -w net.ipv4.ip_forward=0
$WEB_USER ALL=(ALL) NOPASSWD: $SYSCTL_PATH -w net.ipv4.ip_forward=1
# The portal's redirect lives in its own nftables table, created when the AP
# comes up and deleted when it goes down, and the radio has to be unblocked
# before the AP can start at all. Same story as the sysctl rules above: called
# with sudo, never granted here, and invisible on a stock Pi image.
$WEB_USER ALL=(ALL) NOPASSWD: $NFT_PATH add table ip ledmatrix
$WEB_USER ALL=(ALL) NOPASSWD: $NFT_PATH delete table ip ledmatrix
$WEB_USER ALL=(ALL) NOPASSWD: $RFKILL_PATH unblock wifi
# NetworkManager's dnsmasq drop-in directory, exact path.
$WEB_USER ALL=(ALL) NOPASSWD: $MKDIR_PATH -p /etc/NetworkManager/dnsmasq-shared.d
#
# iptables is deliberately NOT granted here. Its rules are built from the live
# interface name and port, so a rule covering them needs a trailing wildcard --
# and `iptables --modprobe=/path/to/anything` runs that path as root, so
# `NOPASSWD: iptables *` is a root shell for the web user by another name. That
# is a worse outcome than the gap it would close, which today is masked anyway
# by the blanket NOPASSWD rule on stock Pi images.
#
# Closing it safely means a wrapper script that builds the rules itself and
# takes only an interface and a port, granted the way safe_plugin_rm.sh already
# is. That belongs in its own change rather than being smuggled into this one.
# Allow copying hostapd and dnsmasq config files into place
$WEB_USER ALL=(ALL) NOPASSWD: /usr/bin/cp /tmp/hostapd.conf /etc/hostapd/hostapd.conf
+25 -4
View File
@@ -178,11 +178,21 @@ class PluginHealthTracker:
)
return self._health_state[plugin_id]
# Fields the circuit breaker is rebuilt from after a restart. Everything
# else in a health record is reporting, read only for display.
_DURABLE_FIELDS = ('consecutive_failures', 'circuit_state',
'circuit_opened_time', 'half_open_start_time')
def _durable(self, state: Dict[str, Any]) -> tuple:
"""The part of a health record whose loss would change behaviour."""
return tuple(state.get(field) for field in self._DURABLE_FIELDS)
def record_success(self, plugin_id: str) -> None:
"""Record a successful plugin execution."""
state = self.get_health_state(plugin_id)
current_time = time.time()
durable_before = self._durable(state)
# Reset consecutive failures
state['consecutive_failures'] = 0
state['total_successes'] = state.get('total_successes', 0) + 1
@@ -198,9 +208,20 @@ class PluginHealthTracker:
# Shouldn't happen, but handle it
state['circuit_state'] = CircuitState.CLOSED.value
state['circuit_opened_time'] = None
self._save_health_state(plugin_id, state)
# A healthy plugin reports success every cycle, and in that steady state
# the only fields changed above are a counter and a timestamp that
# nothing reads back after a restart. Persisting them anyway rewrites a
# small file per plugin per cycle: on a rig running 24 plugins, a
# five-minute sample measured 22 rewrites, about 4.4 a minute or 6,300 a
# day. Those land on an SD card, where the cost is an erase-block cycle
# rather than the 400 bytes involved, and where wear is what eventually
# kills the card.
# In-memory state is still updated every time, so the health API and web
# UI show exactly what they did before; only the write is skipped.
if self._durable(state) != durable_before:
self._save_health_state(plugin_id, state)
def record_failure(self, plugin_id: str, error: Optional[Exception] = None) -> None:
"""Record a failed plugin execution."""
state = self.get_health_state(plugin_id)
-68
View File
@@ -62,9 +62,6 @@ class StartupValidator:
# Validate plugins if plugin manager is available
if self.plugin_manager:
self._validate_plugins()
# Warn when the running systemd unit no longer matches the repo's
self._validate_systemd_units()
is_valid = len(self.errors) == 0
@@ -77,71 +74,6 @@ class StartupValidator:
return (is_valid, self.errors.copy(), self.warnings.copy())
#: Units this project installs, and where each is installed to.
_UNITS = (
("systemd/ledmatrix.service", "/etc/systemd/system/ledmatrix.service"),
("systemd/ledmatrix-web.service", "/etc/systemd/system/ledmatrix-web.service"),
)
def _validate_systemd_units(self) -> None:
"""Warn when an installed unit has drifted from the repo's template.
Nothing re-applies these after the first install. `git pull` -- which is
what the web UI's update button runs -- brings a new template into the
checkout, but nothing copies it to /etc/systemd/system and nothing runs
`systemctl daemon-reload`, so the unit that actually runs is whatever
first_time_install.sh wrote on day one.
That makes every hardening added to a unit inert on existing installs.
Measured on one rig: the installed unit was thirteen days older than the
repo's and differed in content, so a MemoryMax the repo had specified
was not being enforced at all -- `systemctl show` reported
MemoryMax=infinity.
A warning rather than an error, and certainly not a silent rewrite:
editing files under /etc and restarting services is the installer's job,
not something a display process should do to a machine while it boots.
The remedy is to re-run scripts/install/install_service.sh.
"""
try:
project_root = Path(__file__).resolve().parent.parent
for template_rel, installed_path in self._UNITS:
template = project_root / template_rel
installed = Path(installed_path)
if not template.is_file() or not installed.is_file():
continue
# The template carries placeholders the installer substitutes,
# so compare the substituted form rather than the raw file.
expected = template.read_text(encoding="utf-8")
expected = expected.replace("__PROJECT_ROOT_DIR__", str(project_root))
expected = expected.replace("__USER__", "root")
try:
actual = installed.read_text(encoding="utf-8")
except PermissionError:
continue
if self._unit_body(expected) != self._unit_body(actual):
self.warnings.append(
f"{installed.name} differs from {template_rel}; the "
"installed unit is not refreshed by an update, so "
"settings added to the template are not in effect. "
"Re-run scripts/install/install_service.sh to apply them."
)
except OSError as e:
self.logger.debug("Could not compare systemd units: %s", e)
@staticmethod
def _unit_body(text: str) -> str:
"""A unit's meaningful lines: no comments, no blanks, no ordering noise."""
lines = []
for line in text.splitlines():
line = line.strip()
if line and not line.startswith("#"):
lines.append(line)
return "\n".join(sorted(lines))
def _validate_config(self) -> None:
"""Validate configuration files."""
try:
-12
View File
@@ -8,18 +8,6 @@ Type=simple
User=root
WorkingDirectory=__PROJECT_ROOT_DIR__
Environment=PYTHONDONTWRITEBYTECODE=1
# glibc gives each allocating thread its own malloc arena, up to 8 x CPU count,
# and an arena that has grown is never handed back to the OS. This process runs
# 9 threads on a 3-core Pi, so the ceiling is 24 arenas -- and a rig measured at
# 1030 MB resident held 23 large anonymous mappings on 64 MB-aligned addresses,
# 920 MB of them, while the live data it was actually holding (widest scroll
# strip seen: 35,746 x 64) accounts for roughly 15 MB. That gap is arena bloat,
# not leaked objects: RSS was flat across repeated sampling, not climbing.
#
# Capping the arenas trades a little allocator concurrency for a large amount of
# resident memory on a device that has neither to spare. 2 is the usual value;
# raise it if frame times regress.
Environment=MALLOC_ARENA_MAX=2
ExecStart=/usr/bin/python3 __PROJECT_ROOT_DIR__/run.py
# Restart=always, not on-failure: run.py exiting 0 (a clean shutdown path taken
# for a reason that no longer applies, e.g. a config reload) would otherwise leave
+110
View File
@@ -0,0 +1,110 @@
"""A healthy plugin must not rewrite its health record every cycle.
Every successful plugin update called record_success(), which persisted the
record unconditionally. In steady state the only fields that had changed were
total_successes and last_success_time -- a counter and a timestamp that
health_monitor reads for display and that nothing reads back after a restart.
Measured on a rig running 24 plugins: about 17 health-file rewrites a minute,
roughly 25,000 a day. Each is ~400 bytes, but they land on an SD card where
the unit of cost is an erase-block cycle, not the byte count, and where wear is
what eventually kills the card.
The circuit breaker still needs its own state to survive a restart, so the
write is kept for exactly the fields it is rebuilt from -- and a failure, a
circuit opening, or a recovery must still be written the moment it happens.
"""
import time
import pytest
from src.plugin_system.plugin_health import PluginHealthTracker, CircuitState
class _Cache:
"""Counts writes; serves back whatever was last written."""
def __init__(self):
self.store = {}
self.writes = 0
def set(self, key, data, ttl=None, **kwargs):
self.writes += 1
self.store[key] = data
def get(self, key, max_age=None, memory_ttl=None, **kwargs):
return self.store.get(key)
@pytest.fixture
def tracker():
cache = _Cache()
t = PluginHealthTracker(cache_manager=cache)
return t, cache
def test_steady_state_success_stops_writing(tracker):
"""The regression: 100 healthy cycles used to be 100 SD writes."""
t, cache = tracker
t.record_success("weather")
first = cache.writes
for _ in range(100):
t.record_success("weather")
assert cache.writes == first, (
f"{cache.writes - first} redundant writes across 100 healthy cycles"
)
def test_the_counters_are_still_accurate_in_memory(tracker):
"""Skipping the write must not skip the bookkeeping."""
t, _ = tracker
for _ in range(10):
t.record_success("weather")
state = t.get_health_state("weather")
assert state["total_successes"] == 10
assert state["last_success_time"] is not None
assert state["last_success_time"] <= time.time()
def test_a_failure_is_written_immediately(tracker):
t, cache = tracker
t.record_success("weather")
before = cache.writes
t.record_failure("weather", RuntimeError("boom"))
assert cache.writes > before, "a failure must reach disk"
def test_recovery_after_failure_is_written(tracker):
"""consecutive_failures returning to 0 is durable state changing."""
t, cache = tracker
t.record_failure("weather", RuntimeError("boom"))
before = cache.writes
t.record_success("weather")
assert cache.writes > before, "recovery must reach disk"
assert t.get_health_state("weather")["consecutive_failures"] == 0
def test_a_closing_circuit_is_written(tracker):
"""Success in half-open closes the circuit -- that must survive a restart."""
t, cache = tracker
state = t.get_health_state("weather")
state["circuit_state"] = CircuitState.HALF_OPEN.value
state["half_open_start_time"] = time.time()
before = cache.writes
t.record_success("weather")
assert cache.writes > before, "a circuit transition must reach disk"
assert t.get_health_state("weather")["circuit_state"] == CircuitState.CLOSED.value
def test_durable_state_survives_a_restart(tracker):
"""What is skipped must genuinely not matter to the breaker."""
t, cache = tracker
for _ in range(3):
t.record_failure("weather", RuntimeError("boom"))
for _ in range(50):
t.record_success("weather")
revived = PluginHealthTracker(cache_manager=cache)
state = revived.get_health_state("weather")
assert state["consecutive_failures"] == 0
assert state["circuit_state"] == CircuitState.CLOSED.value
-120
View File
@@ -1,120 +0,0 @@
"""The captive portal's fixed-argument sudo calls must be granted.
The installers write two allow-lists, /etc/sudoers.d/ledmatrix_web and
ledmatrix_wifi. A sudo call absent from both needs a password, which a service
cannot supply, so it fails.
Four such calls were ungranted, all of them captive-portal teardown/setup:
sysctl -w net.ipv4.ip_forward=0|1 wifi_manager.py:788, 883
nft add|delete table ip ledmatrix wifi_manager.py:835, 895
rfkill unblock wifi wifi_manager.py:1811
mkdir -p .../dnsmasq-shared.d wifi_manager.py:922
It goes unnoticed because a stock Raspberry Pi image ships
/etc/sudoers.d/010_pi-nopasswd granting the default user
`ALL=(ALL) NOPASSWD: ALL`, which satisfies every gap in both files. It only
bites once that blanket rule is removed or the service runs as another user.
Scope, deliberately narrow: this pins the four commands above, each of which
can be written out literally. The portal makes further sudo calls whose
arguments are built at runtime -- iptables and nft rules carrying an interface
name and a port, `ip addr`, `ip link` -- and those cannot be granted safely
here. A rule covering them needs a trailing wildcard, and
`iptables --modprobe=/path/to/anything` runs that path as root, so
`NOPASSWD: iptables *` is a root shell for the web user by another name.
Closing that half needs a privileged helper that builds the rules itself and
takes only an interface and a port, granted the way safe_plugin_rm.sh already
is. That is a design decision, not a one-line grant, and belongs in its own
change.
"""
import re
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parent.parent
INSTALLERS = (
ROOT / "first_time_install.sh",
ROOT / "scripts" / "install" / "configure_wifi_permissions.sh",
)
#: Commands this change grants, each fully literal in the source.
REQUIRED = (
("sysctl", "-w", "net.ipv4.ip_forward=0"),
("sysctl", "-w", "net.ipv4.ip_forward=1"),
("nft", "add", "table", "ip", "ledmatrix"),
("nft", "delete", "table", "ip", "ledmatrix"),
("rfkill", "unblock", "wifi"),
("mkdir", "-p", "/etc/NetworkManager/dnsmasq-shared.d"),
)
#: Tools with an option that executes a program of the caller's choosing.
#: A trailing wildcard on any of these is a privilege escalation.
EXEC_CAPABLE = ("iptables", "ip6tables", "nft", "tcpdump", "find", "awk",
"sed", "perl", "python", "python3", "env")
def _grant_lines():
lines = []
for installer in INSTALLERS:
if not installer.is_file():
continue
for line in installer.read_text(encoding="utf-8", errors="replace").splitlines():
if "NOPASSWD:" in line:
lines.append(line.split("NOPASSWD:", 1)[1])
return lines
def _normalised_grants():
"""Grants with binary-path variables reduced to tool names.
Rules are written as `$SYSCTL_PATH -w ...`, so matching the literal
"sysctl" finds nothing and every rule looks absent -- which is exactly how
an earlier version of this test reported six gaps that did not exist.
Only NOPASSWD lines are considered, because taking the whole script let a
variable definition such as NFT_PATH=$(command -v nft) satisfy the check on
its own while the grant itself had been deleted.
"""
text = "\n".join(_grant_lines())
text = re.sub(r"\$\{?([A-Z][A-Z0-9_]*)_PATH\}?", lambda m: m.group(1).lower(), text)
return re.sub(r"/usr/(?:s?bin)/", "", text)
def test_the_installers_are_present():
missing = [str(p.relative_to(ROOT)) for p in INSTALLERS if not p.is_file()]
assert not missing, f"installer(s) missing: {missing}"
@pytest.mark.parametrize("command", REQUIRED, ids=lambda c: " ".join(c))
def test_the_command_is_granted(command):
"""Whole command, not just the binary.
Checking only the binary made this far weaker than it looked: with
`sysctl` present anywhere, deleting the ip_forward=0 grant still passed,
and the portal would then be unable to restore forwarding on teardown.
"""
pattern = r"\s+".join(re.escape(word) for word in command)
assert re.search(pattern, _normalised_grants()), (
f"no installer grants `{' '.join(command)}`")
def test_no_wildcard_on_a_tool_that_can_exec():
"""`NOPASSWD: iptables *` hands the web user root.
iptables --modprobe=/path runs that path as root. This caught a grant added
in this very change, which is why it is here.
"""
offenders = []
for rule in _grant_lines():
rule = rule.strip()
if not rule.endswith("*"):
continue
haystack = rule.replace("_PATH", "").lower()
for tool in EXEC_CAPABLE:
if re.search(rf"(^|/|\s|\$){tool}(\s|$)", haystack):
offenders.append(rule)
break
assert not offenders, (
"wildcard grant on a tool that can execute another program:\n "
+ "\n ".join(offenders))
-95
View File
@@ -1,95 +0,0 @@
"""The display unit must cap glibc's malloc arenas.
glibc hands each allocating thread its own malloc arena, up to 8 x CPU count,
and an arena that has grown is never returned to the OS. This process runs
threads for the render loop, the update workers and the background fetchers, so
on a 3-core Pi the ceiling is 24 arenas.
Measured on a live rig, 2.5 hours in:
RSS 1030 MB
Private_Dirty 988 MB
anonymous mappings > 10 MB 23 (ceiling is 8 x 3 = 24)
largest few 104, 79, 66, 63, 63 MB, on 64 MB-aligned addresses
against live data that accounts for perhaps 15 MB -- the widest scroll strip
observed was 35,746 x 64, about 7 MB as RGB and the same again for its numpy
mirror. Repeated sampling showed RSS flat between 990 and 1030 MB rather than
climbing, so this is arena bloat rather than a leak: memory Python has freed
but glibc is holding per-arena.
The device had 59 MB free at the time.
Capping the arena count trades a little allocator concurrency for that resident
memory. The render loop is latency-sensitive, so if p99 frame time regresses the
right response is to raise this rather than remove it.
"""
import re
from pathlib import Path
import pytest
UNIT = (Path(__file__).resolve().parent.parent / "systemd" / "ledmatrix.service")
#: The value the unit is expected to carry. 2 is the usual choice for a
#: threaded Python process; 1-4 all keep some of the saving, but only one of
#: them is what this project ships.
EXPECTED_ARENA_MAX = 2
def _environment(unit_text):
return dict(
line.split("=", 2)[1:3] if line.count("=") >= 2 else (line.split("=", 1)[1], "")
for line in unit_text.splitlines()
if line.startswith("Environment=")
)
def test_the_unit_exists():
assert UNIT.is_file(), f"{UNIT} is missing"
def test_malloc_arena_max_is_capped():
env = _environment(UNIT.read_text(encoding="utf-8"))
assert "MALLOC_ARENA_MAX" in env, (
"the display unit does not cap glibc arenas; on a 3-core Pi the default "
"ceiling is 24 and a measured rig held 23 of them, 920 MB"
)
value = int(env["MALLOC_ARENA_MAX"])
# Pinned, not a range. A range let a change to 4 -- which hands most of the
# saving back -- pass unnoticed, which was the point of the finding that
# prompted this. Raising it is a legitimate response to a frame-time
# regression, but it should be a visible edit here rather than a silent
# drift, so the number lives in one place and changing it shows up in
# review.
assert value == EXPECTED_ARENA_MAX, (
f"MALLOC_ARENA_MAX={value}, expected {EXPECTED_ARENA_MAX}. If this was "
"raised deliberately because frame times regressed, update "
"EXPECTED_ARENA_MAX here and say so in the commit."
)
def test_the_reason_is_recorded_next_to_it():
"""A bare tuning knob invites removal by whoever meets it next."""
text = UNIT.read_text(encoding="utf-8")
index = text.index("Environment=MALLOC_ARENA_MAX")
preamble = text[:index].splitlines()[-12:]
comment = "\n".join(line for line in preamble if line.startswith("#"))
assert "arena" in comment.lower(), "no explanation precedes the setting"
assert re.search(r"\d", comment), (
"the explanation cites no measurement, so a reader cannot tell whether "
"it still applies to their hardware"
)
@pytest.mark.parametrize("unit", ["ledmatrix.service"])
def test_the_unit_still_parses_as_ini(unit):
"""systemd will refuse a malformed unit, and the panel stays dark."""
import configparser
path = UNIT.parent / unit
parser = configparser.ConfigParser(strict=False)
# systemd allows repeated keys; ConfigParser needs them merged, not rejected.
parser.read_string(path.read_text(encoding="utf-8"))
assert parser.has_section("Service")
assert parser.has_option("Service", "ExecStart")
-133
View File
@@ -1,133 +0,0 @@
"""An installed unit that no longer matches the repo's must be reported.
Nothing re-applies systemd units after the first install. `git pull` -- what
the web UI's update button runs -- brings a new template into the checkout, but
no code in web_interface/ or src/ copies it to /etc/systemd/system or runs
`systemctl daemon-reload`. The unit that actually runs is whatever
first_time_install.sh wrote on day one.
So every hardening added to a unit is inert on existing installs. Measured on a
live rig: the installed unit was dated 2026-08-06 and the repo's 2026-08-19,
and they differed -- with the result that a MemoryMax=85% present in the repo's
template was not being enforced at all. `systemctl show` reported
MemoryMax=infinity.
This is a warning, not an error, and deliberately not a silent rewrite:
editing files under /etc and restarting services is the installer's job, not
something a display process should do to a machine while it boots.
"""
import logging
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from src.startup_validator import StartupValidator
@pytest.fixture
def validator():
v = StartupValidator(config_manager=MagicMock())
v.logger = logging.getLogger("test")
v.warnings = []
v.errors = []
return v
def test_a_matching_unit_produces_no_warning(validator, tmp_path):
"""The installed unit, substituted exactly as the installer would."""
project_root = Path("src/startup_validator.py").resolve().parent.parent
template_rel = "systemd/ledmatrix.service"
template = project_root / template_rel
if not template.is_file():
pytest.skip("repo unit template not present")
installed = tmp_path / "ledmatrix.service"
installed.write_text(
template.read_text(encoding="utf-8")
.replace("__PROJECT_ROOT_DIR__", str(project_root))
.replace("__USER__", "root"),
encoding="utf-8")
validator._UNITS = ((template_rel, str(installed)),)
validator._validate_systemd_units()
assert not validator.warnings, f"a matching unit warned: {validator.warnings}"
assert not validator.errors
def test_comments_and_blank_lines_are_not_drift():
"""Otherwise every comment the repo adds would look like a changed unit."""
a = "[Service]\n# explains a setting\nExecStart=/x\nRestart=always\n"
b = "[Service]\nExecStart=/x\n\nRestart=always\n"
assert StartupValidator._unit_body(a) == StartupValidator._unit_body(b)
def test_a_changed_directive_is_drift():
a = "[Service]\nExecStart=/x\nMemoryMax=85%\n"
b = "[Service]\nExecStart=/x\n"
assert StartupValidator._unit_body(a) != StartupValidator._unit_body(b)
def test_reordered_directives_are_not_drift():
"""systemd does not care about order within a section, so neither should this."""
a = "[Service]\nExecStart=/x\nRestart=always\n"
b = "[Service]\nRestart=always\nExecStart=/x\n"
assert StartupValidator._unit_body(a) == StartupValidator._unit_body(b)
def test_cosmetic_differences_do_not_warn(validator, tmp_path):
"""Through the real comparison, not the helper.
The repo's template carries explanatory comments the installed copy may not
have, and the installer does not preserve ordering or blank lines. If those
counted as drift, every boot would warn and the warning would be ignored.
Asserting this on _unit_body alone would not catch a comparison that stopped
calling it -- which is exactly what a careless edit does.
"""
project_root = Path("src/startup_validator.py").resolve().parent.parent
template_rel = "systemd/ledmatrix.service"
template = project_root / template_rel
if not template.is_file():
pytest.skip("repo unit template not present")
substituted = (template.read_text(encoding="utf-8")
.replace("__PROJECT_ROOT_DIR__", str(project_root))
.replace("__USER__", "root"))
# Same directives, stripped of comments and blank lines and reordered.
directives = sorted(line.strip() for line in substituted.splitlines()
if line.strip() and not line.strip().startswith("#"))
installed = tmp_path / "ledmatrix.service"
installed.write_text("\n".join(reversed(directives)) + "\n", encoding="utf-8")
validator._UNITS = ((template_rel, str(installed)),)
validator._validate_systemd_units()
assert not validator.warnings, (
f"cosmetic-only difference reported as drift: {validator.warnings}")
def test_drift_is_reported_as_a_warning(validator, tmp_path):
"""The whole point: a real difference must surface, and only as a warning."""
installed = tmp_path / "ledmatrix.service"
installed.write_text("[Service]\nExecStart=/usr/bin/python3 /x/run.py\n")
project_root = Path("src/startup_validator.py").resolve().parent.parent
template_rel = "systemd/ledmatrix.service"
template = project_root / template_rel
if not template.is_file():
pytest.skip("repo unit template not present")
validator._UNITS = ((template_rel, str(installed)),)
validator._validate_systemd_units()
assert validator.warnings, "a differing unit produced no warning"
assert "install_service.sh" in validator.warnings[0], (
"the warning does not tell the user how to fix it")
assert not validator.errors, "drift must not be fatal at startup"
def test_a_missing_installed_unit_is_silent(validator, tmp_path):
"""Development checkouts have no /etc/systemd unit; that is not drift."""
validator._UNITS = (("systemd/ledmatrix.service", str(tmp_path / "absent.service")),)
validator._validate_systemd_units()
assert not validator.warnings
assert not validator.errors