rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-25 18:40:06 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

Controller: keep every wrolpi-* service visible in the UI.

Browse source 
show_only_when_running hid wrolpi-upgrade the moment it exited — exactly
when its logs matter after a failed upgrade — and hid wrolpi-api-dev
whenever it was stopped.  Reinterpret the flag: hide a service only when
its unit is not installed on the box (LoadState=not-found).  Discovery
now unions loaded units (--all, catching transient and exited units)
with installed unit files, and discovered services are listed regardless
of running state.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Roland 2026-07-06 20:06:20 -06:00
parent 910503b641
commit 51a12b199e
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
2 changed files with 131 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

81
controller/lib/systemd.py
View file

@ -107,6 +107,16 @@ def get_service_status(name: str) -> dict:
timeout=5,
)
is_enabled = enabled_result.stdout.strip()
# Loaded state distinguishes a stopped unit from one that does not
# exist on this box at all.
load_result = subprocess.run(
["systemctl", "show", "-p", "LoadState", "--value", systemd_name],
capture_output=True,
text=True,
timeout=5,
)
load_state = load_result.stdout.strip()
except (subprocess.TimeoutExpired, FileNotFoundError):
return {
"name": name,
@ -114,6 +124,7 @@ def get_service_status(name: str) -> dict:
"status": "unknown",
"active": "unknown",
"enabled": False,
"installed": False,
"port": service_config.get("port"),
"viewable": service_config.get("viewable", False),
"view_path": service_config.get("view_path", ""),
@ -138,6 +149,7 @@ def get_service_status(name: str) -> dict:
"name": name,
"systemd_name": systemd_name,
"status": status,
"installed": load_state not in ("not-found", ""),
"active": is_active,
"enabled": is_enabled == "enabled",
"port": service_config.get("port"),
@ -148,41 +160,45 @@ def get_service_status(name: str) -> dict:
}
def discover_running_wrolpi_services() -> list[str]:
def discover_wrolpi_services() -> list[str]:
"""
Discover running wrolpi-* systemd services not in the managed services config.
Discover wrolpi-* systemd services not in the managed services config.
Unions loaded units (--all also catches transient units and units that
have exited this boot) with installed unit files, so a finished oneshot
(e.g. wrolpi-upgrade) or a dev service stays visible in the UI and its
logs remain reachable.
Returns:
list of systemd unit names (e.g., ["wrolpi-fix-media-permissions.service"])
list of service names without the .service suffix
"""
managed_systemd_names = {
s.get("systemd_name", s["name"])
for s in get_managed_services()
}
try:
result = subprocess.run(
["systemctl", "list-units", "wrolpi-*", "--type=service", "--no-legend", "--no-pager"],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode != 0:
return []
except (subprocess.TimeoutExpired, FileNotFoundError):
return []
commands = (
["systemctl", "list-units", "wrolpi-*", "--type=service",
"--all", "--plain", "--no-legend", "--no-pager"],
["systemctl", "list-unit-files", "wrolpi-*", "--type=service",
"--no-legend", "--no-pager"],
)
discovered = []
for line in result.stdout.strip().splitlines():
parts = line.split()
if not parts:
discovered: list[str] = []
for cmd in commands:
try:
result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
except (subprocess.TimeoutExpired, FileNotFoundError):
continue
# Unit name is the first column, e.g. "wrolpi-fix-media-permissions.service"
unit_name = parts[0]
# Strip .service suffix to get the base name
service_name = unit_name.removesuffix(".service")
if service_name not in managed_systemd_names:
discovered.append(service_name)
if result.returncode != 0:
continue
for line in result.stdout.strip().splitlines():
unit_name = next((p for p in line.split() if p.endswith(".service")), None)
if not unit_name:
continue
service_name = unit_name.removesuffix(".service")
if service_name not in managed_systemd_names and service_name not in discovered:
discovered.append(service_name)
return discovered
@ -258,7 +274,10 @@ def get_all_services_status() -> list[dict]:
Get status of all managed services, plus any running wrolpi-* services
discovered dynamically.
Services with show_only_when_running=True are excluded when not running.
Services with show_only_when_running=True are hidden only when their
unit is not installed on this box (e.g. wrolpi-api-dev on production).
Installed-but-stopped services stay visible so a finished oneshot
(wrolpi-upgrade) keeps its logs reachable from the UI.
Returns:
list of service status dicts
@ -266,16 +285,14 @@ def get_all_services_status() -> list[dict]:
results = []
for service in get_managed_services():
status = get_service_status(service["name"])
# Skip services with show_only_when_running if not running
if service.get("show_only_when_running") and status.get("status") != "running":
if service.get("show_only_when_running") and not status.get("installed"):
continue
results.append(status)
# Discover any running wrolpi-* services not in the managed list.
for name in discover_running_wrolpi_services():
status = get_discovered_service_status(name)
if status.get("status") == "running":
results.append(status)
# Any other wrolpi-* services (installed or transient), regardless of
# running state.
for name in discover_wrolpi_services():
results.append(get_discovered_service_status(name))
return results

98
controller/test/test_systemd.py
View file

@ -10,8 +10,8 @@ from controller.lib.systemd import (
get_managed_services,
get_service_config,
get_service_status,
discover_wrolpi_services,
get_all_services_status,
discover_running_wrolpi_services,
get_discovered_service_status,
_get_systemd_name,
start_service,
@ -149,7 +149,7 @@ class TestGetAllServicesStatus:
def test_returns_list(self):
"""Should return a list."""
with mock.patch("controller.lib.systemd.discover_running_wrolpi_services", return_value=[]):
with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
with mock.patch("subprocess.run") as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
result = get_all_services_status()
@ -157,7 +157,7 @@ class TestGetAllServicesStatus:
def test_returns_status_for_each_visible_service(self):
"""Should return status for each visible managed service."""
with mock.patch("controller.lib.systemd.discover_running_wrolpi_services", return_value=[]):
with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
with mock.patch("subprocess.run") as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
result = get_all_services_status()
@ -165,21 +165,37 @@ class TestGetAllServicesStatus:
# All services should be returned when they're all "running" (active)
assert len(result) == len(managed)
def test_filters_show_only_when_running_services_when_stopped(self):
"""Should exclude show_only_when_running services when they're not running."""
with mock.patch("controller.lib.systemd.discover_running_wrolpi_services", return_value=[]):
with mock.patch("subprocess.run") as mock_run:
# Simulate all services as "inactive" (stopped)
mock_run.return_value = mock.Mock(returncode=0, stdout="inactive", stderr="")
def test_includes_stopped_but_installed_services(self):
"""show_only_when_running services stay visible while stopped as long
as the unit is installed: a finished oneshot (wrolpi-upgrade) must
keep its logs reachable from the UI."""
def side_effect(cmd, **kwargs):
out = "loaded" if cmd[:2] == ["systemctl", "show"] else "inactive"
return mock.Mock(returncode=0, stdout=out, stderr="")
with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
with mock.patch("subprocess.run", side_effect=side_effect):
result = get_all_services_status()
assert len(result) == len(get_managed_services())
assert "wrolpi-upgrade" in [s["name"] for s in result]
def test_hides_optional_services_only_when_not_installed(self):
"""show_only_when_running services are hidden only when the unit does
not exist on this box (e.g. wrolpi-api-dev on production)."""
def side_effect(cmd, **kwargs):
out = "not-found" if cmd[:2] == ["systemctl", "show"] else "inactive"
return mock.Mock(returncode=0, stdout=out, stderr="")
with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
with mock.patch("subprocess.run", side_effect=side_effect):
result = get_all_services_status()
managed = get_managed_services()
# Count services without show_only_when_running flag
always_visible = [s for s in managed if not s.get("show_only_when_running")]
assert len(result) == len(always_visible)
def test_includes_show_only_when_running_services_when_running(self):
"""Should include show_only_when_running services when they're running."""
with mock.patch("controller.lib.systemd.discover_running_wrolpi_services", return_value=[]):
with mock.patch("controller.lib.systemd.discover_wrolpi_services", return_value=[]):
with mock.patch("subprocess.run") as mock_run:
# Simulate all services as "active" (running)
mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="")
@ -191,6 +207,56 @@ class TestGetAllServicesStatus:
service_names = [s["name"] for s in result]
assert "wrolpi-upgrade" in service_names
def test_includes_discovered_services_even_when_stopped(self):
"""Discovered wrolpi-* units (e.g. a finished script service) appear
in the list regardless of running state."""
def side_effect(cmd, **kwargs):
out = "loaded" if cmd[:2] == ["systemctl", "show"] else "inactive"
return mock.Mock(returncode=0, stdout=out, stderr="")
with mock.patch("controller.lib.systemd.discover_wrolpi_services",
return_value=["wrolpi-custom-task"]):
with mock.patch("subprocess.run", side_effect=side_effect):
result = get_all_services_status()
names = [s["name"] for s in result]
assert "wrolpi-custom-task" in names
class TestDiscoverWrolpiServices:
"""Discovery unions loaded units (--all, catches transient units) with
installed unit files, so exited oneshots and dev services are visible."""
def _side_effect(self, list_units_out, list_unit_files_out):
def side_effect(cmd, **kwargs):
if cmd[:2] == ["systemctl", "list-units"]:
out = list_units_out
elif cmd[:2] == ["systemctl", "list-unit-files"]:
out = list_unit_files_out
else:
out = ""
return mock.Mock(returncode=0, stdout=out, stderr="")
return side_effect
def test_unions_loaded_and_installed_units(self):
list_units = "wrolpi-transient.service loaded active running Task\n"
list_unit_files = ("wrolpi-custom.service static -\n"
"wrolpi-api.service enabled enabled\n")
with mock.patch("subprocess.run",
side_effect=self._side_effect(list_units, list_unit_files)):
found = discover_wrolpi_services()
assert "wrolpi-transient" in found
assert "wrolpi-custom" in found
# Managed services are excluded from discovery.
assert "wrolpi-api" not in found
def test_deduplicates_units_present_in_both_listings(self):
list_units = "wrolpi-custom.service loaded inactive dead Task\n"
list_unit_files = "wrolpi-custom.service static -\n"
with mock.patch("subprocess.run",
side_effect=self._side_effect(list_units, list_unit_files)):
found = discover_wrolpi_services()
assert found.count("wrolpi-custom") == 1
class TestServiceActions:
"""Tests for service action functions (start, stop, restart, enable, disable)."""
@ -275,7 +341,7 @@ class TestGetServiceLogs:
class TestDiscoverRunningWrolpiServices:
"""Tests for discover_running_wrolpi_services function."""
"""Tests for discover_wrolpi_services function."""
def test_discovers_unknown_services(self):
"""Should discover wrolpi-* services not in managed config."""
@ -286,7 +352,7 @@ class TestDiscoverRunningWrolpiServices:
)
with mock.patch("subprocess.run") as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout=systemctl_output, stderr="")
result = discover_running_wrolpi_services()
result = discover_wrolpi_services()
# wrolpi-api is managed, so only the other two should be discovered
assert "wrolpi-api" not in result
assert "wrolpi-fix-media-permissions" in result
@ -296,13 +362,13 @@ class TestDiscoverRunningWrolpiServices:
"""Should return empty list when systemctl fails."""
with mock.patch("subprocess.run") as mock_run:
mock_run.return_value = mock.Mock(returncode=1, stdout="", stderr="error")
result = discover_running_wrolpi_services()
result = discover_wrolpi_services()
assert result == []
def test_returns_empty_on_no_systemctl(self):
"""Should return empty list when systemctl is not found."""
with mock.patch("subprocess.run", side_effect=FileNotFoundError()):
result = discover_running_wrolpi_services()
result = discover_wrolpi_services()
assert result == []
def test_returns_empty_when_no_extra_services(self):
@ -310,7 +376,7 @@ class TestDiscoverRunningWrolpiServices:
systemctl_output = "wrolpi-api.service loaded active running WROLPi API\n"
with mock.patch("subprocess.run") as mock_run:
mock_run.return_value = mock.Mock(returncode=0, stdout=systemctl_output, stderr="")
result = discover_running_wrolpi_services()
result = discover_wrolpi_services()
assert result == []