Use session dbus for os agent tests (#4180)

tests/dbus/agent/boards/test_board.py

@@ -1,48 +1,27 @@
 """Test Boards manager."""
-
-from unittest.mock import patch
-
+# pylint: disable=import-error
 from dbus_fast.aio.message_bus import MessageBus
-from dbus_fast.aio.proxy_object import ProxyInterface
 import pytest
 
 from supervisor.dbus.agent.boards import BoardManager
 from supervisor.exceptions import BoardInvalidError
-from supervisor.utils.dbus import DBUS_INTERFACE_PROPERTIES, DBus
+
+from tests.common import mock_dbus_services
+from tests.dbus_service_mocks.boards import Boards as BoardsService
 
 
-@pytest.fixture(name="dbus_mock_board")
-async def fixture_dbus_mock_board(request: pytest.FixtureRequest, dbus: list[str]):
-    """Mock Boards dbus object to particular board name for tests."""
-    call_dbus = DBus.call_dbus
-
-    async def mock_call_dbus_specify_board(
-        proxy_interface: ProxyInterface,
-        method: str,
-        *args,
-        unpack_variants: bool = True,
-    ):
-        if (
-            proxy_interface.introspection.name == DBUS_INTERFACE_PROPERTIES
-            and method == "call_get_all"
-            and proxy_interface.path == "/io/hass/os/Boards"
-        ):
-            return {"Board": request.param}
-
-        return call_dbus(
-            proxy_interface, method, *args, unpack_variants=unpack_variants
-        )
-
-    with patch(
-        "supervisor.utils.dbus.DBus.call_dbus", new=mock_call_dbus_specify_board
-    ):
-        yield dbus
+@pytest.fixture(name="boards_service", autouse=True)
+async def fixture_boards_service(dbus_session_bus: MessageBus) -> BoardsService:
+    """Mock Boards dbus service."""
+    yield (await mock_dbus_services({"boards": None}, dbus_session_bus))["boards"]
 
 
-async def test_dbus_board(dbus: list[str], dbus_bus: MessageBus):
+async def test_dbus_board(dbus_session_bus: MessageBus):
     """Test DBus Board load."""
+    await mock_dbus_services({"boards_yellow": None}, dbus_session_bus)
+
     board = BoardManager()
-    await board.connect(dbus_bus)
+    await board.connect(dbus_session_bus)
 
     assert board.board == "Yellow"
     assert board.yellow.power_led is True
@@ -51,11 +30,15 @@ async def test_dbus_board(dbus: list[str], dbus_bus: MessageBus):
         assert not board.supervised
 
 
-@pytest.mark.parametrize("dbus_mock_board", ["Supervised"], indirect=True)
-async def test_dbus_board_supervised(dbus_mock_board: list[str], dbus_bus: MessageBus):
+async def test_dbus_board_supervised(
+    boards_service: BoardsService, dbus_session_bus: MessageBus
+):
     """Test DBus Board load with supervised board."""
+    await mock_dbus_services({"boards_supervised": None}, dbus_session_bus)
+    boards_service.board = "Supervised"
+
     board = BoardManager()
-    await board.connect(dbus_bus)
+    await board.connect(dbus_session_bus)
 
     assert board.board == "Supervised"
     assert board.supervised
@@ -64,11 +47,14 @@ async def test_dbus_board_supervised(dbus_mock_board: list[str], dbus_bus: Messa
         assert not board.yellow
 
 
-@pytest.mark.parametrize("dbus_mock_board", ["NotReal"], indirect=True)
-async def test_dbus_board_other(dbus_mock_board: list[str], dbus_bus: MessageBus):
+async def test_dbus_board_other(
+    boards_service: BoardsService, dbus_session_bus: MessageBus
+):
     """Test DBus Board load with board that has no dbus object."""
+    boards_service.board = "NotReal"
+
     board = BoardManager()
-    await board.connect(dbus_bus)
+    await board.connect(dbus_session_bus)
 
     assert board.board == "NotReal"
 

tests/dbus/agent/boards/test_yellow.py

@@ -1,16 +1,28 @@
 """Test Yellow board."""
-
+# pylint: disable=import-error
 import asyncio
 
 from dbus_fast.aio.message_bus import MessageBus
+import pytest
 
 from supervisor.dbus.agent.boards.yellow import Yellow
 
+from tests.common import mock_dbus_services
+from tests.dbus_service_mocks.boards_yellow import Yellow as YellowService
 
-async def test_dbus_yellow(dbus: list[str], dbus_bus: MessageBus):
+
+@pytest.fixture(name="yellow_service", autouse=True)
+async def fixture_yellow_service(dbus_session_bus: MessageBus) -> YellowService:
+    """Mock Yellow Board dbus service."""
+    yield (await mock_dbus_services({"boards_yellow": None}, dbus_session_bus))[
+        "boards_yellow"
+    ]
+
+
+async def test_dbus_yellow(dbus_session_bus: MessageBus):
     """Test Yellow board load."""
     yellow = Yellow()
-    await yellow.connect(dbus_bus)
+    await yellow.connect(dbus_session_bus)
 
     assert yellow.name == "Yellow"
     assert yellow.disk_led is True
@@ -18,37 +30,40 @@ async def test_dbus_yellow(dbus: list[str], dbus_bus: MessageBus):
     assert yellow.power_led is True
 
 
-async def test_dbus_yellow_set_disk_led(dbus: list[str], dbus_bus: MessageBus):
+async def test_dbus_yellow_set_disk_led(
+    yellow_service: YellowService, dbus_session_bus: MessageBus
+):
     """Test setting disk led for Yellow board."""
     yellow = Yellow()
-    await yellow.connect(dbus_bus)
+    await yellow.connect(dbus_session_bus)
 
-    dbus.clear()
     yellow.disk_led = False
-    await asyncio.sleep(0)
-
-    assert dbus == ["/io/hass/os/Boards/Yellow-io.hass.os.Boards.Yellow.DiskLED"]
+    await asyncio.sleep(0)  # Set property via dbus is separate async task
+    await yellow_service.ping()
+    assert yellow.disk_led is False
 
 
-async def test_dbus_yellow_set_heartbeat_led(dbus: list[str], dbus_bus: MessageBus):
+async def test_dbus_yellow_set_heartbeat_led(
+    yellow_service: YellowService, dbus_session_bus: MessageBus
+):
     """Test setting heartbeat led for Yellow board."""
     yellow = Yellow()
-    await yellow.connect(dbus_bus)
+    await yellow.connect(dbus_session_bus)
 
-    dbus.clear()
     yellow.heartbeat_led = False
-    await asyncio.sleep(0)
-
-    assert dbus == ["/io/hass/os/Boards/Yellow-io.hass.os.Boards.Yellow.HeartbeatLED"]
+    await asyncio.sleep(0)  # Set property via dbus is separate async task
+    await yellow_service.ping()
+    assert yellow.heartbeat_led is False
 
 
-async def test_dbus_yellow_set_power_led(dbus: list[str], dbus_bus: MessageBus):
+async def test_dbus_yellow_set_power_led(
+    yellow_service: YellowService, dbus_session_bus: MessageBus
+):
     """Test setting power led for Yellow board."""
     yellow = Yellow()
-    await yellow.connect(dbus_bus)
+    await yellow.connect(dbus_session_bus)
 
-    dbus.clear()
     yellow.power_led = False
-    await asyncio.sleep(0)
-
-    assert dbus == ["/io/hass/os/Boards/Yellow-io.hass.os.Boards.Yellow.PowerLED"]
+    await asyncio.sleep(0)  # Set property via dbus is separate async task
+    await yellow_service.ping()
+    assert yellow.power_led is False

tests/dbus/agent/conftest.py

@@ -0,0 +1,24 @@
+"""Shared fixtures for OS Agent tests."""
+
+from dbus_fast.aio.message_bus import MessageBus
+import pytest
+
+from tests.common import mock_dbus_services
+from tests.dbus_service_mocks.base import DBusServiceMock
+
+
+@pytest.fixture
+async def os_agent_services(dbus_session_bus: MessageBus) -> dict[str, DBusServiceMock]:
+    """Mock all services os agent connects to."""
+    yield await mock_dbus_services(
+        {
+            "os_agent": None,
+            "apparmor": None,
+            "cgroup": None,
+            "datadisk": None,
+            "system": None,
+            "boards": None,
+            "boards_yellow": None,
+        },
+        dbus_session_bus,
+    )

tests/dbus/agent/test_agent.py

@@ -1,27 +1,41 @@
 """Test OSAgent dbus interface."""
+# pylint: disable=import-error
+from dbus_fast.aio.message_bus import MessageBus
+import pytest
 
-import asyncio
+from supervisor.dbus.agent import OSAgent
 
-from supervisor.coresys import CoreSys
-
-from tests.common import fire_property_change_signal
+from tests.dbus_service_mocks.base import DBusServiceMock
+from tests.dbus_service_mocks.os_agent import OSAgent as OSAgentService
 
 
-async def test_dbus_osagent(coresys: CoreSys):
-    """Test coresys dbus connection."""
-    assert coresys.dbus.agent.version is None
-    assert coresys.dbus.agent.diagnostics is None
+@pytest.fixture(name="os_agent_service", autouse=True)
+async def fixture_os_agent_service(
+    os_agent_services: dict[str, DBusServiceMock]
+) -> OSAgentService:
+    """Mock OS Agent dbus service."""
+    yield os_agent_services["os_agent"]
 
-    await coresys.dbus.agent.connect(coresys.dbus.bus)
-    await coresys.dbus.agent.update()
 
-    assert coresys.dbus.agent.version == "1.1.0"
-    assert coresys.dbus.agent.diagnostics
+async def test_dbus_osagent(
+    os_agent_service: OSAgentService, dbus_session_bus: MessageBus
+):
+    """Test OS Agent properties."""
+    os_agent = OSAgent()
 
-    fire_property_change_signal(coresys.dbus.agent, {"Diagnostics": False})
-    await asyncio.sleep(0)
-    assert coresys.dbus.agent.diagnostics is False
+    assert os_agent.version is None
+    assert os_agent.diagnostics is None
 
-    fire_property_change_signal(coresys.dbus.agent, {}, ["Diagnostics"])
-    await asyncio.sleep(0)
-    assert coresys.dbus.agent.diagnostics is True
+    await os_agent.connect(dbus_session_bus)
+
+    assert os_agent.version == "1.1.0"
+    assert os_agent.diagnostics
+
+    os_agent_service.emit_properties_changed({"Diagnostics": False})
+    await os_agent_service.ping()
+    assert os_agent.diagnostics is False
+
+    os_agent_service.emit_properties_changed({}, ["Diagnostics"])
+    await os_agent_service.ping()
+    await os_agent_service.ping()
+    assert os_agent.diagnostics is True

tests/dbus/agent/test_apparmor.py

@@ -1,66 +1,90 @@
 """Test AppArmor/Agent dbus interface."""
-import asyncio
+# pylint: disable=import-error
 from pathlib import Path
 
+from dbus_fast.aio.message_bus import MessageBus
 import pytest
 
-from supervisor.coresys import CoreSys
+from supervisor.dbus.agent import OSAgent
 from supervisor.exceptions import DBusNotConnectedError
 
-from tests.common import fire_property_change_signal
+from tests.dbus_service_mocks.apparmor import AppArmor as AppArmorService
+from tests.dbus_service_mocks.base import DBusServiceMock
 
 
-async def test_dbus_osagent_apparmor(coresys: CoreSys):
-    """Test coresys dbus connection."""
-    assert coresys.dbus.agent.apparmor.version is None
-
-    await coresys.dbus.agent.connect(coresys.dbus.bus)
-    await coresys.dbus.agent.update()
-
-    assert coresys.dbus.agent.apparmor.version == "2.13.2"
-
-    fire_property_change_signal(coresys.dbus.agent.apparmor, {"ParserVersion": "1.0.0"})
-    await asyncio.sleep(0)
-    assert coresys.dbus.agent.apparmor.version == "1.0.0"
-
-    fire_property_change_signal(coresys.dbus.agent.apparmor, {}, ["ParserVersion"])
-    await asyncio.sleep(0)
-    assert coresys.dbus.agent.apparmor.version == "2.13.2"
+@pytest.fixture(name="apparmor_service", autouse=True)
+async def fixture_apparmor_service(
+    os_agent_services: dict[str, DBusServiceMock]
+) -> AppArmorService:
+    """Mock AppArmor dbus service."""
+    yield os_agent_services["apparmor"]
 
 
-async def test_dbus_osagent_apparmor_load(coresys: CoreSys, dbus: list[str]):
+async def test_dbus_osagent_apparmor(
+    apparmor_service: AppArmorService, dbus_session_bus: MessageBus
+):
+    """Test AppArmor properties."""
+    os_agent = OSAgent()
+
+    assert os_agent.apparmor.version is None
+
+    await os_agent.connect(dbus_session_bus)
+
+    assert os_agent.apparmor.version == "2.13.2"
+
+    apparmor_service.emit_properties_changed({"ParserVersion": "1.0.0"})
+    await apparmor_service.ping()
+    assert os_agent.apparmor.version == "1.0.0"
+
+    apparmor_service.emit_properties_changed({}, ["ParserVersion"])
+    await apparmor_service.ping()
+    await apparmor_service.ping()
+    assert os_agent.apparmor.version == "2.13.2"
+
+
+async def test_dbus_osagent_apparmor_load(
+    apparmor_service: AppArmorService, dbus_session_bus: MessageBus
+):
     """Load AppArmor Profile on host."""
+    os_agent = OSAgent()
+
     with pytest.raises(DBusNotConnectedError):
-        await coresys.dbus.agent.apparmor.load_profile(
+        await os_agent.apparmor.load_profile(
             Path("/data/apparmor/profile"), Path("/data/apparmor/cache")
         )
 
-    await coresys.dbus.agent.connect(coresys.dbus.bus)
+    await os_agent.connect(dbus_session_bus)
 
-    dbus.clear()
     assert (
-        await coresys.dbus.agent.apparmor.load_profile(
+        await os_agent.apparmor.load_profile(
             Path("/data/apparmor/profile"), Path("/data/apparmor/cache")
         )
         is None
     )
-    assert dbus == ["/io/hass/os/AppArmor-io.hass.os.AppArmor.LoadProfile"]
+    assert apparmor_service.LoadProfile.calls == [
+        ("/data/apparmor/profile", "/data/apparmor/cache")
+    ]
 
 
-async def test_dbus_osagent_apparmor_unload(coresys: CoreSys, dbus: list[str]):
+async def test_dbus_osagent_apparmor_unload(
+    apparmor_service: AppArmorService, dbus_session_bus: MessageBus
+):
     """Unload AppArmor Profile on host."""
+    os_agent = OSAgent()
+
     with pytest.raises(DBusNotConnectedError):
-        await coresys.dbus.agent.apparmor.unload_profile(
+        await os_agent.apparmor.unload_profile(
             Path("/data/apparmor/profile"), Path("/data/apparmor/cache")
         )
 
-    await coresys.dbus.agent.connect(coresys.dbus.bus)
+    await os_agent.connect(dbus_session_bus)
 
-    dbus.clear()
     assert (
-        await coresys.dbus.agent.apparmor.unload_profile(
+        await os_agent.apparmor.unload_profile(
             Path("/data/apparmor/profile"), Path("/data/apparmor/cache")
         )
         is None
     )
-    assert dbus == ["/io/hass/os/AppArmor-io.hass.os.AppArmor.UnloadProfile"]
+    assert apparmor_service.UnloadProfile.calls == [
+        ("/data/apparmor/profile", "/data/apparmor/cache")
+    ]

tests/dbus/agent/test_cgroup.py

@@ -1,21 +1,33 @@
 """Test CGroup/Agent dbus interface."""
-
+# pylint: disable=import-error
+from dbus_fast.aio.message_bus import MessageBus
 import pytest
 
-from supervisor.coresys import CoreSys
+from supervisor.dbus.agent import OSAgent
 from supervisor.exceptions import DBusNotConnectedError
 
+from tests.dbus_service_mocks.base import DBusServiceMock
+from tests.dbus_service_mocks.cgroup import CGroup as CGroupService
 
-async def test_dbus_osagent_cgroup_add_devices(coresys: CoreSys, dbus: list[str]):
+
+@pytest.fixture(name="cgroup_service", autouse=True)
+async def fixture_cgroup_service(
+    os_agent_services: dict[str, DBusServiceMock]
+) -> CGroupService:
+    """Mock CGroup dbus service."""
+    yield os_agent_services["cgroup"]
+
+
+async def test_dbus_osagent_cgroup_add_devices(
+    cgroup_service: CGroupService, dbus_session_bus: MessageBus
+):
     """Test wipe data partition on host."""
+    os_agent = OSAgent()
+
     with pytest.raises(DBusNotConnectedError):
-        await coresys.dbus.agent.cgroup.add_devices_allowed("9324kl23j4kl", "*:* rwm")
+        await os_agent.cgroup.add_devices_allowed("9324kl23j4kl", "*:* rwm")
 
-    await coresys.dbus.agent.connect(coresys.dbus.bus)
+    await os_agent.connect(dbus_session_bus)
 
-    dbus.clear()
-    assert (
-        await coresys.dbus.agent.cgroup.add_devices_allowed("9324kl23j4kl", "*:* rwm")
-        is None
-    )
-    assert dbus == ["/io/hass/os/CGroup-io.hass.os.CGroup.AddDevicesAllowed"]
+    assert await os_agent.cgroup.add_devices_allowed("9324kl23j4kl", "*:* rwm") is None
+    assert cgroup_service.AddDevicesAllowed.calls == [("9324kl23j4kl", "*:* rwm")]

tests/dbus/agent/test_datadisk.py

@@ -1,54 +1,72 @@
 """Test Datadisk/Agent dbus interface."""
-import asyncio
+# pylint: disable=import-error
 from pathlib import Path
 
+from dbus_fast.aio.message_bus import MessageBus
 import pytest
 
-from supervisor.coresys import CoreSys
+from supervisor.dbus.agent import OSAgent
 from supervisor.exceptions import DBusNotConnectedError
 
-from tests.common import fire_property_change_signal
+from tests.dbus_service_mocks.base import DBusServiceMock
+from tests.dbus_service_mocks.datadisk import DataDisk as DataDiskService
 
 
-async def test_dbus_osagent_datadisk(coresys: CoreSys):
-    """Test coresys dbus connection."""
-    assert coresys.dbus.agent.datadisk.current_device is None
-
-    await coresys.dbus.agent.connect(coresys.dbus.bus)
-    await coresys.dbus.agent.update()
-
-    assert coresys.dbus.agent.datadisk.current_device.as_posix() == "/dev/sda"
-
-    fire_property_change_signal(
-        coresys.dbus.agent.datadisk, {"CurrentDevice": "/dev/sda1"}
-    )
-    await asyncio.sleep(0)
-    assert coresys.dbus.agent.datadisk.current_device.as_posix() == "/dev/sda1"
-
-    fire_property_change_signal(coresys.dbus.agent.datadisk, {}, ["CurrentDevice"])
-    await asyncio.sleep(0)
-    assert coresys.dbus.agent.datadisk.current_device.as_posix() == "/dev/sda"
+@pytest.fixture(name="datadisk_service", autouse=True)
+async def fixture_datadisk_service(
+    os_agent_services: dict[str, DBusServiceMock]
+) -> DataDiskService:
+    """Mock DataDisk dbus service."""
+    yield os_agent_services["datadisk"]
 
 
-async def test_dbus_osagent_datadisk_change_device(coresys: CoreSys, dbus: list[str]):
+async def test_dbus_osagent_datadisk(
+    datadisk_service: DataDiskService, dbus_session_bus: MessageBus
+):
+    """Test OS-Agent datadisk properties."""
+    os_agent = OSAgent()
+
+    assert os_agent.datadisk.current_device is None
+
+    await os_agent.connect(dbus_session_bus)
+
+    assert os_agent.datadisk.current_device.as_posix() == "/dev/sda"
+
+    datadisk_service.emit_properties_changed({"CurrentDevice": "/dev/sda1"})
+    await datadisk_service.ping()
+    assert os_agent.datadisk.current_device.as_posix() == "/dev/sda1"
+
+    datadisk_service.emit_properties_changed({}, ["CurrentDevice"])
+    await datadisk_service.ping()
+    await datadisk_service.ping()
+    assert os_agent.datadisk.current_device.as_posix() == "/dev/sda"
+
+
+async def test_dbus_osagent_datadisk_change_device(
+    datadisk_service: DataDiskService, dbus_session_bus: MessageBus
+):
     """Change datadisk on device."""
+    os_agent = OSAgent()
+
     with pytest.raises(DBusNotConnectedError):
-        await coresys.dbus.agent.datadisk.change_device(Path("/dev/sdb"))
+        await os_agent.datadisk.change_device(Path("/dev/sdb"))
 
-    await coresys.dbus.agent.connect(coresys.dbus.bus)
+    await os_agent.connect(dbus_session_bus)
 
-    dbus.clear()
-    assert await coresys.dbus.agent.datadisk.change_device(Path("/dev/sdb")) is None
-    assert dbus == ["/io/hass/os/DataDisk-io.hass.os.DataDisk.ChangeDevice"]
+    assert await os_agent.datadisk.change_device(Path("/dev/sdb")) is None
+    assert datadisk_service.ChangeDevice.calls == [("/dev/sdb",)]
 
 
-async def test_dbus_osagent_datadisk_reload_device(coresys: CoreSys, dbus: list[str]):
+async def test_dbus_osagent_datadisk_reload_device(
+    datadisk_service: DataDiskService, dbus_session_bus: MessageBus
+):
     """Change datadisk on device."""
+    os_agent = OSAgent()
+
     with pytest.raises(DBusNotConnectedError):
-        await coresys.dbus.agent.datadisk.reload_device()
+        await os_agent.datadisk.reload_device()
 
-    await coresys.dbus.agent.connect(coresys.dbus.bus)
+    await os_agent.connect(dbus_session_bus)
 
-    dbus.clear()
-    assert await coresys.dbus.agent.datadisk.reload_device() is None
-    assert dbus == ["/io/hass/os/DataDisk-io.hass.os.DataDisk.ReloadDevice"]
+    assert await os_agent.datadisk.reload_device() is None
+    assert datadisk_service.ReloadDevice.calls == [tuple()]

tests/dbus/agent/test_system.py

@@ -1,18 +1,33 @@
 """Test System/Agent dbus interface."""
-
+# pylint: disable=import-error
+from dbus_fast.aio.message_bus import MessageBus
 import pytest
 
-from supervisor.coresys import CoreSys
+from supervisor.dbus.agent import OSAgent
 from supervisor.exceptions import DBusNotConnectedError
 
+from tests.dbus_service_mocks.base import DBusServiceMock
+from tests.dbus_service_mocks.system import System as SystemService
 
-async def test_dbus_osagent_system_wipe(coresys: CoreSys, dbus: list[str]):
+
+@pytest.fixture(name="system_service", autouse=True)
+async def fixture_system_service(
+    os_agent_services: dict[str, DBusServiceMock]
+) -> SystemService:
+    """Mock System dbus service."""
+    yield os_agent_services["system"]
+
+
+async def test_dbus_osagent_system_wipe(
+    system_service: SystemService, dbus_session_bus: MessageBus
+):
     """Test wipe data partition on host."""
+    os_agent = OSAgent()
+
     with pytest.raises(DBusNotConnectedError):
-        await coresys.dbus.agent.system.schedule_wipe_device()
+        await os_agent.system.schedule_wipe_device()
 
-    await coresys.dbus.agent.connect(coresys.dbus.bus)
+    await os_agent.connect(dbus_session_bus)
 
-    dbus.clear()
-    assert await coresys.dbus.agent.system.schedule_wipe_device() is None
-    assert dbus == ["/io/hass/os/System-io.hass.os.System.ScheduleWipeDevice"]
+    assert await os_agent.system.schedule_wipe_device() is None
+    assert system_service.ScheduleWipeDevice.calls == [tuple()]

tests/dbus_service_mocks/apparmor.py

@@ -0,0 +1,40 @@
+"""Mock of OS Agent AppArmor dbus service."""
+
+from dbus_fast.service import PropertyAccess, dbus_property
+
+from .base import DBusServiceMock, dbus_method
+
+BUS_NAME = "io.hass.os"
+
+
+def setup(object_path: str | None = None) -> DBusServiceMock:
+    """Create dbus mock object."""
+    return AppArmor()
+
+
+# pylint: disable=invalid-name
+
+
+class AppArmor(DBusServiceMock):
+    """AppArmor mock.
+
+    gdbus introspect --system --dest io.hass.os --object-path /io/hass/os/AppArmor
+    """
+
+    object_path = "/io/hass/os/AppArmor"
+    interface = "io.hass.os.AppArmor"
+
+    @dbus_property(access=PropertyAccess.READ)
+    def ParserVersion(self) -> "s":
+        """Get ParserVersion."""
+        return "2.13.2"
+
+    @dbus_method()
+    def LoadProfile(self, arg_0: "s", arg_1: "s") -> "b":
+        """Load profile."""
+        return True
+
+    @dbus_method()
+    def UnloadProfile(self, arg_0: "s", arg_1: "s") -> "b":
+        """Unload profile."""
+        return True

tests/dbus_service_mocks/boards.py

@@ -0,0 +1,31 @@
+"""Mock of OS Agent Boards dbus service."""
+
+from dbus_fast.service import PropertyAccess, dbus_property
+
+from .base import DBusServiceMock
+
+BUS_NAME = "io.hass.os"
+
+
+def setup(object_path: str | None = None) -> DBusServiceMock:
+    """Create dbus mock object."""
+    return Boards()
+
+
+# pylint: disable=invalid-name
+
+
+class Boards(DBusServiceMock):
+    """Boards mock.
+
+    gdbus introspect --system --dest io.hass.os --object-path /io/hass/os/Boards
+    """
+
+    object_path = "/io/hass/os/Boards"
+    interface = "io.hass.os.Boards"
+    board = "Yellow"
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Board(self) -> "s":
+        """Get Board."""
+        return self.board

tests/dbus_service_mocks/boards_supervised.py

@@ -0,0 +1,23 @@
+"""Mock of OS Agent Boards Supervised dbus service."""
+
+from .base import DBusServiceMock
+
+BUS_NAME = "io.hass.os"
+
+
+def setup(object_path: str | None = None) -> DBusServiceMock:
+    """Create dbus mock object."""
+    return Supervised()
+
+
+# pylint: disable=invalid-name
+
+
+class Supervised(DBusServiceMock):
+    """Supervised mock.
+
+    gdbus introspect --system --dest io.hass.os --object-path /io/hass/os/Boards/Supervised
+    """
+
+    object_path = "/io/hass/os/Boards/Supervised"
+    interface = "io.hass.os.Boards.Supervised"

tests/dbus_service_mocks/boards_yellow.py

@@ -0,0 +1,55 @@
+"""Mock of OS Agent Boards Yellow dbus service."""
+
+from dbus_fast.service import dbus_property
+
+from .base import DBusServiceMock
+
+BUS_NAME = "io.hass.os"
+
+
+def setup(object_path: str | None = None) -> DBusServiceMock:
+    """Create dbus mock object."""
+    return Yellow()
+
+
+# pylint: disable=invalid-name
+
+
+class Yellow(DBusServiceMock):
+    """Yellow mock.
+
+    gdbus introspect --system --dest io.hass.os --object-path /io/hass/os/Boards/Yellow
+    """
+
+    object_path = "/io/hass/os/Boards/Yellow"
+    interface = "io.hass.os.Boards.Yellow"
+
+    @dbus_property()
+    def HeartbeatLED(self) -> "b":
+        """Get Heartbeat LED."""
+        return True
+
+    @HeartbeatLED.setter
+    def HeartbeatLED(self, value: "b"):
+        """Set Heartbeat LED."""
+        self.emit_properties_changed({"HeartbeatLED": value})
+
+    @dbus_property()
+    def PowerLED(self) -> "b":
+        """Get Power LED."""
+        return True
+
+    @PowerLED.setter
+    def PowerLED(self, value: "b"):
+        """Set Power LED."""
+        self.emit_properties_changed({"PowerLED": value})
+
+    @dbus_property()
+    def DiskLED(self) -> "b":
+        """Get Disk LED."""
+        return True
+
+    @DiskLED.setter
+    def DiskLED(self, value: "b"):
+        """Set Disk LED."""
+        self.emit_properties_changed({"DiskLED": value})

tests/dbus_service_mocks/cgroup.py

@@ -0,0 +1,28 @@
+"""Mock of OS Agent CGroup dbus service."""
+
+from .base import DBusServiceMock, dbus_method
+
+BUS_NAME = "io.hass.os"
+
+
+def setup(object_path: str | None = None) -> DBusServiceMock:
+    """Create dbus mock object."""
+    return CGroup()
+
+
+# pylint: disable=invalid-name
+
+
+class CGroup(DBusServiceMock):
+    """CGroup mock.
+
+    gdbus introspect --system --dest io.hass.os --object-path /io/hass/os/CGroup
+    """
+
+    object_path = "/io/hass/os/CGroup"
+    interface = "io.hass.os.CGroup"
+
+    @dbus_method()
+    def AddDevicesAllowed(self, arg_0: "s", arg_1: "s") -> "b":
+        """Load profile."""
+        return True

tests/dbus_service_mocks/datadisk.py

@@ -0,0 +1,40 @@
+"""Mock of OS Agent DataDisk dbus service."""
+
+from dbus_fast.service import PropertyAccess, dbus_property
+
+from .base import DBusServiceMock, dbus_method
+
+BUS_NAME = "io.hass.os"
+
+
+def setup(object_path: str | None = None) -> DBusServiceMock:
+    """Create dbus mock object."""
+    return DataDisk()
+
+
+# pylint: disable=invalid-name
+
+
+class DataDisk(DBusServiceMock):
+    """DataDisk mock.
+
+    gdbus introspect --system --dest io.hass.os --object-path /io/hass/os/DataDisk
+    """
+
+    object_path = "/io/hass/os/DataDisk"
+    interface = "io.hass.os.DataDisk"
+
+    @dbus_property(access=PropertyAccess.READ)
+    def CurrentDevice(self) -> "s":
+        """Get Current Device."""
+        return "/dev/sda"
+
+    @dbus_method()
+    def ChangeDevice(self, arg_0: "s") -> "b":
+        """Change device."""
+        return True
+
+    @dbus_method()
+    def ReloadDevice(self) -> "b":
+        """Reload device."""
+        return True

tests/dbus_service_mocks/os_agent.py

@@ -0,0 +1,35 @@
+"""Mock of os agent dbus service."""
+
+from dbus_fast.service import PropertyAccess, dbus_property
+
+from .base import DBusServiceMock
+
+BUS_NAME = "io.hass.os"
+
+
+def setup(object_path: str | None = None) -> DBusServiceMock:
+    """Create dbus mock object."""
+    return OSAgent()
+
+
+# pylint: disable=invalid-name
+
+
+class OSAgent(DBusServiceMock):
+    """OS-agent mock.
+
+    gdbus introspect --system --dest io.hass.os --object-path /io/hass/os
+    """
+
+    object_path = "/io/hass/os"
+    interface = "io.hass.os"
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Version(self) -> "s":
+        """Get Version."""
+        return "1.1.0"
+
+    @dbus_property(access=PropertyAccess.READ)
+    def Diagnostics(self) -> "b":
+        """Get Diagnostics."""
+        return True

tests/dbus_service_mocks/system.py

@@ -0,0 +1,33 @@
+"""Mock of OS Agent System dbus service."""
+
+from .base import DBusServiceMock, dbus_method
+
+BUS_NAME = "io.hass.os"
+
+
+def setup(object_path: str | None = None) -> DBusServiceMock:
+    """Create dbus mock object."""
+    return System()
+
+
+# pylint: disable=invalid-name
+
+
+class System(DBusServiceMock):
+    """System mock.
+
+    gdbus introspect --system --dest io.hass.os --object-path /io/hass/os/System
+    """
+
+    object_path = "/io/hass/os/System"
+    interface = "io.hass.os.System"
+
+    @dbus_method()
+    def ScheduleWipeDevice(self) -> "b":
+        """Schedule wipe device."""
+        return True
+
+    @dbus_method()
+    def WipeDevice(self) -> "b":
+        """Wipe device."""
+        return True