update check active

This commit is contained in:
2026-07-20 16:08:36 +07:00
parent 4a159cad71
commit db1ce68800
17 changed files with 1230 additions and 73 deletions

View File

@@ -1,5 +1,8 @@
from __future__ import annotations
from datetime import datetime, timezone
from app.core.command_runner import CommandError
from app.core.command_runner import CommandRunner
@@ -31,17 +34,60 @@ class ServiceManager:
raise RuntimeError(f"Service is not active: {service_name}")
def get_service_status(self, service_name: str) -> dict[str, object]:
active = self._query(["systemctl", "is-active", service_name]) == "active"
enabled = self._query(["systemctl", "is-enabled", service_name]) == "enabled"
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,
"status": "running" if active else "stopped",
"healthy": active,
"status": status,
"checkedAt": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"),
"errorMessage": error_message,
}
def _query(self, command: list[str]) -> str:
try:
return self.command_runner.run(command).stdout.strip()
except Exception:
return "unknown"
@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