diff --git a/tests/dbus/agent/boards/test_board.py b/tests/dbus/agent/boards/test_board.py index 5a9b0295f..f78f8edbb 100644 --- a/tests/dbus/agent/boards/test_board.py +++ b/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" diff --git a/tests/dbus/agent/boards/test_yellow.py b/tests/dbus/agent/boards/test_yellow.py index b74e27a29..76915bf5e 100644 --- a/tests/dbus/agent/boards/test_yellow.py +++ b/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 diff --git a/tests/dbus/agent/conftest.py b/tests/dbus/agent/conftest.py new file mode 100644 index 000000000..08554c116 --- /dev/null +++ b/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, + ) diff --git a/tests/dbus/agent/test_agent.py b/tests/dbus/agent/test_agent.py index 4b892aab6..63c064d73 100644 --- a/tests/dbus/agent/test_agent.py +++ b/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 diff --git a/tests/dbus/agent/test_apparmor.py b/tests/dbus/agent/test_apparmor.py index fcc58357d..6e9352ac3 100644 --- a/tests/dbus/agent/test_apparmor.py +++ b/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") + ] diff --git a/tests/dbus/agent/test_cgroup.py b/tests/dbus/agent/test_cgroup.py index 30c643cbb..ae08b5977 100644 --- a/tests/dbus/agent/test_cgroup.py +++ b/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")] diff --git a/tests/dbus/agent/test_datadisk.py b/tests/dbus/agent/test_datadisk.py index 564a2fa15..2413fe0e0 100644 --- a/tests/dbus/agent/test_datadisk.py +++ b/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()] diff --git a/tests/dbus/agent/test_system.py b/tests/dbus/agent/test_system.py index 417298aed..c5a3d7bf4 100644 --- a/tests/dbus/agent/test_system.py +++ b/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()] diff --git a/tests/dbus_service_mocks/apparmor.py b/tests/dbus_service_mocks/apparmor.py new file mode 100644 index 000000000..650627cb2 --- /dev/null +++ b/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 diff --git a/tests/dbus_service_mocks/boards.py b/tests/dbus_service_mocks/boards.py new file mode 100644 index 000000000..8b30a9db5 --- /dev/null +++ b/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 diff --git a/tests/dbus_service_mocks/boards_supervised.py b/tests/dbus_service_mocks/boards_supervised.py new file mode 100644 index 000000000..bc698a8ca --- /dev/null +++ b/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" diff --git a/tests/dbus_service_mocks/boards_yellow.py b/tests/dbus_service_mocks/boards_yellow.py new file mode 100644 index 000000000..0274ddbfa --- /dev/null +++ b/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}) diff --git a/tests/dbus_service_mocks/cgroup.py b/tests/dbus_service_mocks/cgroup.py new file mode 100644 index 000000000..c8e86e58f --- /dev/null +++ b/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 diff --git a/tests/dbus_service_mocks/datadisk.py b/tests/dbus_service_mocks/datadisk.py new file mode 100644 index 000000000..32864473e --- /dev/null +++ b/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 diff --git a/tests/dbus_service_mocks/os_agent.py b/tests/dbus_service_mocks/os_agent.py new file mode 100644 index 000000000..80dd5e5eb --- /dev/null +++ b/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 diff --git a/tests/dbus_service_mocks/system.py b/tests/dbus_service_mocks/system.py new file mode 100644 index 000000000..7dec60ce5 --- /dev/null +++ b/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