mirror of
https://github.com/home-assistant/supervisor.git
synced 2026-04-17 15:23:21 +01:00
bb4525bf3f
Handle external shutdown events (ACPI power button, hypervisor shutdown) by listening to logind's PrepareForShutdown signal. A delay inhibitor lock is acquired on startup so Supervisor has time to gracefully stop all managed services before the host proceeds with shutdown. Changes: - Add inhibit() and prepare_for_shutdown() methods to Logind D-Bus interface - Enable Unix FD negotiation on the D-Bus message bus for inhibitor lock FDs - Add background monitor task in HostManager that listens for the signal - Track the monitor task and cancel it cleanly via new unload() method - Wire host unload into Core.stop() stage 2 for clean shutdown - Add PrepareForShutdown signal constant to dbus/const.py Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
"""Test login dbus interface."""
|
|
|
|
# pylint: disable=import-error
|
|
from dbus_fast.aio.message_bus import MessageBus
|
|
import pytest
|
|
|
|
from supervisor.dbus.logind import Logind
|
|
from supervisor.exceptions import DBusNotConnectedError
|
|
|
|
from tests.common import mock_dbus_services
|
|
from tests.dbus_service_mocks.logind import Logind as LogindService
|
|
|
|
|
|
@pytest.fixture(name="logind_service")
|
|
async def fixture_logind_service(dbus_session_bus: MessageBus) -> LogindService:
|
|
"""Mock logind dbus service."""
|
|
yield (await mock_dbus_services({"logind": None}, dbus_session_bus))["logind"]
|
|
|
|
|
|
async def test_reboot(logind_service: LogindService, dbus_session_bus: MessageBus):
|
|
"""Test reboot."""
|
|
logind_service.Reboot.calls.clear()
|
|
logind = Logind()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await logind.reboot()
|
|
|
|
await logind.connect(dbus_session_bus)
|
|
|
|
assert await logind.reboot() is None
|
|
assert logind_service.Reboot.calls == [(False,)]
|
|
|
|
|
|
async def test_power_off(logind_service: LogindService, dbus_session_bus: MessageBus):
|
|
"""Test power off."""
|
|
logind_service.PowerOff.calls.clear()
|
|
logind = Logind()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await logind.power_off()
|
|
|
|
await logind.connect(dbus_session_bus)
|
|
|
|
assert await logind.power_off() is None
|
|
assert logind_service.PowerOff.calls == [(False,)]
|
|
|
|
|
|
async def test_inhibit(logind_service: LogindService, dbus_session_bus: MessageBus):
|
|
"""Test taking an inhibitor lock."""
|
|
logind_service.Inhibit.calls.clear()
|
|
logind = Logind()
|
|
|
|
with pytest.raises(DBusNotConnectedError):
|
|
await logind.inhibit("shutdown", "test", "testing", "delay")
|
|
|
|
await logind.connect(dbus_session_bus)
|
|
|
|
await logind.inhibit("shutdown", "Test", "Testing inhibit", "delay")
|
|
assert logind_service.Inhibit.calls == [
|
|
("shutdown", "Test", "Testing inhibit", "delay")
|
|
]
|
|
|
|
|
|
async def test_prepare_for_shutdown_signal(
|
|
logind_service: LogindService, dbus_session_bus: MessageBus
|
|
):
|
|
"""Test PrepareForShutdown signal."""
|
|
logind = Logind()
|
|
await logind.connect(dbus_session_bus)
|
|
|
|
async with logind.prepare_for_shutdown() as signal:
|
|
logind_service.PrepareForShutdown()
|
|
await logind_service.ping()
|
|
|
|
msg = await signal.wait_for_signal()
|
|
assert msg == [True]
|
|
|
|
|
|
async def test_dbus_logind_connect_error(
|
|
dbus_session_bus: MessageBus, caplog: pytest.LogCaptureFixture
|
|
):
|
|
"""Test connecting to logind error."""
|
|
logind = Logind()
|
|
await logind.connect(dbus_session_bus)
|
|
assert "No systemd-logind support on the host" in caplog.text
|