jeans/supervisor
jeans/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-07-15 13:16:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add start transient unit dbus method (#4237)

Browse Source
This commit is contained in:
Mike Degatano 2023-04-12 13:08:02 -04:00 committed by GitHub
parent f5b6feec77
commit dc4a753fe3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 105 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
supervisor/dbus/const.py
View File

@ -315,3 +315,22 @@ class ResolvConfMode(str, Enum):
STATIC = "static"
STUB = "stub"
UPLINK = "uplink"
class StopUnitMode(str, Enum):
"""Mode for stopping the unit."""
REPLACE = "replace"
FAIL = "fail"
IGNORE_DEPENDENCIES = "ignore-dependencies"
IGNORE_REQUIREMENTS = "ignore-requirements"
class StartUnitMode(str, Enum):
"""Mode for starting the unit."""
REPLACE = "replace"
FAIL = "fail"
IGNORE_DEPENDENCIES = "ignore-dependencies"
IGNORE_REQUIREMENTS = "ignore-requirements"
ISOLATE = "isolate"

28
supervisor/dbus/systemd.py
View File

@ -1,6 +1,7 @@
"""Interface to Systemd over D-Bus."""
import logging
from dbus_fast import Variant
from dbus_fast.aio.message_bus import MessageBus
from ..exceptions import DBusError, DBusInterfaceError
@ -13,6 +14,8 @@ from .const import (
DBUS_IFACE_SYSTEMD_MANAGER,
DBUS_NAME_SYSTEMD,
DBUS_OBJECT_SYSTEMD,
StartUnitMode,
StopUnitMode,
)
from .interface import DBusInterfaceProxy, dbus_property
from .utils import dbus_connected
@ -73,24 +76,24 @@ class Systemd(DBusInterfaceProxy):
await self.dbus.Manager.call_power_off()
@dbus_connected
async def start_unit(self, unit, mode) -> str:
async def start_unit(self, unit: str, mode: StartUnitMode) -> str:
"""Start a systemd service unit. Returns object path of job."""
return await self.dbus.Manager.call_start_unit(unit, mode)
return await self.dbus.Manager.call_start_unit(unit, mode.value)
@dbus_connected
async def stop_unit(self, unit, mode) -> str:
async def stop_unit(self, unit: str, mode: StopUnitMode) -> str:
"""Stop a systemd service unit. Returns object path of job."""
return await self.dbus.Manager.call_stop_unit(unit, mode)
return await self.dbus.Manager.call_stop_unit(unit, mode.value)
@dbus_connected
async def reload_unit(self, unit, mode) -> str:
async def reload_unit(self, unit: str, mode: StartUnitMode) -> str:
"""Reload a systemd service unit. Returns object path of job."""
return await self.dbus.Manager.call_reload_or_restart_unit(unit, mode)
return await self.dbus.Manager.call_reload_or_restart_unit(unit, mode.value)
@dbus_connected
async def restart_unit(self, unit, mode) -> str:
async def restart_unit(self, unit: str, mode: StartUnitMode) -> str:
"""Restart a systemd service unit. Returns object path of job."""
return await self.dbus.Manager.call_restart_unit(unit, mode)
return await self.dbus.Manager.call_restart_unit(unit, mode.value)
@dbus_connected
async def list_units(
@ -98,3 +101,12 @@ class Systemd(DBusInterfaceProxy):
) -> list[tuple[str, str, str, str, str, str, str, int, str, str]]:
"""Return a list of available systemd services."""
return await self.dbus.Manager.call_list_units()
@dbus_connected
async def start_transient_unit(
self, unit: str, mode: StartUnitMode, properties: list[tuple[str, Variant]]
) -> str:
"""Start a transient unit which is released when stopped or on reboot. Returns object path of job."""
return await self.dbus.Manager.call_start_transient_unit(
unit, mode.value, properties, []
)

9
supervisor/host/services.py
View File

@ -5,6 +5,7 @@ import logging
import attr
from ..coresys import CoreSysAttributes
from ..dbus.const import StartUnitMode, StopUnitMode
from ..exceptions import HassioError, HostNotSupportedError, HostServiceError
_LOGGER: logging.Logger = logging.getLogger(__name__)
@ -42,7 +43,7 @@ class ServiceManager(CoreSysAttributes):
self._check_dbus(unit)
_LOGGER.info("Starting local service %s", unit)
return self.sys_dbus.systemd.start_unit(unit, MOD_REPLACE)
return self.sys_dbus.systemd.start_unit(unit, StartUnitMode.REPLACE)
def stop(self, unit) -> Awaitable[str]:
"""Stop a service on host.
@ -52,7 +53,7 @@ class ServiceManager(CoreSysAttributes):
self._check_dbus(unit)
_LOGGER.info("Stopping local service %s", unit)
return self.sys_dbus.systemd.stop_unit(unit, MOD_REPLACE)
return self.sys_dbus.systemd.stop_unit(unit, StopUnitMode.REPLACE)
def reload(self, unit) -> Awaitable[str]:
"""Reload a service on host.
@ -62,7 +63,7 @@ class ServiceManager(CoreSysAttributes):
self._check_dbus(unit)
_LOGGER.info("Reloading local service %s", unit)
return self.sys_dbus.systemd.reload_unit(unit, MOD_REPLACE)
return self.sys_dbus.systemd.reload_unit(unit, StartUnitMode.REPLACE)
def restart(self, unit) -> Awaitable[str]:
"""Restart a service on host.
@ -72,7 +73,7 @@ class ServiceManager(CoreSysAttributes):
self._check_dbus(unit)
_LOGGER.info("Restarting local service %s", unit)
return self.sys_dbus.systemd.restart_unit(unit, MOD_REPLACE)
return self.sys_dbus.systemd.restart_unit(unit, StartUnitMode.REPLACE)
def exists(self, unit) -> bool:
"""Check if a unit exists and return True."""

62
tests/dbus/test_systemd.py
View File

@ -1,8 +1,10 @@
"""Test hostname dbus interface."""
# pylint: disable=import-error
from dbus_fast import Variant
from dbus_fast.aio.message_bus import MessageBus
import pytest
from supervisor.dbus.const import StartUnitMode, StopUnitMode
from supervisor.dbus.systemd import Systemd
from supervisor.exceptions import DBusNotConnectedError
@ -65,12 +67,12 @@ async def test_start_unit(
systemd = Systemd()
with pytest.raises(DBusNotConnectedError):
await systemd.start_unit("test_unit", "replace")
await systemd.start_unit("test_unit", StartUnitMode.REPLACE)
await systemd.connect(dbus_session_bus)
assert (
await systemd.start_unit("test_unit", "replace")
await systemd.start_unit("test_unit", StartUnitMode.REPLACE)
== "/org/freedesktop/systemd1/job/7623"
)
assert systemd_service.StartUnit.calls == [("test_unit", "replace")]
@ -82,12 +84,12 @@ async def test_stop_unit(systemd_service: SystemdService, dbus_session_bus: Mess
systemd = Systemd()
with pytest.raises(DBusNotConnectedError):
await systemd.stop_unit("test_unit", "replace")
await systemd.stop_unit("test_unit", StopUnitMode.REPLACE)
await systemd.connect(dbus_session_bus)
assert (
await systemd.stop_unit("test_unit", "replace")
await systemd.stop_unit("test_unit", StopUnitMode.REPLACE)
== "/org/freedesktop/systemd1/job/7623"
)
assert systemd_service.StopUnit.calls == [("test_unit", "replace")]
@ -101,12 +103,12 @@ async def test_restart_unit(
systemd = Systemd()
with pytest.raises(DBusNotConnectedError):
await systemd.restart_unit("test_unit", "replace")
await systemd.restart_unit("test_unit", StartUnitMode.REPLACE)
await systemd.connect(dbus_session_bus)
assert (
await systemd.restart_unit("test_unit", "replace")
await systemd.restart_unit("test_unit", StartUnitMode.REPLACE)
== "/org/freedesktop/systemd1/job/7623"
)
assert systemd_service.RestartUnit.calls == [("test_unit", "replace")]
@ -120,12 +122,12 @@ async def test_reload_unit(
systemd = Systemd()
with pytest.raises(DBusNotConnectedError):
await systemd.reload_unit("test_unit", "replace")
await systemd.reload_unit("test_unit", StartUnitMode.REPLACE)
await systemd.connect(dbus_session_bus)
assert (
await systemd.reload_unit("test_unit", "replace")
await systemd.reload_unit("test_unit", StartUnitMode.REPLACE)
== "/org/freedesktop/systemd1/job/7623"
)
assert systemd_service.ReloadOrRestartUnit.calls == [("test_unit", "replace")]
@ -146,3 +148,47 @@ async def test_list_units(dbus_session_bus: MessageBus):
assert units[1][2] == "not-found"
assert units[3][0] == "zram-swap.service"
assert units[3][2] == "loaded"
async def test_start_transient_unit(
systemd_service: SystemdService, dbus_session_bus: MessageBus
):
"""Test start transient unit."""
systemd_service.StartTransientUnit.calls.clear()
systemd = Systemd()
with pytest.raises(DBusNotConnectedError):
await systemd.start_transient_unit(
"tmp-test.mount",
StartUnitMode.FAIL,
[],
)
await systemd.connect(dbus_session_bus)
assert (
await systemd.start_transient_unit(
"tmp-test.mount",
StartUnitMode.FAIL,
[
("Description", Variant("s", "Test")),
("What", Variant("s", "//homeassistant/config")),
("Type", Variant("s", "cifs")),
("Options", Variant("s", "username=homeassistant,password=password")),
],
)
== "/org/freedesktop/systemd1/job/7623"
)
assert systemd_service.StartTransientUnit.calls == [
(
"tmp-test.mount",
"fail",
[
["Description", Variant("s", "Test")],
["What", Variant("s", "//homeassistant/config")],
["Type", Variant("s", "cifs")],
["Options", Variant("s", "username=homeassistant,password=password")],
],
[],
)
]

7
tests/dbus_service_mocks/systemd.py
View File

@ -664,6 +664,13 @@ class Systemd(DBusServiceMock):
"""Restart a service unit."""
return "/org/freedesktop/systemd1/job/7623"
@dbus_method()
def StartTransientUnit(
self, name: "s", mode: "s", properties: "a(sv)", aux: "a(sa(sv))"
) -> "o":
"""Start a transient service unit."""
return "/org/freedesktop/systemd1/job/7623"
@dbus_method()
def ListUnits(
self,