94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from app.core.command_runner import CommandError
|
|
from app.core.command_runner import CommandRunner
|
|
|
|
|
|
class ServiceManager:
|
|
def __init__(self, command_runner: CommandRunner) -> None:
|
|
self.command_runner = command_runner
|
|
|
|
def enable_service(self, service_name: str) -> None:
|
|
self.command_runner.run(["systemctl", "enable", service_name])
|
|
|
|
def disable_service(self, service_name: str) -> None:
|
|
self.command_runner.run(["systemctl", "disable", service_name])
|
|
|
|
def start_service(self, service_name: str) -> None:
|
|
self.command_runner.run(["systemctl", "start", service_name])
|
|
|
|
def stop_service(self, service_name: str) -> None:
|
|
self.command_runner.run(["systemctl", "stop", service_name])
|
|
|
|
def restart_service(self, service_name: str) -> None:
|
|
self.command_runner.run(["systemctl", "restart", service_name])
|
|
|
|
def reset_failed(self, service_name: str) -> None:
|
|
self.command_runner.run(["systemctl", "reset-failed", service_name])
|
|
|
|
def assert_service_active(self, service_name: str) -> None:
|
|
result = self.command_runner.run(["systemctl", "is-active", service_name])
|
|
if result.stdout.strip() != "active":
|
|
raise RuntimeError(f"Service is not active: {service_name}")
|
|
|
|
def get_service_status(self, service_name: str) -> dict[str, object]:
|
|
command = [
|
|
"systemctl",
|
|
"show",
|
|
service_name,
|
|
"--no-pager",
|
|
"--property=LoadState",
|
|
"--property=ActiveState",
|
|
"--property=SubState",
|
|
"--property=UnitFileState",
|
|
]
|
|
error_message: str | None = None
|
|
try:
|
|
output = self.command_runner.run(command, timeout=5).stdout
|
|
except CommandError as error:
|
|
output = error.stdout
|
|
error_message = str(error)
|
|
except Exception as error:
|
|
output = ""
|
|
error_message = str(error)
|
|
|
|
properties = self._parse_properties(output)
|
|
load_state = properties.get("LoadState", "unknown") or "unknown"
|
|
active_state = properties.get("ActiveState", "unknown") or "unknown"
|
|
sub_state = properties.get("SubState", "unknown") or "unknown"
|
|
unit_file_state = properties.get("UnitFileState", "unknown") or "unknown"
|
|
active = load_state == "loaded" and active_state == "active"
|
|
enabled = unit_file_state in {"enabled", "enabled-runtime"}
|
|
|
|
if load_state == "not-found":
|
|
status = "not-found"
|
|
elif active_state != "unknown":
|
|
status = active_state
|
|
else:
|
|
status = "unknown"
|
|
|
|
return {
|
|
"serviceName": service_name,
|
|
"loadState": load_state,
|
|
"activeState": active_state,
|
|
"subState": sub_state,
|
|
"unitFileState": unit_file_state,
|
|
"active": active,
|
|
"enabled": enabled,
|
|
"healthy": active,
|
|
"status": status,
|
|
"checkedAt": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"),
|
|
"errorMessage": error_message,
|
|
}
|
|
|
|
@staticmethod
|
|
def _parse_properties(output: str) -> dict[str, str]:
|
|
properties: dict[str, str] = {}
|
|
for line in output.splitlines():
|
|
key, separator, value = line.partition("=")
|
|
if separator and key:
|
|
properties[key.strip()] = value.strip()
|
|
return properties
|