supervisor/tests/dbus/test_login.py at 1615b50e6b4cd45a4e6b3235b95a924895efd48e · home-assistant/supervisor · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
"""Test login dbus interface."""

# pylint: disable=import-error
import os

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)

    fd = await logind.inhibit("shutdown", "Test", "Testing inhibit", "delay")
    assert logind_service.Inhibit.calls == [
        ("shutdown", "Test", "Testing inhibit", "delay")
    ]
    if fd is not None:
        os.close(fd)


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