From daeec266cc357a65c3a51c6bdf075f0c08e01f07 Mon Sep 17 00:00:00 2001 From: Mike Degatano Date: Wed, 8 Mar 2023 05:10:24 -0500 Subject: [PATCH] Mock services on session dbus in tests (#4160) * Mock services on session dbus in tests * methods emit prop changes for testing --- .github/workflows/ci.yaml | 4 +- setup.cfg | 2 + supervisor/resolution/validate.py | 4 +- tests/common.py | 36 ++ tests/conftest.py | 38 ++ tests/dbus/test_hostname.py | 70 +-- tests/dbus/test_login.py | 39 +- tests/dbus/test_rauc.py | 86 ++-- tests/dbus/test_resolved.py | 122 ++--- tests/dbus/test_systemd.py | 153 +++--- tests/dbus/test_timedate.py | 83 ++-- tests/dbus_service_mocks/__init__.py | 6 + tests/dbus_service_mocks/base.py | 70 +++ tests/dbus_service_mocks/hostname.py | 158 ++++++ tests/dbus_service_mocks/logind.py | 31 ++ tests/dbus_service_mocks/rauc.py | 211 +++++++++ tests/dbus_service_mocks/resolved.py | 138 ++++++ tests/dbus_service_mocks/systemd.py | 685 +++++++++++++++++++++++++++ tests/dbus_service_mocks/timedate.py | 95 ++++ 19 files changed, 1766 insertions(+), 265 deletions(-) create mode 100644 tests/dbus_service_mocks/__init__.py create mode 100644 tests/dbus_service_mocks/base.py create mode 100644 tests/dbus_service_mocks/hostname.py create mode 100644 tests/dbus_service_mocks/logind.py create mode 100644 tests/dbus_service_mocks/rauc.py create mode 100644 tests/dbus_service_mocks/resolved.py create mode 100644 tests/dbus_service_mocks/systemd.py create mode 100644 tests/dbus_service_mocks/timedate.py diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 3766de636..c0693ec1c 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -13,7 +13,7 @@ env: DEFAULT_CAS: v1.0.2 concurrency: - group: '${{ github.workflow }}-${{ github.ref }}' + group: "${{ github.workflow }}-${{ github.ref }}" cancel-in-progress: true jobs: @@ -370,7 +370,7 @@ jobs: - name: Install additional system dependencies run: | sudo apt-get update - sudo apt-get install -y --no-install-recommends libpulse0 libudev1 + sudo apt-get install -y --no-install-recommends libpulse0 libudev1 dbus dbus-x11 - name: Register Python problem matcher run: | echo "::add-matcher::.github/workflows/matchers/python.json" diff --git a/setup.cfg b/setup.cfg index d38d3685c..27e468bdc 100644 --- a/setup.cfg +++ b/setup.cfg @@ -27,3 +27,5 @@ ignore = E203, D202, W504 +per-file-ignores = + tests/dbus_service_mocks/*.py: F821,F722 diff --git a/supervisor/resolution/validate.py b/supervisor/resolution/validate.py index 5c0de5a33..89b4bb075 100644 --- a/supervisor/resolution/validate.py +++ b/supervisor/resolution/validate.py @@ -6,9 +6,9 @@ import voluptuous as vol from ..const import ATTR_CHECKS, ATTR_ENABLED -def get_valid_modules(folder) -> list[str]: +def get_valid_modules(folder, *, base=__file__) -> list[str]: """Validate check name.""" - module_files = Path(__file__).parent.joinpath(folder) + module_files = Path(base).parent.joinpath(folder) if not module_files.exists(): raise vol.Invalid(f"Module folder '{folder}' not found!") diff --git a/tests/common.py b/tests/common.py index f7bd58641..fede2ebb7 100644 --- a/tests/common.py +++ b/tests/common.py @@ -1,15 +1,20 @@ """Common test functions.""" import asyncio +from importlib import import_module import json from pathlib import Path from typing import Any +from dbus_fast.aio.message_bus import MessageBus from dbus_fast.introspection import Method, Property, Signal from supervisor.dbus.interface import DBusInterface, DBusInterfaceProxy +from supervisor.resolution.validate import get_valid_modules from supervisor.utils.dbus import DBUS_INTERFACE_PROPERTIES from supervisor.utils.yaml import read_yaml_file +from .dbus_service_mocks.base import DBusServiceMock + def get_dbus_name(intr_list: list[Method | Property | Signal], snake_case: str) -> str: """Find name in introspection list, fallback to ignore case match.""" @@ -99,3 +104,34 @@ def exists_fixture(filename: str) -> bool: """Check if a fixture exists.""" path = get_fixture_path(filename) return path.exists() + + +async def mock_dbus_services( + to_mock: dict[str, list[str] | str | None], bus: MessageBus +) -> dict[str, list[DBusServiceMock] | DBusServiceMock]: + """Mock specified dbus services on bus. + + to_mock is dictionary where the key is a dbus service to mock (module must exist + in dbus_service_mocks). Value is the object path for the mocked service. Can also + be a list of object paths or None (if the mocked service defines the object path). + """ + services: dict[str, list[DBusServiceMock] | DBusServiceMock] = {} + requested_names: set[str] = set() + + for module in get_valid_modules("dbus_service_mocks", base=__file__): + if module in to_mock: + service_module = import_module(f"{__package__}.dbus_service_mocks.{module}") + + if service_module.BUS_NAME not in requested_names: + await bus.request_name(service_module.BUS_NAME) + requested_names.add(service_module.BUS_NAME) + + if isinstance(to_mock[module], list): + services[module] = [ + service_module.setup(obj_path).export(bus) + for obj_path in to_mock[module] + ] + else: + services[module] = service_module.setup(to_mock[module]).export(bus) + + return services diff --git a/tests/conftest.py b/tests/conftest.py index 17a5369b9..565c04413 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,8 +1,10 @@ """Common test functions.""" from functools import partial from inspect import unwrap +import os from pathlib import Path import re +import subprocess from typing import Any from unittest.mock import AsyncMock, MagicMock, Mock, PropertyMock, patch from uuid import uuid4 @@ -10,6 +12,7 @@ from uuid import uuid4 from aiohttp import web from aiohttp.test_utils import TestClient from awesomeversion import AwesomeVersion +from dbus_fast import BusType from dbus_fast.aio.message_bus import MessageBus from dbus_fast.aio.proxy_object import ProxyInterface, ProxyObject import pytest @@ -65,8 +68,10 @@ from .common import ( load_binary_fixture, load_fixture, load_json_fixture, + mock_dbus_services, ) from .const import TEST_ADDON_SLUG +from .dbus_service_mocks.base import DBusServiceMock # pylint: disable=redefined-outer-name, protected-access @@ -120,6 +125,39 @@ async def dbus_bus() -> MessageBus: yield bus +@pytest.fixture(scope="session") +def dbus_session() -> None: + """Start a dbus session.""" + dbus_launch = subprocess.run(["dbus-launch"], stdout=subprocess.PIPE, check=False) + envs = dbus_launch.stdout.decode(encoding="utf-8").rstrip() + + for env in envs.split("\n"): + name, value = env.split("=", 1) + os.environ[name] = value + + +@pytest.fixture +async def dbus_session_bus(dbus_session) -> MessageBus: + """Return message bus connected to session dbus.""" + bus = await MessageBus(bus_type=BusType.SESSION).connect() + yield bus + bus.disconnect() + + +@pytest.fixture +async def dbus_services( + request: pytest.FixtureRequest, dbus_session: MessageBus +) -> dict[str, DBusServiceMock | list[DBusServiceMock]]: + """Mock specified dbus services on session bus. + + Should be used indirectly. Provide a dictionary where the key a dbus service to mock + (module must exist in dbus_service_mocks). Value is the object path for the mocked service. + Can also be a list of object paths or None (if the mocked service defines the object path). + """ + with patch("supervisor.dbus.manager.MessageBus.connect", return_value=dbus_session): + yield await mock_dbus_services(request.param, dbus_session) + + def _process_pseudo_variant(data: dict[str, Any]) -> Any: """Process pseudo variant into value.""" if data["_type"] == "ay": diff --git a/tests/dbus/test_hostname.py b/tests/dbus/test_hostname.py index 4b030f48f..7a5ef6fc7 100644 --- a/tests/dbus/test_hostname.py +++ b/tests/dbus/test_hostname.py @@ -1,48 +1,62 @@ """Test hostname dbus interface.""" - -import asyncio - +# pylint: disable=import-error +from dbus_fast.aio.message_bus import MessageBus import pytest -from supervisor.coresys import CoreSys +from supervisor.dbus.hostname import Hostname from supervisor.exceptions import DBusNotConnectedError -from tests.common import fire_property_change_signal +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.hostname import Hostname as HostnameService -async def test_dbus_hostname_info(coresys: CoreSys): - """Test coresys dbus connection.""" - assert coresys.dbus.hostname.hostname is None +@pytest.fixture(name="hostname_service", autouse=True) +async def fixture_hostname_service(dbus_session_bus: MessageBus) -> HostnameService: + """Mock hostname dbus service.""" + yield (await mock_dbus_services({"hostname": None}, dbus_session_bus))["hostname"] - await coresys.dbus.hostname.connect(coresys.dbus.bus) - await coresys.dbus.hostname.update() - assert coresys.dbus.hostname.hostname == "homeassistant-n2" - assert coresys.dbus.hostname.kernel == "5.10.33" +async def test_dbus_hostname_info( + hostname_service: HostnameService, dbus_session_bus: MessageBus +): + """Test hostname properties.""" + hostname = Hostname() + + assert hostname.hostname is None + + await hostname.connect(dbus_session_bus) + + assert hostname.hostname == "homeassistant-n2" + assert hostname.kernel == "5.10.33" assert ( - coresys.dbus.hostname.cpe + hostname.cpe == "cpe:2.3:o:home-assistant:haos:6.0.dev20210504:*:development:*:*:*:odroid-n2:*" ) - assert coresys.dbus.hostname.operating_system == "Home Assistant OS 6.0.dev20210504" + assert hostname.operating_system == "Home Assistant OS 6.0.dev20210504" - fire_property_change_signal(coresys.dbus.hostname, {"StaticHostname": "test"}) - await asyncio.sleep(0) - assert coresys.dbus.hostname.hostname == "test" + hostname_service.emit_properties_changed({"StaticHostname": "test"}) + await hostname_service.ping() + assert hostname.hostname == "test" - fire_property_change_signal(coresys.dbus.hostname, {}, ["StaticHostname"]) - await asyncio.sleep(0) - assert coresys.dbus.hostname.hostname == "homeassistant-n2" + hostname_service.emit_properties_changed({}, ["StaticHostname"]) + await hostname_service.ping() + await hostname_service.ping() # To process the follow-up get all properties call + assert hostname.hostname == "homeassistant-n2" -async def test_dbus_sethostname(coresys: CoreSys, dbus: list[str]): +async def test_dbus_sethostname( + hostname_service: HostnameService, dbus_session_bus: MessageBus +): """Set hostname on backend.""" + hostname = Hostname() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.hostname.set_static_hostname("StarWars") + await hostname.set_static_hostname("StarWars") - await coresys.dbus.hostname.connect(coresys.dbus.bus) + await hostname.connect(dbus_session_bus) - dbus.clear() - await coresys.dbus.hostname.set_static_hostname("StarWars") - assert dbus == [ - "/org/freedesktop/hostname1-org.freedesktop.hostname1.SetStaticHostname" - ] + assert hostname.hostname == "homeassistant-n2" + await hostname.set_static_hostname("StarWars") + assert hostname_service.SetStaticHostname.calls == [("StarWars", False)] + await hostname_service.ping() + assert hostname.hostname == "StarWars" diff --git a/tests/dbus/test_login.py b/tests/dbus/test_login.py index b75649575..4c7fd068d 100644 --- a/tests/dbus/test_login.py +++ b/tests/dbus/test_login.py @@ -1,29 +1,42 @@ """Test login dbus interface.""" +# pylint: disable=import-error +from dbus_fast.aio.message_bus import MessageBus import pytest -from supervisor.coresys import CoreSys +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 -async def test_reboot(coresys: CoreSys, dbus: list[str]): + +@pytest.fixture(name="logind_service", autouse=True) +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 = Logind() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.logind.reboot() + await logind.reboot() - await coresys.dbus.logind.connect(coresys.dbus.bus) + await logind.connect(dbus_session_bus) - dbus.clear() - assert await coresys.dbus.logind.reboot() is None - assert dbus == ["/org/freedesktop/login1-org.freedesktop.login1.Manager.Reboot"] + assert await logind.reboot() is None + assert logind_service.Reboot.calls == [(False,)] -async def test_power_off(coresys: CoreSys, dbus: list[str]): +async def test_power_off(logind_service: LogindService, dbus_session_bus: MessageBus): """Test power off.""" + logind = Logind() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.logind.power_off() + await logind.power_off() - await coresys.dbus.logind.connect(coresys.dbus.bus) + await logind.connect(dbus_session_bus) - dbus.clear() - assert await coresys.dbus.logind.power_off() is None - assert dbus == ["/org/freedesktop/login1-org.freedesktop.login1.Manager.PowerOff"] + assert await logind.power_off() is None + assert logind_service.PowerOff.calls == [(False,)] diff --git a/tests/dbus/test_rauc.py b/tests/dbus/test_rauc.py index 6b6e3c27a..8cb794def 100644 --- a/tests/dbus/test_rauc.py +++ b/tests/dbus/test_rauc.py @@ -1,61 +1,70 @@ """Test rauc dbus interface.""" -import asyncio - +# pylint: disable=import-error +from dbus_fast.aio.message_bus import MessageBus import pytest -from supervisor.coresys import CoreSys from supervisor.dbus.const import RaucState +from supervisor.dbus.rauc import Rauc from supervisor.exceptions import DBusNotConnectedError -from tests.common import fire_property_change_signal +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.rauc import Rauc as RaucService -async def test_rauc(coresys: CoreSys): +@pytest.fixture(name="rauc_service", autouse=True) +async def fixture_rauc_service(dbus_session_bus: MessageBus) -> RaucService: + """Mock rauc dbus service.""" + yield (await mock_dbus_services({"rauc": None}, dbus_session_bus))["rauc"] + + +async def test_rauc_info(rauc_service: RaucService, dbus_session_bus: MessageBus): """Test rauc properties.""" - assert coresys.dbus.rauc.boot_slot is None - assert coresys.dbus.rauc.operation is None - assert coresys.dbus.rauc.last_error is None + rauc = Rauc() - await coresys.dbus.rauc.connect(coresys.dbus.bus) - await coresys.dbus.rauc.update() + assert rauc.boot_slot is None + assert rauc.operation is None + assert rauc.last_error is None - assert coresys.dbus.rauc.boot_slot == "B" - assert coresys.dbus.rauc.operation == "idle" - assert coresys.dbus.rauc.last_error == "" + await rauc.connect(dbus_session_bus) - fire_property_change_signal(coresys.dbus.rauc, {"LastError": "Error!"}) - await asyncio.sleep(0) - assert coresys.dbus.rauc.last_error == "Error!" + assert rauc.boot_slot == "B" + assert rauc.operation == "idle" + assert rauc.last_error == "" - fire_property_change_signal(coresys.dbus.rauc, {}, ["LastError"]) - await asyncio.sleep(0) - assert coresys.dbus.rauc.last_error == "" + rauc_service.emit_properties_changed({"LastError": "Error!"}) + await rauc_service.ping() + assert rauc.last_error == "Error!" + + rauc_service.emit_properties_changed({}, ["LastError"]) + await rauc_service.ping() + await rauc_service.ping() # To process the follow-up get all properties call + assert rauc.last_error == "" -async def test_install(coresys: CoreSys, dbus: list[str]): +async def test_install(dbus_session_bus: MessageBus): """Test install.""" + rauc = Rauc() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.rauc.install("rauc_file") + await rauc.install("rauc_file") - await coresys.dbus.rauc.connect(coresys.dbus.bus) + await rauc.connect(dbus_session_bus) - dbus.clear() - async with coresys.dbus.rauc.signal_completed() as signal: - assert await coresys.dbus.rauc.install("rauc_file") is None + async with rauc.signal_completed() as signal: + assert await rauc.install("rauc_file") is None assert await signal.wait_for_signal() == [0] - assert dbus == ["/-de.pengutronix.rauc.Installer.Install"] - -async def test_get_slot_status(coresys: CoreSys, dbus: list[str]): +async def test_get_slot_status(dbus_session_bus: MessageBus): """Test get slot status.""" + rauc = Rauc() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.rauc.get_slot_status() + await rauc.get_slot_status() - await coresys.dbus.rauc.connect(coresys.dbus.bus) + await rauc.connect(dbus_session_bus) - dbus.clear() - slot_status = await coresys.dbus.rauc.get_slot_status() + slot_status = await rauc.get_slot_status() assert len(slot_status) == 6 assert slot_status[0][0] == "kernel.0" assert slot_status[0][1]["boot-status"] == "good" @@ -65,18 +74,17 @@ async def test_get_slot_status(coresys: CoreSys, dbus: list[str]): assert slot_status[4][1]["boot-status"] == "good" assert slot_status[4][1]["device"] == "/dev/disk/by-partlabel/hassos-kernel1" assert slot_status[4][1]["bootname"] == "B" - assert dbus == ["/-de.pengutronix.rauc.Installer.GetSlotStatus"] -async def test_mark(coresys: CoreSys, dbus: list[str]): +async def test_mark(dbus_session_bus: MessageBus): """Test mark.""" + rauc = Rauc() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.rauc.mark(RaucState.GOOD, "booted") + await rauc.mark(RaucState.GOOD, "booted") - await coresys.dbus.rauc.connect(coresys.dbus.bus) + await rauc.connect(dbus_session_bus) - dbus.clear() - mark = await coresys.dbus.rauc.mark(RaucState.GOOD, "booted") + mark = await rauc.mark(RaucState.GOOD, "booted") assert mark[0] == "kernel.1" assert mark[1] == "marked slot kernel.1 as good" - assert dbus == ["/-de.pengutronix.rauc.Installer.Mark"] diff --git a/tests/dbus/test_resolved.py b/tests/dbus/test_resolved.py index 6461504f5..cdde733db 100644 --- a/tests/dbus/test_resolved.py +++ b/tests/dbus/test_resolved.py @@ -1,12 +1,11 @@ """Test systemd-resolved dbus interface.""" +# pylint: disable=import-error -import asyncio from socket import AF_INET6, inet_aton, inet_pton -from unittest.mock import patch +from dbus_fast.aio.message_bus import MessageBus import pytest -from supervisor.coresys import CoreSys from supervisor.dbus.const import ( DNSOverTLSEnabled, DNSSECValidation, @@ -14,78 +13,56 @@ from supervisor.dbus.const import ( MulticastProtocolEnabled, ResolvConfMode, ) +from supervisor.dbus.resolved import Resolved -from tests.common import fire_property_change_signal - -DNS_IP_FIELDS = [ - "DNS", - "DNSEx", - "FallbackDNS", - "FallbackDNSEx", - "CurrentDNSServer", - "CurrentDNSServerEx", -] +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.resolved import Resolved as ResolvedService -@pytest.fixture(name="coresys_ip_bytes") -async def fixture_coresys_ip_bytes(coresys: CoreSys) -> CoreSys: - """Coresys with ip addresses correctly mocked as bytes.""" - get_properties = coresys.dbus.network.dbus.get_properties - - async def mock_get_properties(dbus_obj, interface): - reply = await get_properties(interface) - - for field in DNS_IP_FIELDS: - if field in reply and len(reply[field]) > 0: - if isinstance(reply[field][0], list): - for entry in reply[field]: - entry[2] = bytes(entry[2]) - else: - reply[field][2] = bytes(reply[field][2]) - - return reply - - with patch("supervisor.utils.dbus.DBus.get_properties", new=mock_get_properties): - yield coresys +@pytest.fixture(name="resolved_service", autouse=True) +async def fixture_resolved_service(dbus_session_bus: MessageBus) -> ResolvedService: + """Mock resolved dbus service.""" + yield (await mock_dbus_services({"resolved": None}, dbus_session_bus))["resolved"] -async def test_dbus_resolved_info(coresys_ip_bytes: CoreSys): +async def test_dbus_resolved_info( + resolved_service: ResolvedService, dbus_session_bus: MessageBus +): """Test systemd-resolved dbus connection.""" - coresys = coresys_ip_bytes + resolved = Resolved() - assert coresys.dbus.resolved.dns is None + assert resolved.dns is None - await coresys.dbus.resolved.connect(coresys.dbus.bus) - await coresys.dbus.resolved.update() + await resolved.connect(dbus_session_bus) - assert coresys.dbus.resolved.llmnr_hostname == "homeassistant" - assert coresys.dbus.resolved.llmnr == MulticastProtocolEnabled.YES - assert coresys.dbus.resolved.multicast_dns == MulticastProtocolEnabled.RESOLVE - assert coresys.dbus.resolved.dns_over_tls == DNSOverTLSEnabled.NO + assert resolved.llmnr_hostname == "homeassistant" + assert resolved.llmnr == MulticastProtocolEnabled.YES + assert resolved.multicast_dns == MulticastProtocolEnabled.RESOLVE + assert resolved.dns_over_tls == DNSOverTLSEnabled.NO - assert len(coresys.dbus.resolved.dns) == 2 - assert coresys.dbus.resolved.dns[0] == [0, 2, inet_aton("127.0.0.1")] - assert coresys.dbus.resolved.dns[1] == [0, 10, inet_pton(AF_INET6, "::1")] - assert len(coresys.dbus.resolved.dns_ex) == 2 - assert coresys.dbus.resolved.dns_ex[0] == [0, 2, inet_aton("127.0.0.1"), 0, ""] - assert coresys.dbus.resolved.dns_ex[1] == [0, 10, inet_pton(AF_INET6, "::1"), 0, ""] + assert len(resolved.dns) == 2 + assert resolved.dns[0] == [0, 2, inet_aton("127.0.0.1")] + assert resolved.dns[1] == [0, 10, inet_pton(AF_INET6, "::1")] + assert len(resolved.dns_ex) == 2 + assert resolved.dns_ex[0] == [0, 2, inet_aton("127.0.0.1"), 0, ""] + assert resolved.dns_ex[1] == [0, 10, inet_pton(AF_INET6, "::1"), 0, ""] - assert len(coresys.dbus.resolved.fallback_dns) == 2 - assert coresys.dbus.resolved.fallback_dns[0] == [0, 2, inet_aton("1.1.1.1")] - assert coresys.dbus.resolved.fallback_dns[1] == [ + assert len(resolved.fallback_dns) == 2 + assert resolved.fallback_dns[0] == [0, 2, inet_aton("1.1.1.1")] + assert resolved.fallback_dns[1] == [ 0, 10, inet_pton(AF_INET6, "2606:4700:4700::1111"), ] - assert len(coresys.dbus.resolved.fallback_dns_ex) == 2 - assert coresys.dbus.resolved.fallback_dns_ex[0] == [ + assert len(resolved.fallback_dns_ex) == 2 + assert resolved.fallback_dns_ex[0] == [ 0, 2, inet_aton("1.1.1.1"), 0, "cloudflare-dns.com", ] - assert coresys.dbus.resolved.fallback_dns_ex[1] == [ + assert resolved.fallback_dns_ex[1] == [ 0, 10, inet_pton(AF_INET6, "2606:4700:4700::1111"), @@ -93,8 +70,8 @@ async def test_dbus_resolved_info(coresys_ip_bytes: CoreSys): "cloudflare-dns.com", ] - assert coresys.dbus.resolved.current_dns_server == [0, 2, inet_aton("127.0.0.1")] - assert coresys.dbus.resolved.current_dns_server_ex == [ + assert resolved.current_dns_server == [0, 2, inet_aton("127.0.0.1")] + assert resolved.current_dns_server_ex == [ 0, 2, inet_aton("127.0.0.1"), @@ -102,25 +79,26 @@ async def test_dbus_resolved_info(coresys_ip_bytes: CoreSys): "", ] - assert len(coresys.dbus.resolved.domains) == 1 - assert coresys.dbus.resolved.domains[0] == [0, "local.hass.io", False] + assert len(resolved.domains) == 1 + assert resolved.domains[0] == [0, "local.hass.io", False] - assert coresys.dbus.resolved.transaction_statistics == [0, 100000] - assert coresys.dbus.resolved.cache_statistics == [10, 50000, 10000] - assert coresys.dbus.resolved.dnssec == DNSSECValidation.NO - assert coresys.dbus.resolved.dnssec_statistics == [0, 0, 0, 0] - assert coresys.dbus.resolved.dnssec_supported is False - assert coresys.dbus.resolved.dnssec_negative_trust_anchors == [ + assert resolved.transaction_statistics == [0, 100000] + assert resolved.cache_statistics == [10, 50000, 10000] + assert resolved.dnssec == DNSSECValidation.NO + assert resolved.dnssec_statistics == [0, 0, 0, 0] + assert resolved.dnssec_supported is False + assert resolved.dnssec_negative_trust_anchors == [ "168.192.in-addr.arpa", "local", ] - assert coresys.dbus.resolved.dns_stub_listener == DNSStubListenerEnabled.NO - assert coresys.dbus.resolved.resolv_conf_mode == ResolvConfMode.FOREIGN + assert resolved.dns_stub_listener == DNSStubListenerEnabled.NO + assert resolved.resolv_conf_mode == ResolvConfMode.FOREIGN - fire_property_change_signal(coresys.dbus.resolved, {"LLMNRHostname": "test"}) - await asyncio.sleep(0) - assert coresys.dbus.resolved.llmnr_hostname == "test" + resolved_service.emit_properties_changed({"LLMNRHostname": "test"}) + await resolved_service.ping() + assert resolved.llmnr_hostname == "test" - fire_property_change_signal(coresys.dbus.resolved, {}, ["LLMNRHostname"]) - await asyncio.sleep(0) - assert coresys.dbus.resolved.llmnr_hostname == "homeassistant" + resolved_service.emit_properties_changed({}, ["LLMNRHostname"]) + await resolved_service.ping() + await resolved_service.ping() # To process the follow-up get all properties call + assert resolved.llmnr_hostname == "homeassistant" diff --git a/tests/dbus/test_systemd.py b/tests/dbus/test_systemd.py index 5312cd740..ec8a5ba28 100644 --- a/tests/dbus/test_systemd.py +++ b/tests/dbus/test_systemd.py @@ -1,143 +1,142 @@ """Test hostname dbus interface.""" - -from unittest.mock import patch - +# pylint: disable=import-error +from dbus_fast.aio.message_bus import MessageBus import pytest -from supervisor.coresys import CoreSys -from supervisor.dbus.const import DBUS_NAME_SYSTEMD +from supervisor.dbus.systemd import Systemd from supervisor.exceptions import DBusNotConnectedError -from tests.common import load_json_fixture +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.systemd import Systemd as SystemdService -async def test_dbus_systemd_info(coresys: CoreSys): - """Test coresys dbus connection.""" - assert coresys.dbus.systemd.boot_timestamp is None - assert coresys.dbus.systemd.startup_time is None - - await coresys.dbus.systemd.connect(coresys.dbus.bus) - - async def mock_get_properties(dbus_obj, interface): - return load_json_fixture( - f"{DBUS_NAME_SYSTEMD.replace('.', '_')}_properties.json" - ) - - with patch("supervisor.utils.dbus.DBus.get_properties", new=mock_get_properties): - await coresys.dbus.systemd.update() - - assert coresys.dbus.systemd.boot_timestamp == 1632236713344227 - assert coresys.dbus.systemd.startup_time == 45.304696 +@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_reboot(coresys: CoreSys, dbus: list[str]): +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_reboot(systemd_service: SystemdService, dbus_session_bus: MessageBus): """Test reboot.""" + systemd = Systemd() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.systemd.reboot() + await systemd.reboot() - await coresys.dbus.systemd.connect(coresys.dbus.bus) + await systemd.connect(dbus_session_bus) - dbus.clear() - assert await coresys.dbus.systemd.reboot() is None - assert dbus == ["/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.Reboot"] + assert await systemd.reboot() is None + assert systemd_service.Reboot.calls == [tuple()] -async def test_power_off(coresys: CoreSys, dbus: list[str]): +async def test_power_off(systemd_service: SystemdService, dbus_session_bus: MessageBus): """Test power off.""" + systemd = Systemd() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.systemd.power_off() + await systemd.power_off() - await coresys.dbus.systemd.connect(coresys.dbus.bus) + await systemd.connect(dbus_session_bus) - dbus.clear() - assert await coresys.dbus.systemd.power_off() is None - assert dbus == [ - "/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.PowerOff" - ] + assert await systemd.power_off() is None + assert systemd_service.PowerOff.calls == [tuple()] -async def test_start_unit(coresys: CoreSys, dbus: list[str]): +async def test_start_unit( + systemd_service: SystemdService, dbus_session_bus: MessageBus +): """Test start unit.""" + systemd = Systemd() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.systemd.start_unit("test_unit", "replace") + await systemd.start_unit("test_unit", "replace") - await coresys.dbus.systemd.connect(coresys.dbus.bus) + await systemd.connect(dbus_session_bus) - dbus.clear() assert ( - await coresys.dbus.systemd.start_unit("test_unit", "replace") + await systemd.start_unit("test_unit", "replace") == "/org/freedesktop/systemd1/job/7623" ) - assert dbus == [ - "/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.StartUnit" - ] + assert systemd_service.StartUnit.calls == [("test_unit", "replace")] -async def test_stop_unit(coresys: CoreSys, dbus: list[str]): +async def test_stop_unit(systemd_service: SystemdService, dbus_session_bus: MessageBus): """Test stop unit.""" + systemd = Systemd() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.systemd.stop_unit("test_unit", "replace") + await systemd.stop_unit("test_unit", "replace") - await coresys.dbus.systemd.connect(coresys.dbus.bus) + await systemd.connect(dbus_session_bus) - dbus.clear() assert ( - await coresys.dbus.systemd.stop_unit("test_unit", "replace") + await systemd.stop_unit("test_unit", "replace") == "/org/freedesktop/systemd1/job/7623" ) - assert dbus == [ - "/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.StopUnit" - ] + assert systemd_service.StopUnit.calls == [("test_unit", "replace")] -async def test_restart_unit(coresys: CoreSys, dbus: list[str]): +async def test_restart_unit( + systemd_service: SystemdService, dbus_session_bus: MessageBus +): """Test restart unit.""" + systemd = Systemd() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.systemd.restart_unit("test_unit", "replace") + await systemd.restart_unit("test_unit", "replace") - await coresys.dbus.systemd.connect(coresys.dbus.bus) + await systemd.connect(dbus_session_bus) - dbus.clear() assert ( - await coresys.dbus.systemd.restart_unit("test_unit", "replace") + await systemd.restart_unit("test_unit", "replace") == "/org/freedesktop/systemd1/job/7623" ) - assert dbus == [ - "/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.RestartUnit" - ] + assert systemd_service.RestartUnit.calls == [("test_unit", "replace")] -async def test_reload_unit(coresys: CoreSys, dbus: list[str]): +async def test_reload_unit( + systemd_service: SystemdService, dbus_session_bus: MessageBus +): """Test reload unit.""" + systemd = Systemd() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.systemd.reload_unit("test_unit", "replace") + await systemd.reload_unit("test_unit", "replace") - await coresys.dbus.systemd.connect(coresys.dbus.bus) + await systemd.connect(dbus_session_bus) - dbus.clear() assert ( - await coresys.dbus.systemd.reload_unit("test_unit", "replace") + await systemd.reload_unit("test_unit", "replace") == "/org/freedesktop/systemd1/job/7623" ) - assert dbus == [ - "/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.ReloadOrRestartUnit" - ] + assert systemd_service.ReloadOrRestartUnit.calls == [("test_unit", "replace")] -async def test_list_units(coresys: CoreSys, dbus: list[str]): +async def test_list_units(dbus_session_bus: MessageBus): """Test list units.""" + systemd = Systemd() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.systemd.list_units() + await systemd.list_units() - await coresys.dbus.systemd.connect(coresys.dbus.bus) + await systemd.connect(dbus_session_bus) - dbus.clear() - units = await coresys.dbus.systemd.list_units() + 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" - assert dbus == [ - "/org/freedesktop/systemd1-org.freedesktop.systemd1.Manager.ListUnits" - ] diff --git a/tests/dbus/test_timedate.py b/tests/dbus/test_timedate.py index 641062890..90a11554f 100644 --- a/tests/dbus/test_timedate.py +++ b/tests/dbus/test_timedate.py @@ -1,62 +1,81 @@ """Test TimeDate dbus interface.""" -import asyncio +# pylint: disable=import-error from datetime import datetime, timezone +from dbus_fast.aio.message_bus import MessageBus import pytest -from supervisor.coresys import CoreSys +from supervisor.dbus.timedate import TimeDate from supervisor.exceptions import DBusNotConnectedError -from tests.common import fire_property_change_signal +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.timedate import TimeDate as TimeDateService -async def test_dbus_timezone(coresys: CoreSys): - """Test coresys dbus connection.""" - assert coresys.dbus.timedate.dt_utc is None - assert coresys.dbus.timedate.ntp is None +@pytest.fixture(name="timedate_service", autouse=True) +async def fixture_timedate_service(dbus_session_bus: MessageBus) -> TimeDateService: + """Mock timedate dbus service.""" + yield (await mock_dbus_services({"timedate": None}, dbus_session_bus))["timedate"] - await coresys.dbus.timedate.connect(coresys.dbus.bus) - await coresys.dbus.timedate.update() - assert coresys.dbus.timedate.dt_utc == datetime( +async def test_timedate_info( + timedate_service: TimeDateService, dbus_session_bus: MessageBus +): + """Test timedate properties.""" + timedate = TimeDate() + + assert timedate.dt_utc is None + assert timedate.ntp is None + + await timedate.connect(dbus_session_bus) + + assert timedate.dt_utc == datetime( 2021, 5, 19, 8, 36, 54, 405718, tzinfo=timezone.utc ) - assert coresys.dbus.timedate.ntp is True + assert timedate.ntp is True - assert ( - coresys.dbus.timedate.dt_utc.isoformat() == "2021-05-19T08:36:54.405718+00:00" - ) + assert timedate.dt_utc.isoformat() == "2021-05-19T08:36:54.405718+00:00" - fire_property_change_signal(coresys.dbus.timedate, {"NTP": False}) - await asyncio.sleep(0) - assert coresys.dbus.timedate.ntp is False + timedate_service.emit_properties_changed({"NTP": False}) + await timedate_service.ping() + assert timedate.ntp is False - fire_property_change_signal(coresys.dbus.timedate, {}, ["NTP"]) - await asyncio.sleep(0) - assert coresys.dbus.timedate.ntp is True + timedate_service.emit_properties_changed({}, ["NTP"]) + await timedate_service.ping() + await timedate_service.ping() # To process the follow-up get all properties call + assert timedate.ntp is True -async def test_dbus_settime(coresys: CoreSys, dbus: list[str]): +async def test_dbus_settime( + timedate_service: TimeDateService, dbus_session_bus: MessageBus +): """Set timestamp on backend.""" + timedate = TimeDate() + test_dt = datetime(2021, 5, 19, 8, 36, 54, 405718, tzinfo=timezone.utc) with pytest.raises(DBusNotConnectedError): - await coresys.dbus.timedate.set_time(test_dt) + await timedate.set_time(test_dt) - await coresys.dbus.timedate.connect(coresys.dbus.bus) + await timedate.connect(dbus_session_bus) - dbus.clear() - assert await coresys.dbus.timedate.set_time(test_dt) is None - assert dbus == ["/org/freedesktop/timedate1-org.freedesktop.timedate1.SetTime"] + assert await timedate.set_time(test_dt) is None + assert timedate_service.SetTime.calls == [(1621413414405718, False, False)] -async def test_dbus_setntp(coresys: CoreSys, dbus: list[str]): +async def test_dbus_setntp( + timedate_service: TimeDateService, dbus_session_bus: MessageBus +): """Disable NTP on backend.""" + timedate = TimeDate() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.timedate.set_ntp(False) + await timedate.set_ntp(False) - await coresys.dbus.timedate.connect(coresys.dbus.bus) + await timedate.connect(dbus_session_bus) - dbus.clear() - assert await coresys.dbus.timedate.set_ntp(False) is None - assert dbus == ["/org/freedesktop/timedate1-org.freedesktop.timedate1.SetNTP"] + assert timedate.ntp is True + assert await timedate.set_ntp(False) is None + assert timedate_service.SetNTP.calls == [(False, False)] + await timedate_service.ping() + assert timedate.ntp is False diff --git a/tests/dbus_service_mocks/__init__.py b/tests/dbus_service_mocks/__init__.py new file mode 100644 index 000000000..3cff67074 --- /dev/null +++ b/tests/dbus_service_mocks/__init__.py @@ -0,0 +1,6 @@ +"""Mocks of dbus services supervisor uses. + +The introspection of these mock services must match the real service for +the parts supervisor uses. Parts supervisor does not use can be omitted +for sake of brevity since the mocks are hand-written (for now). +""" diff --git a/tests/dbus_service_mocks/base.py b/tests/dbus_service_mocks/base.py new file mode 100644 index 000000000..b461d7e14 --- /dev/null +++ b/tests/dbus_service_mocks/base.py @@ -0,0 +1,70 @@ +"""Baseclass for a dbus service mock.""" + +import asyncio +from functools import wraps +from typing import Any, no_type_check_decorator + +from dbus_fast import Message +from dbus_fast.aio.message_bus import MessageBus +from dbus_fast.service import ServiceInterface, method + +# pylint: disable=invalid-name + + +def dbus_method(name: str = None, disabled: bool = False): + """Make DBus method with call tracking. + + Identical to dbus_fast.service.method wrapper except all calls to it are tracked. + Can then test that methods with no output were called or the right arguments were + used if the output is static. + """ + orig_decorator = method(name=name, disabled=disabled) + + @no_type_check_decorator + def decorator(fn): + calls: list[list[Any]] = [] + + @wraps(fn) + def track_calls(self, *args): + calls.append(args) + return fn(self, *args) + + wrapped = orig_decorator(track_calls) + wrapped.__dict__["calls"] = calls + + return wrapped + + return decorator + + +class DBusServiceMock(ServiceInterface): + """Base dbus service mock.""" + + object_path: str + interface: str + bus: MessageBus | None = None + + def __init__(self): + """Initialize dbus service mock.""" + super().__init__(self.interface) + + def export(self, bus: MessageBus) -> "DBusServiceMock": + """Export object onto bus.""" + self.bus = bus + bus.export(self.object_path, self) + return self + + async def ping(self, *, sleep: bool = True): + """Ping object to check for signals.""" + await self.bus.call( + Message( + destination=self.bus.unique_name, + interface="org.freedesktop.DBus.Peer", + path=self.object_path, + member="Ping", + ) + ) + # This is called to force dbus to process messages for the object + # So in general we sleep(0) after to clear the new task + if sleep: + await asyncio.sleep(0) diff --git a/tests/dbus_service_mocks/hostname.py b/tests/dbus_service_mocks/hostname.py new file mode 100644 index 000000000..292dcdda8 --- /dev/null +++ b/tests/dbus_service_mocks/hostname.py @@ -0,0 +1,158 @@ +"""Mock of hostname dbus service.""" + +from json import dumps + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.hostname1" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Hostname() + + +# pylint: disable=invalid-name + + +class Hostname(DBusServiceMock): + """Hostname mock. + + gdbus introspect --system --dest org.freedesktop.hostname1 --object-path /org/freedesktop/hostname1 + """ + + object_path = "/org/freedesktop/hostname1" + interface = "org.freedesktop.hostname1" + + @dbus_property(access=PropertyAccess.READ) + def Hostname(self) -> "s": + """Get Hostname.""" + return "homeassistant-n2" + + @dbus_property(access=PropertyAccess.READ) + def StaticHostname(self) -> "s": + """Get StaticHostname.""" + return "homeassistant-n2" + + @dbus_property(access=PropertyAccess.READ) + def PrettyHostname(self) -> "s": + """Get PrettyHostname.""" + return "" + + @dbus_property(access=PropertyAccess.READ) + def IconName(self) -> "s": + """Get IconName.""" + return "computer-embedded" + + @dbus_property(access=PropertyAccess.READ) + def Chassis(self) -> "s": + """Get Chassis.""" + return "embedded" + + @dbus_property(access=PropertyAccess.READ) + def Deployment(self) -> "s": + """Get Deployment.""" + return "development" + + @dbus_property(access=PropertyAccess.READ) + def Location(self) -> "s": + """Get Location.""" + return "" + + @dbus_property(access=PropertyAccess.READ) + def KernelName(self) -> "s": + """Get KernelName.""" + return "Linux" + + @dbus_property(access=PropertyAccess.READ) + def KernelRelease(self) -> "s": + """Get KernelRelease.""" + return "5.10.33" + + @dbus_property(access=PropertyAccess.READ) + def KernelVersion(self) -> "s": + """Get KernelVersion.""" + return "#1 SMP PREEMPT Wed May 5 00:55:38 UTC 2021" + + @dbus_property(access=PropertyAccess.READ) + def OperatingSystemPrettyName(self) -> "s": + """Get OperatingSystemPrettyName.""" + return "Home Assistant OS 6.0.dev20210504" + + @dbus_property(access=PropertyAccess.READ) + def OperatingSystemCPEName(self) -> "s": + """Get OperatingSystemCPEName.""" + return "cpe:2.3:o:home-assistant:haos:6.0.dev20210504:*:development:*:*:*:odroid-n2:*" + + @dbus_property(access=PropertyAccess.READ) + def HomeURL(self) -> "s": + """Get HomeURL.""" + return "https://hass.io/" + + @dbus_method() + def SetHostname(self, hostname: "s", interactive: "b") -> None: + """Set hostname.""" + self.emit_properties_changed({"Hostname": hostname}) + + @dbus_method() + def SetStaticHostname(self, hostname: "s", interactive: "b") -> None: + """Set static hostname.""" + self.emit_properties_changed({"StaticHostname": hostname}) + + @dbus_method() + def SetPrettyHostname(self, hostname: "s", interactive: "b") -> None: + """Set pretty hostname.""" + self.emit_properties_changed({"PrettyHostname": hostname}) + + @dbus_method() + def SetIconName(self, icon: "s", interactive: "b") -> None: + """Set icon name.""" + self.emit_properties_changed({"IconName": icon}) + + @dbus_method() + def SetChassis(self, chassis: "s", interactive: "b") -> None: + """Set chassis.""" + self.emit_properties_changed({"Chassis": chassis}) + + @dbus_method() + def SetDeployment(self, deployment: "s", interactive: "b") -> None: + """Set deployment.""" + self.emit_properties_changed({"Deployment": deployment}) + + @dbus_method() + def SetLocation(self, location: "s", interactive: "b") -> None: + """Set location.""" + self.emit_properties_changed({"Location": location}) + + @dbus_method() + def GetProductUUID(self, interactive: "b") -> "ay": + """Get product UUID.""" + return bytearray("d153e353-2a32-4763-b930-b27fbc980da5", encoding="utf-8") + + @dbus_method() + def Describe(self) -> "s": + """Describe.""" + return dumps( + { + "Hostname": "odroid-dev", + "StaticHostname": "odroid-dev", + "PrettyHostname": None, + "DefaultHostname": "homeassistant", + "HostnameSource": "static", + "IconName": "computer-embedded", + "Chassis": "embedded", + "Deployment": "development", + "Location": None, + "KernelName": "Linux", + "KernelRelease": "5.15.88", + "KernelVersion": "#1 SMP PREEMPT Mon Jan 16 23:45:23 UTC 2023", + "OperatingSystemPrettyName": "Home Assistant OS 10.0.dev20230116", + "OperatingSystemCPEName": "cpe:2.3:o:home-assistant:haos:10.0.dev20230116:*:development:*:*:*:odroid-n2:*", + "OperatingSystemHomeURL": "https://hass.io/", + "HardwareVendor": None, + "HardwareModel": None, + "ProductUUID": None, + } + ) diff --git a/tests/dbus_service_mocks/logind.py b/tests/dbus_service_mocks/logind.py new file mode 100644 index 000000000..99a0e1a39 --- /dev/null +++ b/tests/dbus_service_mocks/logind.py @@ -0,0 +1,31 @@ +"""Mock of logind dbus service.""" + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.login1" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Logind() + + +# pylint: disable=invalid-name,missing-function-docstring + + +class Logind(DBusServiceMock): + """Logind mock. + + gdbus introspect --system --dest org.freedesktop.login1 --object-path /org/freedesktop/login1 + """ + + object_path = "/org/freedesktop/login1" + interface = "org.freedesktop.login1.Manager" + + @dbus_method() + def Reboot(self, interactive: "b") -> None: + """Reboot.""" + + @dbus_method() + def PowerOff(self, interactive: "b") -> None: + """PowerOff.""" diff --git a/tests/dbus_service_mocks/rauc.py b/tests/dbus_service_mocks/rauc.py new file mode 100644 index 000000000..72b302882 --- /dev/null +++ b/tests/dbus_service_mocks/rauc.py @@ -0,0 +1,211 @@ +"""Mock of rauc dbus service.""" + +from dbus_fast import Variant +from dbus_fast.service import PropertyAccess, dbus_property, signal + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "de.pengutronix.rauc" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Rauc() + + +# pylint: disable=invalid-name + + +class Rauc(DBusServiceMock): + """Rauc mock. + + gdbus introspect --system --dest de.pengutronix.rauc --object-path / + """ + + object_path = "/" + interface = "de.pengutronix.rauc.Installer" + + @dbus_property(access=PropertyAccess.READ) + def Operation(self) -> "s": + """Return operation.""" + return "idle" + + @dbus_property(access=PropertyAccess.READ) + def LastError(self) -> "s": + """Return last error.""" + return "" + + @dbus_property(access=PropertyAccess.READ) + def Progress(self) -> "(isi)": + """Return progress.""" + return [0, "", 0] + + @dbus_property(access=PropertyAccess.READ) + def Compatible(self) -> "s": + """Return compatible.""" + return "haos-odroid-n2" + + @dbus_property(access=PropertyAccess.READ) + def Variant(self) -> "s": + """Return variant.""" + return "" + + @dbus_property(access=PropertyAccess.READ) + def BootSlot(self) -> "s": + """Return boot slot.""" + return "B" + + @signal() + def Completed(self) -> "i": + """Signal completed.""" + return 0 + + @dbus_method() + def Install(self, source: "s"): + """Install source.""" + self.Completed() + + @dbus_method() + def InstallBundle(self, source: "s", arg: "a{sv}"): + """Install source bundle.""" + self.Completed() + + @dbus_method() + def Mark(self, state: "s", slot_identifier: "s") -> "ss": + """Mark slot.""" + return ["kernel.1", "marked slot kernel.1 as good"] + + @dbus_method() + def GetPrimary(self) -> "s": + """Get primary slot.""" + return "kernel.0" + + @dbus_method() + def GetSlotStatus(self) -> "a(sa{sv})": + """Get slot status.""" + return [ + ( + "kernel.0", + { + "activated.count": Variant("u", 9), + "activated.timestamp": Variant("s", "2022-08-23T21:03:22Z"), + "boot-status": Variant("s", "good"), + "bundle.compatible": Variant("s", "haos-odroid-n2"), + "sha256": Variant( + "s", + "c624db648b8401fae37ee5bb1a6ec90bdf4183aef364b33314a73c7198e49d5b", + ), + "state": Variant("s", "inactive"), + "size": Variant("t", 10371072), + "installed.count": Variant("u", 9), + "class": Variant("s", "kernel"), + "device": Variant("s", "/dev/disk/by-partlabel/hassos-kernel0"), + "type": Variant("s", "raw"), + "bootname": Variant("s", "A"), + "bundle.version": Variant("s", "9.0.dev20220818"), + "installed.timestamp": Variant("s", "2022-08-23T21:03:16Z"), + "status": Variant("s", "ok"), + }, + ), + ( + "boot.0", + { + "bundle.compatible": Variant("s", "haos-odroid-n2"), + "sha256": Variant( + "s", + "a5019b335f33be2cf89c96bb2d0695030adb72c1d13d650a5bbe1806dd76d6cc", + ), + "state": Variant("s", "inactive"), + "size": Variant("t", 25165824), + "installed.count": Variant("u", 19), + "class": Variant("s", "boot"), + "device": Variant("s", "/dev/disk/by-partlabel/hassos-boot"), + "type": Variant("s", "vfat"), + "status": Variant("s", "ok"), + "bundle.version": Variant("s", "9.0.dev20220824"), + "installed.timestamp": Variant("s", "2022-08-25T21:11:46Z"), + }, + ), + ( + "rootfs.0", + { + "bundle.compatible": Variant("s", "haos-odroid-n2"), + "parent": Variant("s", "kernel.0"), + "state": Variant("s", "inactive"), + "size": Variant("t", 117456896), + "sha256": Variant( + "s", + "7d908b4d578d072b1b0f75de8250fd97b6e119bff09518a96fffd6e4aec61721", + ), + "class": Variant("s", "rootfs"), + "device": Variant("s", "/dev/disk/by-partlabel/hassos-system0"), + "type": Variant("s", "raw"), + "status": Variant("s", "ok"), + "bundle.version": Variant("s", "9.0.dev20220818"), + "installed.timestamp": Variant("s", "2022-08-23T21:03:21Z"), + "installed.count": Variant("u", 9), + }, + ), + ( + "spl.0", + { + "bundle.compatible": Variant("s", "haos-odroid-n2"), + "sha256": Variant( + "s", + "9856a94df1d6abbc672adaf95746ec76abd3a8191f9d08288add6bb39e63ef45", + ), + "state": Variant("s", "inactive"), + "size": Variant("t", 8388608), + "installed.count": Variant("u", 19), + "class": Variant("s", "spl"), + "device": Variant("s", "/dev/disk/by-partlabel/hassos-boot"), + "type": Variant("s", "raw"), + "status": Variant("s", "ok"), + "bundle.version": Variant("s", "9.0.dev20220824"), + "installed.timestamp": Variant("s", "2022-08-25T21:11:51Z"), + }, + ), + ( + "kernel.1", + { + "activated.count": Variant("u", 10), + "activated.timestamp": Variant("s", "2022-08-25T21:11:52Z"), + "boot-status": Variant("s", "good"), + "bundle.compatible": Variant("s", "haos-odroid-n2"), + "sha256": Variant( + "s", + "f57e354b8bd518022721e71fafaf278972af966d8f6cbefb4610db13785801c8", + ), + "state": Variant("s", "booted"), + "size": Variant("t", 10371072), + "installed.count": Variant("u", 10), + "class": Variant("s", "kernel"), + "device": Variant("s", "/dev/disk/by-partlabel/hassos-kernel1"), + "type": Variant("s", "raw"), + "bootname": Variant("s", "B"), + "bundle.version": Variant("s", "9.0.dev20220824"), + "installed.timestamp": Variant("s", "2022-08-25T21:11:46Z"), + "status": Variant("s", "ok"), + }, + ), + ( + "rootfs.1", + { + "bundle.compatible": Variant("s", "haos-odroid-n2"), + "parent": Variant("s", "kernel.1"), + "state": Variant("s", "active"), + "size": Variant("t", 117456896), + "sha256": Variant( + "s", + "55936b64d391954ae1aed24dd1460e191e021e78655470051fa7939d12fff68a", + ), + "class": Variant("s", "rootfs"), + "device": Variant("s", "/dev/disk/by-partlabel/hassos-system1"), + "type": Variant("s", "raw"), + "status": Variant("s", "ok"), + "bundle.version": Variant("s", "9.0.dev20220824"), + "installed.timestamp": Variant("s", "2022-08-25T21:11:51Z"), + "installed.count": Variant("u", 10), + }, + ), + ] diff --git a/tests/dbus_service_mocks/resolved.py b/tests/dbus_service_mocks/resolved.py new file mode 100644 index 000000000..313d8e5dd --- /dev/null +++ b/tests/dbus_service_mocks/resolved.py @@ -0,0 +1,138 @@ +"""Mock of resolved dbus service.""" + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock + +BUS_NAME = "org.freedesktop.resolve1" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Resolved() + + +# pylint: disable=invalid-name + + +class Resolved(DBusServiceMock): + """Resolved mock. + + gdbus introspect --system --dest org.freedesktop.resolve1 --object-path /org/freedesktop/resolve1 + """ + + object_path = "/org/freedesktop/resolve1" + interface = "org.freedesktop.resolve1.Manager" + + @dbus_property(access=PropertyAccess.READ) + def LLMNRHostname(self) -> "s": + """Get LLMNRHostname.""" + return "homeassistant" + + @dbus_property(access=PropertyAccess.READ) + def LLMNR(self) -> "s": + """Get LLMNR.""" + return "yes" + + @dbus_property(access=PropertyAccess.READ) + def MulticastDNS(self) -> "s": + """Get MulticastDNS.""" + return "resolve" + + @dbus_property(access=PropertyAccess.READ) + def DNSOverTLS(self) -> "s": + """Get DNSOverTLS.""" + return "no" + + @dbus_property(access=PropertyAccess.READ) + def DNS(self) -> "a(iiay)": + """Get DNS.""" + return [ + [0, 2, bytearray([127, 0, 0, 1])], + [0, 10, bytearray([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1])], + ] + + @dbus_property(access=PropertyAccess.READ) + def DNSEx(self) -> "a(iiayqs)": + """Get DNSEx.""" + return [ + [0, 2, bytearray([127, 0, 0, 1]), 0, ""], + [0, 10, bytearray([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]), 0, ""], + ] + + @dbus_property(access=PropertyAccess.READ) + def FallbackDNS(self) -> "a(iiay)": + """Get FallbackDNS.""" + return [ + [0, 2, bytearray([1, 1, 1, 1])], + [0, 10, bytearray([38, 6, 71, 0, 71, 0, 0, 0, 0, 0, 0, 0, 0, 0, 17, 17])], + ] + + @dbus_property(access=PropertyAccess.READ) + def FallbackDNSEx(self) -> "a(iiayqs)": + """Get FallbackDNSEx.""" + return [ + [0, 2, bytearray([1, 1, 1, 1]), 0, "cloudflare-dns.com"], + [ + 0, + 10, + bytearray([38, 6, 71, 0, 71, 0, 0, 0, 0, 0, 0, 0, 0, 0, 17, 17]), + 0, + "cloudflare-dns.com", + ], + ] + + @dbus_property(access=PropertyAccess.READ) + def CurrentDNSServer(self) -> "(iiay)": + """Get CurrentDNSServer.""" + return [0, 2, bytearray([127, 0, 0, 1])] + + @dbus_property(access=PropertyAccess.READ) + def CurrentDNSServerEx(self) -> "(iiayqs)": + """Get CurrentDNSServerEx.""" + return [0, 2, bytearray([127, 0, 0, 1]), 0, ""] + + @dbus_property(access=PropertyAccess.READ) + def Domains(self) -> "a(isb)": + """Get Domains.""" + return [[0, "local.hass.io", False]] + + @dbus_property(access=PropertyAccess.READ) + def TransactionStatistics(self) -> "(tt)": + """Get TransactionStatistics.""" + return [0, 100000] + + @dbus_property(access=PropertyAccess.READ) + def CacheStatistics(self) -> "(ttt)": + """Get CacheStatistics.""" + return [10, 50000, 10000] + + @dbus_property(access=PropertyAccess.READ) + def DNSSEC(self) -> "s": + """Get DNSSEC.""" + return "no" + + @dbus_property(access=PropertyAccess.READ) + def DNSSECStatistics(self) -> "(tttt)": + """Get DNSSECStatistics.""" + return [0, 0, 0, 0] + + @dbus_property(access=PropertyAccess.READ) + def DNSSECSupported(self) -> "b": + """Get DNSSECSupported.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def DNSSECNegativeTrustAnchors(self) -> "as": + """Get DNSSECNegativeTrustAnchors.""" + return ["168.192.in-addr.arpa", "local"] + + @dbus_property(access=PropertyAccess.READ) + def DNSStubListener(self) -> "s": + """Get DNSStubListener.""" + return "no" + + @dbus_property(access=PropertyAccess.READ) + def ResolvConfMode(self) -> "s": + """Get ResolvConfMode.""" + return "foreign" diff --git a/tests/dbus_service_mocks/systemd.py b/tests/dbus_service_mocks/systemd.py new file mode 100644 index 000000000..8abfbd148 --- /dev/null +++ b/tests/dbus_service_mocks/systemd.py @@ -0,0 +1,685 @@ +"""Mock of systemd dbus service.""" + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.systemd1" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Systemd() + + +# pylint: disable=invalid-name,missing-function-docstring + + +class Systemd(DBusServiceMock): + """Systemd mock. + + gdbus introspect --system --dest org.freedesktop.systemd1 --object-path /org/freedesktop/systemd1 + """ + + object_path = "/org/freedesktop/systemd1" + interface = "org.freedesktop.systemd1.Manager" + + @dbus_property(access=PropertyAccess.READ) + def Version(self) -> "s": + """Get Version.""" + return "245.4-4ubuntu3.11" + + @dbus_property(access=PropertyAccess.READ) + def Features(self) -> "s": + """Get Features.""" + return "+PAM +AUDIT +SELINUX +IMA +APPARMOR +SMACK +SYSVINIT +UTMP +LIBCRYPTSETUP +GCRYPT +GNUTLS +ACL +XZ +LZ4 +SECCOMP +BLKID +ELFUTILS +KMOD +IDN2 -IDN +PCRE2 default-hierarchy=hybrid" + + @dbus_property(access=PropertyAccess.READ) + def Virtualization(self) -> "s": + """Get Virtualization.""" + return "" + + @dbus_property(access=PropertyAccess.READ) + def Architecture(self) -> "s": + """Get Architecture.""" + return "x86-64" + + @dbus_property(access=PropertyAccess.READ) + def Tainted(self) -> "s": + """Get Tainted.""" + return "" + + @dbus_property(access=PropertyAccess.READ) + def FirmwareTimestamp(self) -> "t": + """Get FirmwareTimestamp.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def FirmwareTimestampMonotonic(self) -> "t": + """Get FirmwareTimestampMonotonic.""" + return 28723572 + + @dbus_property(access=PropertyAccess.READ) + def LoaderTimestamp(self) -> "t": + """Get LoaderTimestamp.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def LoaderTimestampMonotonic(self) -> "t": + """Get LoaderTimestampMonotonic.""" + return 12402885 + + @dbus_property(access=PropertyAccess.READ) + def KernelTimestamp(self) -> "t": + """Get KernelTimestamp.""" + return 1632236694969442 + + @dbus_property(access=PropertyAccess.READ) + def KernelTimestampMonotonic(self) -> "t": + """Get KernelTimestampMonotonic.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDTimestamp(self) -> "t": + """Get InitRDTimestamp.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDTimestampMonotonic(self) -> "t": + """Get InitRDTimestampMonotonic.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def UserspaceTimestamp(self) -> "t": + """Get UserspaceTimestamp.""" + return 1632236699147681 + + @dbus_property(access=PropertyAccess.READ) + def UserspaceTimestampMonotonic(self) -> "t": + """Get UserspaceTimestampMonotonic.""" + return 4178239 + + @dbus_property(access=PropertyAccess.READ) + def FinishTimestamp(self) -> "t": + """Get FinishTimestamp.""" + return 1632236713344227 + + @dbus_property(access=PropertyAccess.READ) + def FinishTimestampMonotonic(self) -> "t": + """Get FinishTimestampMonotonic.""" + return 18374785 + + @dbus_property(access=PropertyAccess.READ) + def SecurityStartTimestamp(self) -> "t": + """Get SecurityStartTimestamp.""" + return 1632236699156494 + + @dbus_property(access=PropertyAccess.READ) + def SecurityStartTimestampMonotonic(self) -> "t": + """Get SecurityStartTimestampMonotonic.""" + return 4187052 + + @dbus_property(access=PropertyAccess.READ) + def SecurityFinishTimestamp(self) -> "t": + """Get SecurityFinishTimestamp.""" + return 1632236699156980 + + @dbus_property(access=PropertyAccess.READ) + def SecurityFinishTimestampMonotonic(self) -> "t": + """Get SecurityFinishTimestampMonotonic.""" + return 4187538 + + @dbus_property(access=PropertyAccess.READ) + def GeneratorsStartTimestamp(self) -> "t": + """Get GeneratorsStartTimestamp.""" + return 1632236699281427 + + @dbus_property(access=PropertyAccess.READ) + def GeneratorsStartTimestampMonotonic(self) -> "t": + """Get GeneratorsStartTimestampMonotonic.""" + return 4311984 + + @dbus_property(access=PropertyAccess.READ) + def GeneratorsFinishTimestamp(self) -> "t": + """Get GeneratorsFinishTimestamp.""" + return 1632236699334042 + + @dbus_property(access=PropertyAccess.READ) + def GeneratorsFinishTimestampMonotonic(self) -> "t": + """Get GeneratorsFinishTimestampMonotonic.""" + return 4364600 + + @dbus_property(access=PropertyAccess.READ) + def UnitsLoadStartTimestamp(self) -> "t": + """Get UnitsLoadStartTimestamp.""" + return 1632236699334044 + + @dbus_property(access=PropertyAccess.READ) + def UnitsLoadStartTimestampMonotonic(self) -> "t": + """Get UnitsLoadStartTimestampMonotonic.""" + return 4364602 + + @dbus_property(access=PropertyAccess.READ) + def UnitsLoadFinishTimestamp(self) -> "t": + """Get UnitsLoadFinishTimestamp.""" + return 1632236699424558 + + @dbus_property(access=PropertyAccess.READ) + def UnitsLoadFinishTimestampMonotonic(self) -> "t": + """Get UnitsLoadFinishTimestampMonotonic.""" + return 4455116 + + @dbus_property(access=PropertyAccess.READ) + def InitRDSecurityStartTimestamp(self) -> "t": + """Get InitRDSecurityStartTimestamp.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDSecurityStartTimestampMonotonic(self) -> "t": + """Get InitRDSecurityStartTimestampMonotonic.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDSecurityFinishTimestamp(self) -> "t": + """Get InitRDSecurityFinishTimestamp.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDSecurityFinishTimestampMonotonic(self) -> "t": + """Get InitRDSecurityFinishTimestampMonotonic.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDGeneratorsStartTimestamp(self) -> "t": + """Get InitRDGeneratorsStartTimestamp.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDGeneratorsStartTimestampMonotonic(self) -> "t": + """Get InitRDGeneratorsStartTimestampMonotonic.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDGeneratorsFinishTimestamp(self) -> "t": + """Get InitRDGeneratorsFinishTimestamp.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDGeneratorsFinishTimestampMonotonic(self) -> "t": + """Get InitRDGeneratorsFinishTimestampMonotonic.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDUnitsLoadStartTimestamp(self) -> "t": + """Get InitRDUnitsLoadStartTimestamp.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDUnitsLoadStartTimestampMonotonic(self) -> "t": + """Get InitRDUnitsLoadStartTimestampMonotonic.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDUnitsLoadFinishTimestamp(self) -> "t": + """Get InitRDUnitsLoadFinishTimestamp.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def InitRDUnitsLoadFinishTimestampMonotonic(self) -> "t": + """Get InitRDUnitsLoadFinishTimestampMonotonic.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def LogLevel(self) -> "s": + """Get LogLevel.""" + return "info" + + @dbus_property(access=PropertyAccess.READ) + def LogTarget(self) -> "s": + """Get LogTarget.""" + return "journal-or-kmsg" + + @dbus_property(access=PropertyAccess.READ) + def NNames(self) -> "u": + """Get NNames.""" + return 564 + + @dbus_property(access=PropertyAccess.READ) + def NFailedUnits(self) -> "u": + """Get NFailedUnits.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def NJobs(self) -> "u": + """Get NJobs.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def NInstalledJobs(self) -> "u": + """Get NInstalledJobs.""" + return 1575 + + @dbus_property(access=PropertyAccess.READ) + def NFailedJobs(self) -> "u": + """Get NFailedJobs.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def Progress(self) -> "d": + """Get Progress.""" + return 1.0 + + @dbus_property(access=PropertyAccess.READ) + def Environment(self) -> "as": + """Get Environment.""" + return [ + "LANG=en_US.UTF-8", + "LC_ADDRESS=nb_NO.UTF-8", + "LC_IDENTIFICATION=nb_NO.UTF-8", + "LC_MEASUREMENT=nb_NO.UTF-8", + "LC_MONETARY=nb_NO.UTF-8", + "LC_NAME=nb_NO.UTF-8", + "LC_NUMERIC=nb_NO.UTF-8", + "LC_PAPER=nb_NO.UTF-8", + "LC_TELEPHONE=nb_NO.UTF-8", + "LC_TIME=nb_NO.UTF-8", + "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/snap/bin", + ] + + @dbus_property(access=PropertyAccess.READ) + def ConfirmSpawn(self) -> "b": + """Get ConfirmSpawn.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def ShowStatus(self) -> "b": + """Get ShowStatus.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def UnitPath(self) -> "as": + """Get UnitPath.""" + return [ + "/etc/systemd/system.control", + "/run/systemd/system.control", + "/run/systemd/transient", + "/run/systemd/generator.early", + "/etc/systemd/system", + "/etc/systemd/system.attached", + "/run/systemd/system", + "/run/systemd/system.attached", + "/run/systemd/generator", + "/usr/local/lib/systemd/system", + "/lib/systemd/system", + "/usr/lib/systemd/system", + "/run/systemd/generator.late", + ] + + @dbus_property(access=PropertyAccess.READ) + def DefaultStandardOutput(self) -> "s": + """Get DefaultStandardOutput.""" + return "journal" + + @dbus_property(access=PropertyAccess.READ) + def DefaultStandardError(self) -> "s": + """Get DefaultStandardError.""" + return "journal" + + @dbus_property(access=PropertyAccess.READ) + def RuntimeWatchdogUSec(self) -> "t": + """Get RuntimeWatchdogUSec.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def RebootWatchdogUSec(self) -> "t": + """Get RebootWatchdogUSec.""" + return 600000000 + + @dbus_property(access=PropertyAccess.READ) + def KExecWatchdogUSec(self) -> "t": + """Get KExecWatchdogUSec.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def ServiceWatchdogs(self) -> "b": + """Get ServiceWatchdogs.""" + return True + + @dbus_property(access=PropertyAccess.READ) + def ControlGroup(self) -> "s": + """Get ControlGroup.""" + return "" + + @dbus_property(access=PropertyAccess.READ) + def SystemState(self) -> "s": + """Get SystemState.""" + return "running" + + @dbus_property(access=PropertyAccess.READ) + def ExitCode(self) -> "y": + """Get ExitCode.""" + return 0x00 + + @dbus_property(access=PropertyAccess.READ) + def DefaultTimerAccuracyUSec(self) -> "t": + """Get DefaultTimerAccuracyUSec.""" + return 60000000 + + @dbus_property(access=PropertyAccess.READ) + def DefaultTimeoutStartUSec(self) -> "t": + """Get DefaultTimeoutStartUSec.""" + return 90000000 + + @dbus_property(access=PropertyAccess.READ) + def DefaultTimeoutStopUSec(self) -> "t": + """Get DefaultTimeoutStopUSec.""" + return 90000000 + + @dbus_property(access=PropertyAccess.READ) + def DefaultTimeoutAbortUSec(self) -> "t": + """Get DefaultTimeoutAbortUSec.""" + return 90000000 + + @dbus_property(access=PropertyAccess.READ) + def DefaultRestartUSec(self) -> "t": + """Get DefaultRestartUSec.""" + return 100000 + + @dbus_property(access=PropertyAccess.READ) + def DefaultStartLimitIntervalUSec(self) -> "t": + """Get DefaultStartLimitIntervalUSec.""" + return 10000000 + + @dbus_property(access=PropertyAccess.READ) + def DefaultStartLimitBurst(self) -> "u": + """Get DefaultStartLimitBurst.""" + return 5 + + @dbus_property(access=PropertyAccess.READ) + def DefaultCPUAccounting(self) -> "b": + """Get DefaultCPUAccounting.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def DefaultBlockIOAccounting(self) -> "b": + """Get DefaultBlockIOAccounting.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def DefaultMemoryAccounting(self) -> "b": + """Get DefaultMemoryAccounting.""" + return True + + @dbus_property(access=PropertyAccess.READ) + def DefaultTasksAccounting(self) -> "b": + """Get DefaultTasksAccounting.""" + return True + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitCPU(self) -> "t": + """Get DefaultLimitCPU.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitCPUSoft(self) -> "t": + """Get DefaultLimitCPUSoft.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitFSIZE(self) -> "t": + """Get DefaultLimitFSIZE.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitFSIZESoft(self) -> "t": + """Get DefaultLimitFSIZESoft.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitDATA(self) -> "t": + """Get DefaultLimitDATA.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitDATASoft(self) -> "t": + """Get DefaultLimitDATASoft.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitSTACK(self) -> "t": + """Get DefaultLimitSTACK.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitSTACKSoft(self) -> "t": + """Get DefaultLimitSTACKSoft.""" + return 8388608 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitCORE(self) -> "t": + """Get DefaultLimitCORE.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitCORESoft(self) -> "t": + """Get DefaultLimitCORESoft.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitRSS(self) -> "t": + """Get DefaultLimitRSS.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitRSSSoft(self) -> "t": + """Get DefaultLimitRSSSoft.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitNOFILE(self) -> "t": + """Get DefaultLimitNOFILE.""" + return 524288 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitNOFILESoft(self) -> "t": + """Get DefaultLimitNOFILESoft.""" + return 1024 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitAS(self) -> "t": + """Get DefaultLimitAS.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitASSoft(self) -> "t": + """Get DefaultLimitASSoft.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitNPROC(self) -> "t": + """Get DefaultLimitNPROC.""" + return 127764 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitNPROCSoft(self) -> "t": + """Get DefaultLimitNPROCSoft.""" + return 127764 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitMEMLOCK(self) -> "t": + """Get DefaultLimitMEMLOCK.""" + return 65536 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitMEMLOCKSoft(self) -> "t": + """Get DefaultLimitMEMLOCKSoft.""" + return 65536 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitLOCKS(self) -> "t": + """Get DefaultLimitLOCKS.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitLOCKSSoft(self) -> "t": + """Get DefaultLimitLOCKSSoft.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitSIGPENDING(self) -> "t": + """Get DefaultLimitSIGPENDING.""" + return 127764 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitSIGPENDINGSoft(self) -> "t": + """Get DefaultLimitSIGPENDINGSoft.""" + return 127764 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitMSGQUEUE(self) -> "t": + """Get DefaultLimitMSGQUEUE.""" + return 819200 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitMSGQUEUESoft(self) -> "t": + """Get DefaultLimitMSGQUEUESoft.""" + return 819200 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitNICE(self) -> "t": + """Get DefaultLimitNICE.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitNICESoft(self) -> "t": + """Get DefaultLimitNICESoft.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitRTPRIO(self) -> "t": + """Get DefaultLimitRTPRIO.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitRTPRIOSoft(self) -> "t": + """Get DefaultLimitRTPRIOSoft.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitRTTIME(self) -> "t": + """Get DefaultLimitRTTIME.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultLimitRTTIMESoft(self) -> "t": + """Get DefaultLimitRTTIMESoft.""" + return 18446744073709551615 + + @dbus_property(access=PropertyAccess.READ) + def DefaultTasksMax(self) -> "t": + """Get DefaultTasksMax.""" + return 38329 + + @dbus_property(access=PropertyAccess.READ) + def TimerSlackNSec(self) -> "t": + """Get TimerSlackNSec.""" + return 50000 + + @dbus_property(access=PropertyAccess.READ) + def DefaultOOMPolicy(self) -> "s": + """Get DefaultOOMPolicy.""" + return "stop" + + @dbus_property(access=PropertyAccess.READ) + def DefaultOOMScoreAdjust(self) -> "i": + """Get DefaultOOMScoreAdjust.""" + return 0 + + @dbus_property(access=PropertyAccess.READ) + def CtrlAltDelBurstAction(self) -> "s": + """Get CtrlAltDelBurstAction.""" + return "reboot-force" + + @dbus_method() + def Reboot(self) -> None: + """Reboot host computer.""" + + @dbus_method() + def PowerOff(self) -> None: + """Power off host computer.""" + + @dbus_method() + def StartUnit(self, name: "s", mode: "s") -> "o": + """Start a service unit.""" + return "/org/freedesktop/systemd1/job/7623" + + @dbus_method() + def StopUnit(self, name: "s", mode: "s") -> "o": + """Stop a service unit.""" + return "/org/freedesktop/systemd1/job/7623" + + @dbus_method() + def ReloadOrRestartUnit(self, name: "s", mode: "s") -> "o": + """Reload or restart a service unit.""" + return "/org/freedesktop/systemd1/job/7623" + + @dbus_method() + def RestartUnit(self, name: "s", mode: "s") -> "o": + """Restart a service unit.""" + return "/org/freedesktop/systemd1/job/7623" + + @dbus_method() + def list_units( + self, + ) -> "a(ssssssouso)": + """Return a list of available services.""" + return [ + [ + "etc-machine\\x2did.mount", + "/etc/machine-id", + "loaded", + "active", + "mounted", + "", + "/org/freedesktop/systemd1/unit/etc_2dmachine_5cx2did_2emount", + 0, + "", + "/", + ], + [ + "firewalld.service", + "firewalld.service", + "not-found", + "inactive", + "dead", + "", + "/org/freedesktop/systemd1/unit/firewalld_2eservice", + 0, + "", + "/", + ], + [ + "sys-devices-virtual-tty-ttypd.device", + "/sys/devices/virtual/tty/ttypd", + "loaded", + "active", + "plugged", + "", + "/org/freedesktop/systemd1/unit/sys_2ddevices_2dvirtual_2dtty_2dttypd_2edevice", + 0, + "", + "/", + ], + [ + "zram-swap.service", + "HassOS ZRAM swap", + "loaded", + "active", + "exited", + "", + "/org/freedesktop/systemd1/unit/zram_2dswap_2eservice", + 0, + "", + "/", + ], + ] diff --git a/tests/dbus_service_mocks/timedate.py b/tests/dbus_service_mocks/timedate.py new file mode 100644 index 000000000..745b36f5d --- /dev/null +++ b/tests/dbus_service_mocks/timedate.py @@ -0,0 +1,95 @@ +"""Mock of timedate dbus service.""" + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.timedate1" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return TimeDate() + + +# pylint: disable=invalid-name + + +class TimeDate(DBusServiceMock): + """TimeDate mock. + + gdbus introspect --system --dest org.freedesktop.timedate1 --object-path /org/freedesktop/timedate1 + """ + + object_path = "/org/freedesktop/timedate1" + interface = "org.freedesktop.timedate1" + + @dbus_property(access=PropertyAccess.READ) + def Timezone(self) -> "s": + """Get Timezone.""" + return "Etc/UTC" + + @dbus_property(access=PropertyAccess.READ) + def LocalRTC(self) -> "b": + """Get LocalRTC.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def CanNTP(self) -> "b": + """Get CanNTP.""" + return True + + @dbus_property(access=PropertyAccess.READ) + def NTP(self) -> "b": + """Get NTP.""" + return True + + @dbus_property(access=PropertyAccess.READ) + def NTPSynchronized(self) -> "b": + """Get NTPSynchronized.""" + return True + + @dbus_property(access=PropertyAccess.READ) + def TimeUSec(self) -> "t": + """Get TimeUSec.""" + return 1621413414405718 + + @dbus_property(access=PropertyAccess.READ) + def RTCTimeUSec(self) -> "t": + """Get RTCTimeUSec.""" + return 1621413415000000 + + @dbus_method() + def SetTime(self, usec_utc: "x", relative: "b", interactive: "b") -> None: + """Set time.""" + + @dbus_method() + def SetTimezone(self, timezone: "s", interactive: "b") -> None: + """Set timezone.""" + self.emit_properties_changed({"Timezone": timezone}) + + @dbus_method() + def SetLocalRTC(self, local_rtc: "b", fix_system: "b", interactive: "b") -> None: + """Set local RTC.""" + self.emit_properties_changed({"LocalRTC": local_rtc}) + + @dbus_method() + def SetNTP(self, use_ntp: "b", interactive: "b") -> None: + """Set NTP.""" + self.emit_properties_changed({"NTP": use_ntp}) + + @dbus_method() + def ListTimezones(self) -> "as": + """List timezones.""" + return [ + "Africa/Abidjan", + "America/New_York", + "Antarctica/Casey", + "Asia/Hong_Kong", + "Atlantic/Azores", + "Australia/Sydney", + "Europe/Amsterdam", + "Indian/Chagos", + "Pacific/Apia", + "UTC", + ]