mirror of
https://github.com/home-assistant/supervisor.git
synced 2026-04-02 00:07:16 +01:00
c9a2da34c2
systemd only emits bus signals (including PropertiesChanged) when at least one client has called Subscribe() on the Manager interface. On regular HAOS systems, systemd-logind calls Subscribe which enables signals for all bus clients. However, in environments without systemd-logind (such as the Supervisor devcontainer with systemd), no signals are emitted, causing the firewall unit wait to time out. Explicitly calling Subscribe() has no downsides and makes it clear that the Supervisor relies on these signals. There is no need to call Unsubscribe() as systemd automatically tracks clients and stops emitting signals when all subscribers have disconnected from the bus. Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
260 lines
7.9 KiB
Python
260 lines
7.9 KiB
Python
"""Test hostname dbus interface."""
|
|
|
|
# pylint: disable=import-error
|
|
from dbus_fast import DBusError, Variant
|
|
from dbus_fast.aio.message_bus import MessageBus
|
|
import pytest
|
|
|
|
from supervisor.dbus.const import StartUnitMode, StopUnitMode, UnitActiveState
|
|
from supervisor.dbus.systemd import Systemd
|
|
from supervisor.exceptions import DBusNotConnectedError, DBusSystemdNoSuchUnit
|
|
|
|
from tests.common import mock_dbus_services
|
|
from tests.dbus_service_mocks.systemd import Systemd as SystemdService
|
|
|
|
|
|
@pytest.fixture(name="systemd_service", autouse=True)
|
|
async def fixture_systemd_service(dbus_session_bus: MessageBus) -> SystemdService:
|
|
"""Mock systemd dbus service."""
|
|
yield (await mock_dbus_services({"systemd": None}, dbus_session_bus))["systemd"]
|
|
|
|
|
|
async def test_dbus_systemd_info(dbus_session_bus: MessageBus):
|
|
"""Test systemd properties."""
|
|
systemd = Systemd()
|
|
|
|
assert systemd.boot_timestamp is None
|
|
assert systemd.startup_time is None
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
assert systemd.boot_timestamp == 1632236713344227
|
|
assert systemd.startup_time == 45.304696
|
|
|
|
|
|
async def test_subscribe_on_connect(
|
|
systemd_service: SystemdService, dbus_session_bus: MessageBus
|
|
):
|
|
"""Test that Subscribe is called on connect to enable signal emission."""
|
|
systemd_service.Subscribe.calls.clear()
|
|
systemd = Systemd()
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
assert systemd_service.Subscribe.calls == [()]
|
|
|
|
|
|
async def test_reboot(systemd_service: SystemdService, dbus_session_bus: MessageBus):
|
|
"""Test reboot."""
|
|
systemd_service.Reboot.calls.clear()
|
|
systemd = Systemd()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await systemd.reboot()
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
assert await systemd.reboot() is None
|
|
assert systemd_service.Reboot.calls == [()]
|
|
|
|
|
|
async def test_power_off(systemd_service: SystemdService, dbus_session_bus: MessageBus):
|
|
"""Test power off."""
|
|
systemd_service.PowerOff.calls.clear()
|
|
systemd = Systemd()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await systemd.power_off()
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
assert await systemd.power_off() is None
|
|
assert systemd_service.PowerOff.calls == [()]
|
|
|
|
|
|
async def test_start_unit(
|
|
systemd_service: SystemdService, dbus_session_bus: MessageBus
|
|
):
|
|
"""Test start unit."""
|
|
systemd_service.StartUnit.calls.clear()
|
|
systemd = Systemd()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await systemd.start_unit("test_unit", StartUnitMode.REPLACE)
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
assert (
|
|
await systemd.start_unit("test_unit", StartUnitMode.REPLACE)
|
|
== "/org/freedesktop/systemd1/job/7623"
|
|
)
|
|
assert systemd_service.StartUnit.calls == [("test_unit", "replace")]
|
|
|
|
|
|
async def test_stop_unit(systemd_service: SystemdService, dbus_session_bus: MessageBus):
|
|
"""Test stop unit."""
|
|
systemd_service.StopUnit.calls.clear()
|
|
systemd = Systemd()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await systemd.stop_unit("test_unit", StopUnitMode.REPLACE)
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
assert (
|
|
await systemd.stop_unit("test_unit", StopUnitMode.REPLACE)
|
|
== "/org/freedesktop/systemd1/job/7623"
|
|
)
|
|
assert systemd_service.StopUnit.calls == [("test_unit", "replace")]
|
|
|
|
|
|
async def test_restart_unit(
|
|
systemd_service: SystemdService, dbus_session_bus: MessageBus
|
|
):
|
|
"""Test restart unit."""
|
|
systemd_service.RestartUnit.calls.clear()
|
|
systemd = Systemd()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await systemd.restart_unit("test_unit", StartUnitMode.REPLACE)
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
assert (
|
|
await systemd.restart_unit("test_unit", StartUnitMode.REPLACE)
|
|
== "/org/freedesktop/systemd1/job/7623"
|
|
)
|
|
assert systemd_service.RestartUnit.calls == [("test_unit", "replace")]
|
|
|
|
|
|
async def test_reload_unit(
|
|
systemd_service: SystemdService, dbus_session_bus: MessageBus
|
|
):
|
|
"""Test reload unit."""
|
|
systemd_service.ReloadOrRestartUnit.calls.clear()
|
|
systemd = Systemd()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await systemd.reload_unit("test_unit", StartUnitMode.REPLACE)
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
assert (
|
|
await systemd.reload_unit("test_unit", StartUnitMode.REPLACE)
|
|
== "/org/freedesktop/systemd1/job/7623"
|
|
)
|
|
assert systemd_service.ReloadOrRestartUnit.calls == [("test_unit", "replace")]
|
|
|
|
|
|
async def test_list_units(dbus_session_bus: MessageBus):
|
|
"""Test list units."""
|
|
systemd = Systemd()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await systemd.list_units()
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
units = await systemd.list_units()
|
|
assert len(units) == 4
|
|
assert units[1][0] == "firewalld.service"
|
|
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")),
|
|
],
|
|
[],
|
|
)
|
|
]
|
|
|
|
|
|
async def test_reset_failed_unit(
|
|
systemd_service: SystemdService, dbus_session_bus: MessageBus
|
|
):
|
|
"""Test resetting a failed unit."""
|
|
systemd_service.ResetFailedUnit.calls.clear()
|
|
systemd = Systemd()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await systemd.reset_failed_unit("tmp-test.mount")
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
assert await systemd.reset_failed_unit("tmp-test.mount") is None
|
|
assert systemd_service.ResetFailedUnit.calls == [("tmp-test.mount",)]
|
|
|
|
|
|
async def test_get_unit(systemd_service: SystemdService, dbus_session_bus: MessageBus):
|
|
"""Test getting job ID for unit."""
|
|
await mock_dbus_services(
|
|
{"systemd_unit": "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount"},
|
|
dbus_session_bus,
|
|
)
|
|
systemd_service.GetUnit.calls.clear()
|
|
systemd = Systemd()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await systemd.get_unit("tmp-test.mount")
|
|
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
unit = await systemd.get_unit("tmp-test.mount")
|
|
assert unit.bus_name == "org.freedesktop.systemd1"
|
|
assert unit.object_path == "/org/freedesktop/systemd1/unit/tmp_2dyellow_2emount"
|
|
assert await unit.get_active_state() == UnitActiveState.ACTIVE
|
|
assert systemd_service.GetUnit.calls == [("tmp-test.mount",)]
|
|
|
|
|
|
async def test_get_unit_not_found(
|
|
systemd_service: SystemdService, dbus_session_bus: MessageBus
|
|
):
|
|
"""Test error for non-existent unit name."""
|
|
systemd_service.response_get_unit = DBusError(
|
|
"org.freedesktop.systemd1.NoSuchUnit", "error"
|
|
)
|
|
|
|
systemd = Systemd()
|
|
await systemd.connect(dbus_session_bus)
|
|
|
|
with pytest.raises(DBusSystemdNoSuchUnit):
|
|
await systemd.get_unit("error.mount")
|