66046b6a9d
Previously, the check only considered 'active' as a valid ActiveState. This update ensures the system also accounts for units in the 'activating' state, improving compatibility with transitional states in systemd services. Signed-off-by: Sven Sager <s.sager@kunbus.com>
122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
# SPDX-FileCopyrightText: 2025 KUNBUS GmbH
|
|
# SPDX-License-Identifier: GPL-2.0-or-later
|
|
from enum import Enum
|
|
from logging import getLogger
|
|
from threading import Thread
|
|
from typing import Optional
|
|
|
|
from pydbus import SystemBus
|
|
|
|
log = getLogger(__name__)
|
|
|
|
|
|
class ServiceActions(Enum):
|
|
"""
|
|
Enumeration class for defining configuration actions.
|
|
|
|
This enumeration provides predefined constants for common configuration
|
|
actions. It can be used to ensure consistency when working with or defining
|
|
such actions in a system.
|
|
"""
|
|
|
|
ENABLE = "enable"
|
|
DISABLE = "disable"
|
|
STATUS = "status"
|
|
AVAILABLE = "available"
|
|
|
|
|
|
def simple_systemd(action: ServiceActions, unit: str, unmask: bool = False) -> Optional[bool]:
|
|
"""
|
|
Perform systemd service actions such as enable, disable, check status, or availability.
|
|
|
|
This function interacts with the systemd D-Bus API to manage and query the
|
|
state of services on a system. The supported actions include enabling a systemd
|
|
unit, disabling it, starting/stopping a unit, and checking its status or
|
|
availability. The function supports asynchronous configuration changes through
|
|
threads where applicable.
|
|
|
|
Parameters:
|
|
action (ServiceActions): The action to perform on the systemd service.
|
|
Supported actions are ENABLE, DISABLE, STATUS, and AVAILABLE.
|
|
unit (str): The name of the systemd unit to operate on (e.g., "example.service").
|
|
unmask (bool): When enabling a unit, if True, any masked unit file will
|
|
first be unmasked before proceeding with the operation. Defaults to False.
|
|
|
|
Returns:
|
|
Optional[bool]: The return value depends on the action. For STATUS or
|
|
AVAILABLE actions, it returns True if the unit satisfies the condition
|
|
(e.g., enabled and active, or available and loaded), False otherwise.
|
|
For other actions, it returns None.
|
|
"""
|
|
bus = SystemBus()
|
|
systemd = bus.get(
|
|
"org.freedesktop.systemd1",
|
|
"/org/freedesktop/systemd1",
|
|
)
|
|
systemd_manager = systemd["org.freedesktop.systemd1.Manager"]
|
|
|
|
if action is ServiceActions.ENABLE:
|
|
|
|
def thread_unit_config():
|
|
"""Change configuration asynchronously."""
|
|
lst_change_unmask = []
|
|
if unmask:
|
|
# Dbus call: UnmaskUnitFiles(in as files, in b runtime, out a(sss) changes
|
|
lst_change_unmask = systemd_manager.UnmaskUnitFiles([unit], False)
|
|
|
|
# Dbus call: EnableUnitFiles(in as files, in b runtime, in b force,
|
|
# out b carries_install_info, out a(sss) changes
|
|
lst_change_enable = systemd_manager.EnableUnitFiles([unit], False, False)
|
|
if lst_change_unmask or lst_change_enable:
|
|
# Reload systemd after modified unit property
|
|
systemd_manager.Reload()
|
|
|
|
Thread(target=thread_unit_config, daemon=True).start()
|
|
|
|
# Dbus call: StartUnit(in s name, in s mode, out o job
|
|
systemd_manager.StartUnit(unit, "replace")
|
|
|
|
elif action is ServiceActions.DISABLE:
|
|
|
|
def thread_unit_config():
|
|
"""Change configuration asynchronously."""
|
|
# Dbus call: DisableUnitFiles (in as files, in b runtime, out a(sss) changes)
|
|
change = systemd_manager.DisableUnitFiles([unit], False)
|
|
if change:
|
|
# Reload systemd after modified unit property
|
|
systemd_manager.Reload()
|
|
|
|
Thread(target=thread_unit_config, daemon=True).start()
|
|
|
|
# Dbus call: StopUnit(in s name,in s mode, out o job
|
|
systemd_manager.StopUnit(unit, "replace")
|
|
|
|
elif action is ServiceActions.STATUS:
|
|
try:
|
|
unit_path = systemd_manager.LoadUnit(unit)
|
|
properties = bus.get("org.freedesktop.systemd1", unit_path)
|
|
except Exception:
|
|
log.warning(f"could not get systemd unit {unit}")
|
|
return False
|
|
|
|
return properties.UnitFileState == "enabled" and properties.ActiveState in (
|
|
"active",
|
|
"activating",
|
|
)
|
|
|
|
elif action is ServiceActions.AVAILABLE:
|
|
try:
|
|
unit_path = systemd_manager.LoadUnit(unit)
|
|
properties = bus.get("org.freedesktop.systemd1", unit_path)
|
|
except Exception:
|
|
log.warning(f"could not get systemd unit {unit}")
|
|
return False
|
|
|
|
return properties.LoadState != "not-found"
|
|
|
|
else:
|
|
raise ValueError(f"action {action} not supported")
|
|
|
|
return None
|