From e8bc2b32c4a0b0cb2b5325b75cbd8b2830ea8570 Mon Sep 17 00:00:00 2001 From: Roland Date: Thu, 5 Mar 2026 18:11:25 -0700 Subject: [PATCH] Displaying any wrolpi-* service in the Controller, now that we have these scripts. --- controller/lib/systemd.py | 164 ++++++++++++++++++++---- controller/test/test_systemd.py | 215 +++++++++++++++++++++++++++----- 2 files changed, 326 insertions(+), 53 deletions(-) diff --git a/controller/lib/systemd.py b/controller/lib/systemd.py index 37ee2f09..4b55fe65 100644 --- a/controller/lib/systemd.py +++ b/controller/lib/systemd.py @@ -79,6 +79,9 @@ def get_service_status(name: str) -> dict: """ service_config = get_service_config(name) if not service_config: + # Allow dynamically discovered wrolpi-* services. + if name.startswith("wrolpi-"): + return get_discovered_service_status(name) return {"error": f"Unknown service: {name}"} systemd_name = service_config.get("systemd_name", name) @@ -142,9 +145,115 @@ def get_service_status(name: str) -> dict: } +def discover_running_wrolpi_services() -> list[str]: + """ + Discover running wrolpi-* systemd services not in the managed services config. + + Returns: + list of systemd unit names (e.g., ["wrolpi-fix-media-permissions.service"]) + """ + 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 [] + + discovered = [] + for line in result.stdout.strip().splitlines(): + parts = line.split() + if not parts: + 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) + + return discovered + + +def get_discovered_service_status(name: str) -> dict: + """ + Get status of a dynamically discovered wrolpi-* service (not in managed config). + + Args: + name: Service name (e.g., "wrolpi-fix-media-permissions") + + Returns: + dict with status fields + """ + try: + active_result = subprocess.run( + ["systemctl", "is-active", name], + capture_output=True, + text=True, + timeout=5, + ) + is_active = active_result.stdout.strip() + + enabled_result = subprocess.run( + ["systemctl", "is-enabled", name], + capture_output=True, + text=True, + timeout=5, + ) + is_enabled = enabled_result.stdout.strip() + except (subprocess.TimeoutExpired, FileNotFoundError): + return { + "name": name, + "systemd_name": name, + "status": "unknown", + "active": "unknown", + "enabled": False, + "port": None, + "viewable": False, + "view_path": "", + "use_https": False, + "description": "", + "error": "systemctl not available", + } + + if is_active == "active": + status = "running" + elif is_active == "activating": + status = "running" + elif is_active == "inactive": + status = "stopped" + elif is_active == "failed": + status = "failed" + else: + status = "unknown" + + return { + "name": name, + "systemd_name": name, + "status": status, + "active": is_active, + "enabled": is_enabled == "enabled", + "port": None, + "viewable": False, + "view_path": "", + "use_https": False, + "description": "", + } + + def get_all_services_status() -> list[dict]: """ - Get status of all managed services. + 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. @@ -158,9 +267,26 @@ def get_all_services_status() -> list[dict]: if service.get("show_only_when_running") and status.get("status") != "running": 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) + return results +def _get_systemd_name(name: str) -> Optional[str]: + """Get the systemd unit name for a service, supporting both managed and discovered wrolpi-* services.""" + service_config = get_service_config(name) + if service_config: + return service_config.get("systemd_name", name) + if name.startswith("wrolpi-"): + return name + return None + + def start_service(name: str) -> dict: """ Start a managed service. @@ -171,11 +297,9 @@ def start_service(name: str) -> dict: Returns: dict with success status """ - service_config = get_service_config(name) - if not service_config: + systemd_name = _get_systemd_name(name) + if not systemd_name: return {"success": False, "error": f"Unknown service: {name}"} - - systemd_name = service_config.get("systemd_name", name) result = _run_systemctl("start", systemd_name) return { @@ -196,11 +320,9 @@ def stop_service(name: str) -> dict: Returns: dict with success status """ - service_config = get_service_config(name) - if not service_config: + systemd_name = _get_systemd_name(name) + if not systemd_name: return {"success": False, "error": f"Unknown service: {name}"} - - systemd_name = service_config.get("systemd_name", name) result = _run_systemctl("stop", systemd_name) return { @@ -221,12 +343,10 @@ def restart_service(name: str) -> dict: Returns: dict with success status """ - service_config = get_service_config(name) - if not service_config: + systemd_name = _get_systemd_name(name) + if not systemd_name: return {"success": False, "error": f"Unknown service: {name}"} - systemd_name = service_config.get("systemd_name", name) - # Special handling for self-restart: use Popen to avoid blocking # (the process will be killed before subprocess.run can return) if systemd_name == "wrolpi-controller": @@ -266,11 +386,9 @@ def enable_service(name: str) -> dict: Returns: dict with success status """ - service_config = get_service_config(name) - if not service_config: + systemd_name = _get_systemd_name(name) + if not systemd_name: return {"success": False, "error": f"Unknown service: {name}"} - - systemd_name = service_config.get("systemd_name", name) result = _run_systemctl("enable", systemd_name) return { @@ -291,11 +409,9 @@ def disable_service(name: str) -> dict: Returns: dict with success status """ - service_config = get_service_config(name) - if not service_config: + systemd_name = _get_systemd_name(name) + if not systemd_name: return {"success": False, "error": f"Unknown service: {name}"} - - systemd_name = service_config.get("systemd_name", name) result = _run_systemctl("disable", systemd_name) return { @@ -318,12 +434,10 @@ def get_service_logs(name: str, lines: int = 100, since: Optional[str] = None) - Returns: dict with logs and metadata """ - service_config = get_service_config(name) - if not service_config: + systemd_name = _get_systemd_name(name) + if not systemd_name: return {"error": f"Unknown service: {name}"} - systemd_name = service_config.get("systemd_name", name) - cmd = ["journalctl", "-u", systemd_name, "-n", str(lines), "--no-pager"] if since: cmd.extend(["--since", since]) diff --git a/controller/test/test_systemd.py b/controller/test/test_systemd.py index 95eaf959..d5c47ca9 100644 --- a/controller/test/test_systemd.py +++ b/controller/test/test_systemd.py @@ -11,6 +11,9 @@ from controller.lib.systemd import ( get_service_config, get_service_status, get_all_services_status, + discover_running_wrolpi_services, + get_discovered_service_status, + _get_systemd_name, start_service, stop_service, restart_service, @@ -147,43 +150,47 @@ class TestGetAllServicesStatus: def test_returns_list(self): """Should return a list.""" - with mock.patch("subprocess.run") as mock_run: - mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="") - result = get_all_services_status() - assert isinstance(result, list) + with mock.patch("controller.lib.systemd.discover_running_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() + assert isinstance(result, list) def test_returns_status_for_each_visible_service(self): """Should return status for each visible managed service.""" - with mock.patch("subprocess.run") as mock_run: - mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="") - result = get_all_services_status() - managed = get_managed_services() - # All services should be returned when they're all "running" (active) - assert len(result) == len(managed) + with mock.patch("controller.lib.systemd.discover_running_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() + managed = get_managed_services() + # 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("subprocess.run") as mock_run: - # Simulate all services as "inactive" (stopped) - mock_run.return_value = mock.Mock(returncode=0, stdout="inactive", stderr="") - 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) + 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="") + 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("subprocess.run") as mock_run: - # Simulate all services as "active" (running) - mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="") - result = get_all_services_status() - managed = get_managed_services() - # All services should be visible when running - assert len(result) == len(managed) - # Verify wrolpi-upgrade is in the result - service_names = [s["name"] for s in result] - assert "wrolpi-upgrade" in service_names + with mock.patch("controller.lib.systemd.discover_running_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="") + result = get_all_services_status() + managed = get_managed_services() + # All services should be visible when running + assert len(result) == len(managed) + # Verify wrolpi-upgrade is in the result + service_names = [s["name"] for s in result] + assert "wrolpi-upgrade" in service_names class TestServiceActions: @@ -266,3 +273,155 @@ class TestGetServiceLogs: call_args = mock_run.call_args[0][0] assert "--since" in call_args assert "1h" in call_args + + +class TestDiscoverRunningWrolpiServices: + """Tests for discover_running_wrolpi_services function.""" + + def test_discovers_unknown_services(self): + """Should discover wrolpi-* services not in managed config.""" + systemctl_output = ( + "wrolpi-api.service loaded active running WROLPi API\n" + "wrolpi-fix-media-permissions.service loaded active running Fix permissions\n" + "wrolpi-repair.service loaded active running WROLPi Repair\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() + # 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 + assert "wrolpi-repair" in result + + def test_returns_empty_on_failure(self): + """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() + 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() + assert result == [] + + def test_returns_empty_when_no_extra_services(self): + """Should return empty list when all wrolpi-* services are managed.""" + 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() + assert result == [] + + +class TestGetDiscoveredServiceStatus: + """Tests for get_discovered_service_status function.""" + + def test_returns_running_status(self): + """Should return status for a discovered service.""" + with mock.patch("subprocess.run") as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="") + result = get_discovered_service_status("wrolpi-fix-media-permissions") + assert result["name"] == "wrolpi-fix-media-permissions" + assert result["status"] == "running" + assert result["port"] is None + assert result["viewable"] is False + + def test_handles_no_systemctl(self): + """Should handle missing systemctl.""" + with mock.patch("subprocess.run", side_effect=FileNotFoundError()): + result = get_discovered_service_status("wrolpi-fix-media-permissions") + assert result["status"] == "unknown" + assert "error" in result + + +class TestGetSystemdName: + """Tests for _get_systemd_name helper.""" + + def test_returns_systemd_name_for_managed_service(self): + """Should return systemd_name from config for managed services.""" + result = _get_systemd_name("wrolpi-api") + assert result == "wrolpi-api" + + def test_returns_name_for_discovered_wrolpi_service(self): + """Should return the name itself for discovered wrolpi-* services.""" + result = _get_systemd_name("wrolpi-fix-media-permissions") + assert result == "wrolpi-fix-media-permissions" + + def test_returns_none_for_unknown_non_wrolpi_service(self): + """Should return None for unknown non-wrolpi services.""" + result = _get_systemd_name("some-random-service") + assert result is None + + +class TestGetAllServicesStatusWithDiscovery: + """Tests for get_all_services_status including dynamic discovery.""" + + def test_includes_discovered_running_services(self): + """Should include dynamically discovered running wrolpi-* services.""" + systemctl_list_output = ( + "wrolpi-fix-media-permissions.service loaded active running Fix permissions\n" + ) + + def mock_run_side_effect(cmd, **kwargs): + if cmd[0] == "systemctl" and cmd[1] == "list-units": + return mock.Mock(returncode=0, stdout=systemctl_list_output, stderr="") + # For is-active / is-enabled calls + return mock.Mock(returncode=0, stdout="active", stderr="") + + with mock.patch("subprocess.run", side_effect=mock_run_side_effect): + result = get_all_services_status() + service_names = [s["name"] for s in result] + assert "wrolpi-fix-media-permissions" in service_names + + def test_does_not_duplicate_managed_services(self): + """Discovered services should not duplicate managed services.""" + systemctl_list_output = ( + "wrolpi-api.service loaded active running WROLPi API\n" + ) + + def mock_run_side_effect(cmd, **kwargs): + if cmd[0] == "systemctl" and cmd[1] == "list-units": + return mock.Mock(returncode=0, stdout=systemctl_list_output, stderr="") + return mock.Mock(returncode=0, stdout="active", stderr="") + + with mock.patch("subprocess.run", side_effect=mock_run_side_effect): + result = get_all_services_status() + api_count = sum(1 for s in result if s["name"] == "wrolpi-api") + assert api_count == 1 + + +class TestActionsOnDiscoveredServices: + """Tests that service actions work on discovered wrolpi-* services.""" + + @pytest.mark.parametrize("func,expected_action", [ + (start_service, "start"), + (stop_service, "stop"), + (restart_service, "restart"), + (enable_service, "enable"), + (disable_service, "disable"), + ]) + def test_actions_work_on_discovered_services(self, func, expected_action): + """Should allow actions on discovered wrolpi-* services.""" + with mock.patch("controller.lib.systemd._run_systemctl") as mock_ctl: + mock_ctl.return_value = {"success": True, "output": "", "error": None} + result = func("wrolpi-fix-media-permissions") + assert result["success"] is True + assert result["action"] == expected_action + + def test_logs_work_on_discovered_services(self): + """Should return logs for discovered wrolpi-* services.""" + with mock.patch("subprocess.run") as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout="log output", stderr="") + result = get_service_logs("wrolpi-fix-media-permissions", lines=10) + assert result["service"] == "wrolpi-fix-media-permissions" + assert "logs" in result + + def test_get_service_status_for_discovered_service(self): + """get_service_status should work for discovered wrolpi-* services.""" + with mock.patch("subprocess.run") as mock_run: + mock_run.return_value = mock.Mock(returncode=0, stdout="active", stderr="") + result = get_service_status("wrolpi-fix-media-permissions") + assert result["name"] == "wrolpi-fix-media-permissions" + assert result["status"] == "running"