diff --git a/tests/common.py b/tests/common.py index fede2ebb7..182b06143 100644 --- a/tests/common.py +++ b/tests/common.py @@ -108,12 +108,17 @@ def exists_fixture(filename: str) -> bool: async def mock_dbus_services( to_mock: dict[str, list[str] | str | None], bus: MessageBus -) -> dict[str, list[DBusServiceMock] | DBusServiceMock]: +) -> dict[str, dict[str, 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). + + A dictionary is returned where the key is the dbus service to mock and the value + is the instance of the mocked service. If a list of object paths is provided, + the value is a dictionary where the key is the object path and value is the + mocked instance of the service for that object path. """ services: dict[str, list[DBusServiceMock] | DBusServiceMock] = {} requested_names: set[str] = set() @@ -127,10 +132,10 @@ async def mock_dbus_services( requested_names.add(service_module.BUS_NAME) if isinstance(to_mock[module], list): - services[module] = [ - service_module.setup(obj_path).export(bus) + services[module] = { + obj_path: 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) diff --git a/tests/conftest.py b/tests/conftest.py index 32da58118..319202a95 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -147,12 +147,11 @@ async def dbus_session_bus(dbus_session) -> MessageBus: @pytest.fixture async def dbus_services( request: pytest.FixtureRequest, dbus_session: MessageBus -) -> dict[str, DBusServiceMock | list[DBusServiceMock]]: +) -> dict[str, dict[str, DBusServiceMock] | 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). + Wrapper on mock_dbus_services intended to be used indirectly. + Request param passed to it as to_mock and output returned as fixture value. """ with patch("supervisor.dbus.manager.MessageBus.connect", return_value=dbus_session): yield await mock_dbus_services(request.param, dbus_session) @@ -330,7 +329,7 @@ async def dbus_is_connected(): @pytest.fixture(name="network_manager_services") async def fixture_network_manager_services( dbus_session_bus: MessageBus, -) -> dict[str, DBusServiceMock]: +) -> dict[str, DBusServiceMock | dict[str, DBusServiceMock]]: """Mock all services network manager connects to.""" yield await mock_dbus_services( { @@ -357,14 +356,65 @@ async def fixture_network_manager_services( @pytest.fixture async def network_manager( - network_manager_services: dict[str, DBusServiceMock], dbus_session_bus: MessageBus + network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]], + dbus_session_bus: MessageBus, ) -> NetworkManager: - """Mock NetworkManager.""" + """Mock Network Manager.""" nm_obj = NetworkManager() await nm_obj.connect(dbus_session_bus) yield nm_obj +@pytest.fixture(name="udisks2_services") +async def fixture_udisks2_services( + dbus_session_bus: MessageBus, +) -> dict[str, DBusServiceMock | dict[str, DBusServiceMock]]: + """Mock all services UDisks2 connects to.""" + yield await mock_dbus_services( + { + "udisks2_block": [ + "/org/freedesktop/UDisks2/block_devices/loop0", + "/org/freedesktop/UDisks2/block_devices/mmcblk1", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p1", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p2", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p3", + "/org/freedesktop/UDisks2/block_devices/sda", + "/org/freedesktop/UDisks2/block_devices/sda1", + "/org/freedesktop/UDisks2/block_devices/sdb", + "/org/freedesktop/UDisks2/block_devices/sdb1", + "/org/freedesktop/UDisks2/block_devices/zram1", + ], + "udisks2_drive": [ + "/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291", + "/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6", + "/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56", + ], + "udisks2_filesystem": [ + "/org/freedesktop/UDisks2/block_devices/mmcblk1p1", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p3", + "/org/freedesktop/UDisks2/block_devices/sda1", + "/org/freedesktop/UDisks2/block_devices/sdb1", + "/org/freedesktop/UDisks2/block_devices/zram1", + ], + "udisks2_loop": None, + "udisks2_manager": None, + "udisks2_partition_table": [ + "/org/freedesktop/UDisks2/block_devices/mmcblk1", + "/org/freedesktop/UDisks2/block_devices/sda", + "/org/freedesktop/UDisks2/block_devices/sdb", + ], + "udisks2_parition": [ + "/org/freedesktop/UDisks2/block_devices/mmcblk1p1", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p2", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p3", + "/org/freedesktop/UDisks2/block_devices/sda1", + "/org/freedesktop/UDisks2/block_devices/sdb1", + ], + }, + dbus_session_bus, + ) + + async def mock_dbus_interface( dbus: DBus, dbus_bus: MessageBus, instance: DBusInterface ) -> DBusInterface: diff --git a/tests/dbus/network/setting/test_init.py b/tests/dbus/network/setting/test_init.py index 856ab4cfb..d79ce0bac 100644 --- a/tests/dbus/network/setting/test_init.py +++ b/tests/dbus/network/setting/test_init.py @@ -18,7 +18,7 @@ from tests.dbus_service_mocks.network_connection_settings import ( @pytest.fixture(name="connection_settings_service", autouse=True) async def fixture_connection_settings_service( - network_manager_services: dict[str, DBusServiceMock] + network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]] ) -> ConnectionSettingsService: """Mock Connection Settings service.""" yield network_manager_services["network_connection_settings"] diff --git a/tests/dbus/network/test_connection.py b/tests/dbus/network/test_connection.py index a4718c60d..dfd1331fc 100644 --- a/tests/dbus/network/test_connection.py +++ b/tests/dbus/network/test_connection.py @@ -16,7 +16,7 @@ from tests.dbus_service_mocks.network_active_connection import ( @pytest.fixture(name="active_connection_service", autouse=True) async def fixture_active_connection_service( - network_manager_services: dict[str, DBusServiceMock] + network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]] ) -> ActiveConnectionService: """Mock Active Connection service.""" yield network_manager_services["network_active_connection"] diff --git a/tests/dbus/network/test_interface.py b/tests/dbus/network/test_interface.py index 141260497..9fab883b1 100644 --- a/tests/dbus/network/test_interface.py +++ b/tests/dbus/network/test_interface.py @@ -17,18 +17,22 @@ from tests.dbus_service_mocks.network_device import Device as DeviceService @pytest.fixture(name="device_eth0_service") async def fixture_device_eth0_service( - network_manager_services: dict[str, DBusServiceMock] + network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]] ) -> DeviceService: """Mock Device eth0 service.""" - yield network_manager_services["network_device"][0] + yield network_manager_services["network_device"][ + "/org/freedesktop/NetworkManager/Devices/1" + ] @pytest.fixture(name="device_wlan0_service") async def fixture_device_wlan0_service( - network_manager_services: dict[str, DBusServiceMock] + network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]] ) -> DeviceService: """Mock Device wlan0 service.""" - yield network_manager_services["network_device"][1] + yield network_manager_services["network_device"][ + "/org/freedesktop/NetworkManager/Devices/3" + ] @pytest.fixture(name="device_unmanaged_service") diff --git a/tests/dbus/network/test_network_manager.py b/tests/dbus/network/test_network_manager.py index 0197ff75c..cd203d787 100644 --- a/tests/dbus/network/test_network_manager.py +++ b/tests/dbus/network/test_network_manager.py @@ -22,7 +22,7 @@ from tests.dbus_service_mocks.network_manager import ( @pytest.fixture(name="network_manager_service", autouse=True) async def fixture_network_manager_service( - network_manager_services: dict[str, DBusServiceMock], + network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]], ) -> NetworkManagerService: """Mock NetworkManager dbus service.""" yield network_manager_services["network_manager"] diff --git a/tests/dbus/network/test_wireless.py b/tests/dbus/network/test_wireless.py index 07c47ddce..055185723 100644 --- a/tests/dbus/network/test_wireless.py +++ b/tests/dbus/network/test_wireless.py @@ -15,7 +15,7 @@ from tests.dbus_service_mocks.network_device_wireless import ( @pytest.fixture(name="device_wireless_service", autouse=True) async def fixture_device_wireless_service( - network_manager_services: dict[str, DBusServiceMock] + network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]] ) -> DeviceWirelessService: """Mock Device Wireless service.""" yield network_manager_services["network_device_wireless"] diff --git a/tests/dbus/udisks2/test_block.py b/tests/dbus/udisks2/test_block.py index d9429e744..8014475df 100644 --- a/tests/dbus/udisks2/test_block.py +++ b/tests/dbus/udisks2/test_block.py @@ -1,8 +1,8 @@ """Test UDisks2 Block Device interface.""" -import asyncio from pathlib import Path +from dbus_fast import Variant from dbus_fast.aio.message_bus import MessageBus import pytest @@ -10,10 +10,35 @@ from supervisor.dbus.udisks2.block import UDisks2Block from supervisor.dbus.udisks2.const import FormatType, PartitionTableType from supervisor.dbus.udisks2.data import FormatOptions -from tests.common import fire_property_change_signal +from tests.dbus_service_mocks.base import DBusServiceMock +from tests.dbus_service_mocks.udisks2_block import Block as BlockService -async def test_block_device_info(dbus: list[str], dbus_bus: MessageBus): +@pytest.fixture(name="block_sda_service") +async def fixture_block_sda_service( + udisks2_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]] +) -> BlockService: + """Mock sda Block service.""" + yield udisks2_services["udisks2_block"][ + "/org/freedesktop/UDisks2/block_devices/sda" + ] + + +@pytest.fixture(name="block_sda1_service") +async def fixture_block_sda1_service( + udisks2_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]] +) -> BlockService: + """Mock sda1 Block service.""" + yield udisks2_services["udisks2_block"][ + "/org/freedesktop/UDisks2/block_devices/sda1" + ] + + +async def test_block_device_info( + block_sda_service: BlockService, + block_sda1_service: BlockService, + dbus_session_bus: MessageBus, +): """Test block device info.""" sda = UDisks2Block("/org/freedesktop/UDisks2/block_devices/sda") sda1 = UDisks2Block( @@ -28,8 +53,8 @@ async def test_block_device_info(dbus: list[str], dbus_bus: MessageBus): assert sda1.symlinks is None assert sda1.filesystem is None - await sda.connect(dbus_bus) - await sda1.connect(dbus_bus) + await sda.connect(dbus_session_bus) + await sda1.connect(dbus_session_bus) assert sda.drive == "/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56" assert sda.device == Path("/dev/sda") @@ -51,20 +76,33 @@ async def test_block_device_info(dbus: list[str], dbus_bus: MessageBus): assert sda1.partition_table is None assert sda1.filesystem.mount_points == [] - fire_property_change_signal(sda, {"IdLabel": "test"}) - await asyncio.sleep(0) + block_sda_service.emit_properties_changed({"IdLabel": "test"}) + await block_sda_service.ping() assert sda.id_label == "test" - with pytest.raises(AssertionError): - fire_property_change_signal(sda1) + block_sda_service.emit_properties_changed({}, ["IdLabel"]) + await block_sda_service.ping() + await block_sda_service.ping() + assert sda.id_label == "" + + # Prop changes should not sync for this one + block_sda1_service.emit_properties_changed({"IdLabel": "test"}) + await block_sda1_service.ping() + assert sda1.id_label == "hassos-data" -async def test_format(dbus: list[str], dbus_bus: MessageBus): +async def test_format(block_sda_service: BlockService, dbus_session_bus: MessageBus): """Test formatting block device.""" sda = UDisks2Block("/org/freedesktop/UDisks2/block_devices/sda") - await sda.connect(dbus_bus) + await sda.connect(dbus_session_bus) await sda.format(FormatType.GPT, FormatOptions(label="test")) - assert dbus == [ - "/org/freedesktop/UDisks2/block_devices/sda-org.freedesktop.UDisks2.Block.Format" + assert block_sda_service.Format.calls == [ + ( + "gpt", + { + "label": Variant("s", "test"), + "auth.no_user_interaction": Variant("b", True), + }, + ) ] diff --git a/tests/dbus/udisks2/test_drive.py b/tests/dbus/udisks2/test_drive.py index f4373ae5d..1f165be64 100644 --- a/tests/dbus/udisks2/test_drive.py +++ b/tests/dbus/udisks2/test_drive.py @@ -1,16 +1,50 @@ """Test UDisks2 Drive.""" -import asyncio from datetime import datetime, timezone +from dbus_fast import Variant from dbus_fast.aio.message_bus import MessageBus +import pytest from supervisor.dbus.udisks2.drive import UDisks2Drive -from tests.common import fire_property_change_signal +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.udisks2_drive import Drive as DriveService -async def test_drive_info(dbus: list[str], dbus_bus: MessageBus): +@pytest.fixture(name="drive_ssk_storage_service") +async def fixture_drive_ssk_storage_service( + dbus_session_bus: MessageBus, +) -> DriveService: + """Mock SSK Storage Drive service.""" + yield ( + await mock_dbus_services( + { + "udisks2_drive": "/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56" + }, + dbus_session_bus, + ) + )["udisks2_drive"] + + +@pytest.fixture(name="drive_flash_disk_service") +async def fixture_drive_flash_disk_service( + dbus_session_bus: MessageBus, +) -> DriveService: + """Mock Flash Disk Drive service.""" + yield ( + await mock_dbus_services( + { + "udisks2_drive": "/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6" + }, + dbus_session_bus, + ) + )["udisks2_drive"] + + +async def test_drive_info( + drive_ssk_storage_service: DriveService, dbus_session_bus: MessageBus +): """Test drive info.""" ssk = UDisks2Drive("/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56") @@ -20,7 +54,7 @@ async def test_drive_info(dbus: list[str], dbus_bus: MessageBus): assert ssk.time_detected is None assert ssk.ejectable is None - await ssk.connect(dbus_bus) + await ssk.connect(dbus_session_bus) assert ssk.vendor == "SSK" assert ssk.model == "SSK Storage" @@ -28,17 +62,24 @@ async def test_drive_info(dbus: list[str], dbus_bus: MessageBus): assert ssk.time_detected == datetime(2023, 2, 8, 23, 1, 44, 240492, timezone.utc) assert ssk.ejectable is False - fire_property_change_signal(ssk, {"Ejectable": True}) - await asyncio.sleep(0) + drive_ssk_storage_service.emit_properties_changed({"Ejectable": True}) + await drive_ssk_storage_service.ping() assert ssk.ejectable is True + drive_ssk_storage_service.emit_properties_changed({}, ["Ejectable"]) + await drive_ssk_storage_service.ping() + await drive_ssk_storage_service.ping() + assert ssk.ejectable is False -async def test_eject(dbus: list[str], dbus_bus: MessageBus): + +async def test_eject( + drive_flash_disk_service: DriveService, dbus_session_bus: MessageBus +): """Test eject.""" flash = UDisks2Drive("/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6") - await flash.connect(dbus_bus) + await flash.connect(dbus_session_bus) await flash.eject() - assert dbus == [ - "/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6-org.freedesktop.UDisks2.Drive.Eject" + assert drive_flash_disk_service.Eject.calls == [ + ({"auth.no_user_interaction": Variant("b", True)},) ] diff --git a/tests/dbus/udisks2/test_filesystem.py b/tests/dbus/udisks2/test_filesystem.py index db4c186b1..d5324318f 100644 --- a/tests/dbus/udisks2/test_filesystem.py +++ b/tests/dbus/udisks2/test_filesystem.py @@ -1,26 +1,59 @@ """Test UDisks2 Filesystem.""" -import asyncio from pathlib import Path +from dbus_fast import Variant from dbus_fast.aio.message_bus import MessageBus import pytest from supervisor.dbus.udisks2.data import MountOptions, UnmountOptions from supervisor.dbus.udisks2.filesystem import UDisks2Filesystem -from tests.common import fire_property_change_signal +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.udisks2_filesystem import Filesystem as FilesystemService + + +@pytest.fixture(name="filesystem_sda1_service") +async def fixture_filesystem_sda1_service( + dbus_session_bus: MessageBus, +) -> FilesystemService: + """Mock sda1 Filesystem service.""" + yield ( + await mock_dbus_services( + {"udisks2_filesystem": "/org/freedesktop/UDisks2/block_devices/sda1"}, + dbus_session_bus, + ) + )["udisks2_filesystem"] + + +@pytest.fixture(name="filesystem_sdb1_service") +async def fixture_filesystem_sdb1_service( + dbus_session_bus: MessageBus, +) -> FilesystemService: + """Mock sdb1 Filesystem service.""" + yield ( + await mock_dbus_services( + {"udisks2_filesystem": "/org/freedesktop/UDisks2/block_devices/sdb1"}, + dbus_session_bus, + ) + )["udisks2_filesystem"] @pytest.fixture(name="sda1") -async def fixture_sda1(dbus: list[str], dbus_bus: MessageBus) -> UDisks2Filesystem: +async def fixture_sda1( + filesystem_sda1_service: FilesystemService, dbus_session_bus: MessageBus +) -> UDisks2Filesystem: """Return connected sda1 filesystem object.""" filesystem = UDisks2Filesystem("/org/freedesktop/UDisks2/block_devices/sda1") - await filesystem.connect(dbus_bus) + await filesystem.connect(dbus_session_bus) return filesystem -async def test_filesystem_info(dbus: list[str], dbus_bus: MessageBus): +async def test_filesystem_info( + filesystem_sda1_service: FilesystemService, + filesystem_sdb1_service: FilesystemService, + dbus_session_bus: MessageBus, +): """Test filesystem info.""" sda1 = UDisks2Filesystem("/org/freedesktop/UDisks2/block_devices/sda1") sdb1 = UDisks2Filesystem( @@ -32,53 +65,69 @@ async def test_filesystem_info(dbus: list[str], dbus_bus: MessageBus): assert sdb1.size is None assert sdb1.mount_points is None - await sda1.connect(dbus_bus) - await sdb1.connect(dbus_bus) + await sda1.connect(dbus_session_bus) + await sdb1.connect(dbus_session_bus) assert sda1.size == 250058113024 assert sda1.mount_points == [] assert sdb1.size == 67108864 assert sdb1.mount_points == [Path("/mnt/data/supervisor/media/ext")] - fire_property_change_signal( - sda1, {"MountPoints": [bytearray("/mnt/media", encoding="utf-8")]} - ) - await asyncio.sleep(0) + filesystem_sda1_service.emit_properties_changed({"MountPoints": [b"/mnt/media"]}) + await filesystem_sda1_service.ping() assert sda1.mount_points == [Path("/mnt/media")] - with pytest.raises(AssertionError): - fire_property_change_signal( - sdb1, {"MountPoints": [bytearray("/mnt/media", encoding="utf-8")]} - ) + filesystem_sda1_service.emit_properties_changed({}, ["MountPoints"]) + await filesystem_sda1_service.ping() + await filesystem_sda1_service.ping() + assert sda1.mount_points == [] + + # Prop changes should not sync for this one + filesystem_sdb1_service.emit_properties_changed({"MountPoints": [b"/mnt/media"]}) + await filesystem_sdb1_service.ping() + assert sdb1.mount_points == [Path("/mnt/data/supervisor/media/ext")] -async def test_mount(dbus: list[str], sda1: UDisks2Filesystem): +async def test_mount( + sda1: UDisks2Filesystem, filesystem_sda1_service: FilesystemService +): """Test mount.""" assert await sda1.mount(MountOptions(fstype="gpt")) == "/run/media/dev/hassos_data" - assert dbus == [ - "/org/freedesktop/UDisks2/block_devices/sda1-org.freedesktop.UDisks2.Filesystem.Mount" + assert filesystem_sda1_service.Mount.calls == [ + ( + { + "fstype": Variant("s", "gpt"), + "auth.no_user_interaction": Variant("b", True), + }, + ) ] -async def test_unmount(dbus: list[str], sda1: UDisks2Filesystem): +async def test_unmount( + sda1: UDisks2Filesystem, filesystem_sda1_service: FilesystemService +): """Test unmount.""" await sda1.unmount(UnmountOptions(force=True)) - assert dbus == [ - "/org/freedesktop/UDisks2/block_devices/sda1-org.freedesktop.UDisks2.Filesystem.Unmount" + assert filesystem_sda1_service.Unmount.calls == [ + ({"force": Variant("b", True), "auth.no_user_interaction": Variant("b", True)},) ] -async def test_check(dbus: list[str], sda1: UDisks2Filesystem): +async def test_check( + sda1: UDisks2Filesystem, filesystem_sda1_service: FilesystemService +): """Test check.""" assert await sda1.check() is True - assert dbus == [ - "/org/freedesktop/UDisks2/block_devices/sda1-org.freedesktop.UDisks2.Filesystem.Check" + assert filesystem_sda1_service.Check.calls == [ + ({"auth.no_user_interaction": Variant("b", True)},) ] -async def test_repair(dbus: list[str], sda1: UDisks2Filesystem): +async def test_repair( + sda1: UDisks2Filesystem, filesystem_sda1_service: FilesystemService +): """Test repair.""" assert await sda1.repair() is True - assert dbus == [ - "/org/freedesktop/UDisks2/block_devices/sda1-org.freedesktop.UDisks2.Filesystem.Repair" + assert filesystem_sda1_service.Repair.calls == [ + ({"auth.no_user_interaction": Variant("b", True)},) ] diff --git a/tests/dbus/udisks2/test_manager.py b/tests/dbus/udisks2/test_manager.py index 71fc546be..f33a11734 100644 --- a/tests/dbus/udisks2/test_manager.py +++ b/tests/dbus/udisks2/test_manager.py @@ -1,33 +1,48 @@ """Test UDisks2 Manager interface.""" -import asyncio - from awesomeversion import AwesomeVersion +from dbus_fast import Variant +from dbus_fast.aio.message_bus import MessageBus import pytest -from supervisor.coresys import CoreSys +from supervisor.dbus.udisks2 import UDisks2 from supervisor.dbus.udisks2.data import DeviceSpecification from supervisor.exceptions import DBusNotConnectedError, DBusObjectError -from tests.common import fire_property_change_signal +from tests.dbus_service_mocks.base import DBusServiceMock +from tests.dbus_service_mocks.udisks2_manager import ( + UDisks2Manager as UDisks2ManagerService, +) -async def test_udisks2_manager_info(coresys: CoreSys, dbus: list[str]): +@pytest.fixture(name="udisks2_manager_service", autouse=True) +async def fixture_udisks2_manager_service( + udisks2_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]] +) -> UDisks2ManagerService: + """Mock UDisks2 Manager service.""" + yield udisks2_services["udisks2_manager"] + + +async def test_udisks2_manager_info( + udisks2_manager_service: UDisks2ManagerService, dbus_session_bus: MessageBus +): """Test udisks2 manager dbus connection.""" - dbus.clear() - assert coresys.dbus.udisks2.supported_filesystems is None + udisks2_manager_service.GetBlockDevices.calls.clear() + udisks2 = UDisks2() - await coresys.dbus.udisks2.connect(coresys.dbus.bus) + assert udisks2.supported_filesystems is None - assert coresys.dbus.udisks2.supported_filesystems == [ + await udisks2.connect(dbus_session_bus) + + assert udisks2.supported_filesystems == [ "ext4", "vfat", "ntfs", "exfat", "swap", ] - assert coresys.dbus.udisks2.version == AwesomeVersion("2.9.2") - assert {block.object_path for block in coresys.dbus.udisks2.block_devices} == { + assert udisks2.version == AwesomeVersion("2.9.2") + assert {block.object_path for block in udisks2.block_devices} == { "/org/freedesktop/UDisks2/block_devices/loop0", "/org/freedesktop/UDisks2/block_devices/mmcblk1", "/org/freedesktop/UDisks2/block_devices/mmcblk1p1", @@ -39,73 +54,87 @@ async def test_udisks2_manager_info(coresys: CoreSys, dbus: list[str]): "/org/freedesktop/UDisks2/block_devices/sdb1", "/org/freedesktop/UDisks2/block_devices/zram1", } - assert {drive.object_path for drive in coresys.dbus.udisks2.drives} == { + assert {drive.object_path for drive in udisks2.drives} == { "/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291", "/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56", "/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6", } - assert dbus == [ - "/org/freedesktop/UDisks2/Manager-org.freedesktop.UDisks2.Manager.GetBlockDevices" + assert udisks2_manager_service.GetBlockDevices.calls == [ + ({"auth.no_user_interaction": Variant("b", True)},) ] - dbus.clear() - fire_property_change_signal( - coresys.dbus.udisks2, {"SupportedFilesystems": ["ext4"]} - ) - await asyncio.sleep(0) - assert coresys.dbus.udisks2.supported_filesystems == ["ext4"] - assert dbus == [] + udisks2_manager_service.GetBlockDevices.calls.clear() + udisks2_manager_service.emit_properties_changed({"SupportedFilesystems": ["ext4"]}) + await udisks2_manager_service.ping() + assert udisks2.supported_filesystems == ["ext4"] + assert udisks2_manager_service.GetBlockDevices.calls == [] + + udisks2_manager_service.emit_properties_changed({}, ["SupportedFilesystems"]) + await udisks2_manager_service.ping() + await udisks2_manager_service.ping() + assert udisks2.supported_filesystems == [ + "ext4", + "vfat", + "ntfs", + "exfat", + "swap", + ] + assert udisks2_manager_service.GetBlockDevices.calls == [ + ({"auth.no_user_interaction": Variant("b", True)},) + ] -async def test_get_block_device(coresys: CoreSys): +async def test_get_block_device(dbus_session_bus: MessageBus): """Test get block device by object path.""" + udisks2 = UDisks2() + with pytest.raises(DBusNotConnectedError): - coresys.dbus.udisks2.get_block_device( - "/org/freedesktop/UDisks2/block_devices/sda1" - ) + udisks2.get_block_device("/org/freedesktop/UDisks2/block_devices/sda1") - await coresys.dbus.udisks2.connect(coresys.dbus.bus) + await udisks2.connect(dbus_session_bus) - block_device = coresys.dbus.udisks2.get_block_device( + block_device = udisks2.get_block_device( "/org/freedesktop/UDisks2/block_devices/sda1" ) assert block_device.id_label == "hassos-data" with pytest.raises(DBusObjectError): - coresys.dbus.udisks2.get_block_device("non_existent") + udisks2.get_block_device("non_existent") -async def test_get_drive(coresys: CoreSys): +async def test_get_drive(dbus_session_bus: MessageBus): """Test get drive by object path.""" + udisks2 = UDisks2() + with pytest.raises(DBusNotConnectedError): - coresys.dbus.udisks2.get_drive( - "/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291" - ) + udisks2.get_drive("/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291") - await coresys.dbus.udisks2.connect(coresys.dbus.bus) + await udisks2.connect(dbus_session_bus) - drive = coresys.dbus.udisks2.get_drive( - "/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291" - ) + drive = udisks2.get_drive("/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291") assert drive.id == "BJTD4R-0x97cde291" with pytest.raises(DBusObjectError): - coresys.dbus.udisks2.get_drive("non_existent") + udisks2.get_drive("non_existent") -async def test_resolve_device(coresys: CoreSys, dbus: list[str]): +async def test_resolve_device( + udisks2_manager_service: UDisks2ManagerService, dbus_session_bus: MessageBus +): """Test resolve device.""" + udisks2 = UDisks2() + with pytest.raises(DBusNotConnectedError): - await coresys.dbus.udisks2.resolve_device(DeviceSpecification(path="/dev/sda1")) + await udisks2.resolve_device(DeviceSpecification(path="/dev/sda1")) - await coresys.dbus.udisks2.connect(coresys.dbus.bus) + await udisks2.connect(dbus_session_bus) - dbus.clear() - devices = await coresys.dbus.udisks2.resolve_device( - DeviceSpecification(path="/dev/sda1") - ) + devices = await udisks2.resolve_device(DeviceSpecification(path="/dev/sda1")) assert len(devices) == 1 assert devices[0].id_label == "hassos-data" - assert dbus == [ - "/org/freedesktop/UDisks2/Manager-org.freedesktop.UDisks2.Manager.ResolveDevice" + assert udisks2_manager_service.ResolveDevice.calls == [ + ( + {"path": Variant("s", "/dev/sda1")}, + {"auth.no_user_interaction": Variant("b", True)}, + ) ] diff --git a/tests/dbus/udisks2/test_partition_table.py b/tests/dbus/udisks2/test_partition_table.py index a6ea81001..1c50a9d8c 100644 --- a/tests/dbus/udisks2/test_partition_table.py +++ b/tests/dbus/udisks2/test_partition_table.py @@ -1,7 +1,6 @@ """Test UDisks2 Partition Table.""" -import asyncio - +from dbus_fast import Variant from dbus_fast.aio.message_bus import MessageBus import pytest @@ -9,10 +8,43 @@ from supervisor.dbus.udisks2.const import PartitionTableType from supervisor.dbus.udisks2.data import CreatePartitionOptions from supervisor.dbus.udisks2.partition_table import UDisks2PartitionTable -from tests.common import fire_property_change_signal +from tests.common import mock_dbus_services +from tests.dbus_service_mocks.udisks2_partition_table import ( + PartitionTable as PartitionTableService, +) -async def test_partition_table_info(dbus: list[str], dbus_bus: MessageBus): +@pytest.fixture(name="partition_table_sda_service") +async def fixture_partition_table_sda_service( + dbus_session_bus: MessageBus, +) -> PartitionTableService: + """Mock sda Partition Table service.""" + yield ( + await mock_dbus_services( + {"udisks2_partition_table": "/org/freedesktop/UDisks2/block_devices/sda"}, + dbus_session_bus, + ) + )["udisks2_partition_table"] + + +@pytest.fixture(name="partition_table_sdb_service") +async def fixture_partition_table_sdb_service( + dbus_session_bus: MessageBus, +) -> PartitionTableService: + """Mock sdb Partition Table service.""" + yield ( + await mock_dbus_services( + {"udisks2_partition_table": "/org/freedesktop/UDisks2/block_devices/sdb"}, + dbus_session_bus, + ) + )["udisks2_partition_table"] + + +async def test_partition_table_info( + partition_table_sda_service: PartitionTableService, + partition_table_sdb_service: PartitionTableService, + dbus_session_bus: MessageBus, +): """Test partition table info.""" sda = UDisks2PartitionTable("/org/freedesktop/UDisks2/block_devices/sda") sdb = UDisks2PartitionTable( @@ -24,16 +56,15 @@ async def test_partition_table_info(dbus: list[str], dbus_bus: MessageBus): assert sdb.type is None assert sdb.partitions is None - await sda.connect(dbus_bus) - await sdb.connect(dbus_bus) + await sda.connect(dbus_session_bus) + await sdb.connect(dbus_session_bus) assert sda.type == PartitionTableType.GPT assert sda.partitions == ["/org/freedesktop/UDisks2/block_devices/sda1"] assert sdb.type == PartitionTableType.GPT assert sdb.partitions == ["/org/freedesktop/UDisks2/block_devices/sdb1"] - fire_property_change_signal( - sda, + partition_table_sda_service.emit_properties_changed( { "Partitions": [ "/org/freedesktop/UDisks2/block_devices/sda1", @@ -41,28 +72,36 @@ async def test_partition_table_info(dbus: list[str], dbus_bus: MessageBus): ] }, ) - await asyncio.sleep(0) + await partition_table_sda_service.ping() assert sda.partitions == [ "/org/freedesktop/UDisks2/block_devices/sda1", "/org/freedesktop/UDisks2/block_devices/sda2", ] - with pytest.raises(AssertionError): - fire_property_change_signal( - sdb, - { - "MountPoints": [ - "/org/freedesktop/UDisks2/block_devices/sdb", - "/org/freedesktop/UDisks2/block_devices/sdb", - ] - }, - ) + partition_table_sda_service.emit_properties_changed({}, ["Partitions"]) + await partition_table_sda_service.ping() + await partition_table_sda_service.ping() + assert sda.partitions == ["/org/freedesktop/UDisks2/block_devices/sda1"] + + # Prop changes should not sync for this one + partition_table_sdb_service.emit_properties_changed( + { + "Partitions": [ + "/org/freedesktop/UDisks2/block_devices/sdb1", + "/org/freedesktop/UDisks2/block_devices/sdb2", + ] + }, + ) + await partition_table_sdb_service.ping() + assert sdb.partitions == ["/org/freedesktop/UDisks2/block_devices/sdb1"] -async def test_create_partition(dbus: list[str], dbus_bus: MessageBus): +async def test_create_partition( + partition_table_sda_service: PartitionTableService, dbus_session_bus: MessageBus +): """Test create partition.""" sda = UDisks2PartitionTable("/org/freedesktop/UDisks2/block_devices/sda") - await sda.connect(dbus_bus) + await sda.connect(dbus_session_bus) assert ( await sda.create_partition( @@ -74,6 +113,15 @@ async def test_create_partition(dbus: list[str], dbus_bus: MessageBus): ) == "/org/freedesktop/UDisks2/block_devices/sda2" ) - assert dbus == [ - "/org/freedesktop/UDisks2/block_devices/sda-org.freedesktop.UDisks2.PartitionTable.CreatePartition" + assert partition_table_sda_service.CreatePartition.calls == [ + ( + 0, + 1000000, + "dos", + "hassos-data", + { + "partition-type": Variant("s", "primary"), + "auth.no_user_interaction": Variant("b", True), + }, + ) ] diff --git a/tests/dbus_service_mocks/udisks2_block.py b/tests/dbus_service_mocks/udisks2_block.py new file mode 100644 index 000000000..01c27f2d6 --- /dev/null +++ b/tests/dbus_service_mocks/udisks2_block.py @@ -0,0 +1,549 @@ +"""Mock of UDisks2 Block service.""" + +from ctypes import c_uint64 +from dataclasses import dataclass + +from dbus_fast import Variant +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.UDisks2" +DEFAULT_OBJECT_PATH = "/org/freedesktop/UDisks2/block_devices/sda" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Block(object_path if object_path else DEFAULT_OBJECT_PATH) + + +# pylint: disable=invalid-name + + +@dataclass(slots=True) +class BlockFixture: + """Block fixture.""" + + Device: bytes + PreferredDevice: bytes + Symlinks: list[bytes] + DeviceNumber: c_uint64 + Id: str + Size: c_uint64 + ReadOnly: bool + Drive: str + MDRaid: str + MDRaidMember: str + IdUsage: str + IdType: str + IdVersion: str + IdLabel: str + IdUUID: str + Configuration: list[list[str | dict[str, Variant]]] + CryptoBackingDevice: str + HintPartitionable: bool + HintSystem: bool + HintIgnore: bool + HintAuto: bool + HintName: str + HintIconName: str + HintSymbolicIconName: str + UserspaceMountOptions: list[str] + + +FIXTURES: dict[str, BlockFixture] = { + "/org/freedesktop/UDisks2/block_devices/loop0": BlockFixture( + Device=b"/dev/loop0", + PreferredDevice=b"/dev/loop0", + Symlinks=[], + DeviceNumber=1792, + Id="", + Size=0, + ReadOnly=False, + Drive="/", + MDRaid="/", + MDRaidMember="/", + IdUsage="", + IdType="", + IdVersion="", + IdLabel="", + IdUUID="", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=True, + HintIgnore=False, + HintAuto=False, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), + "/org/freedesktop/UDisks2/block_devices/mmcblk1": BlockFixture( + Device=b"/dev/mmcblk1", + PreferredDevice=b"/dev/mmcblk1", + Symlinks=[ + b"/dev/disk/by-id/mmc-BJTD4R_0x97cde291", + b"/dev/disk/by-path/platform-ffe07000.mmc", + ], + DeviceNumber=45824, + Id="by-id-mmc-BJTD4R_0x97cde291", + Size=31268536320, + ReadOnly=False, + Drive="/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291", + MDRaid="/", + MDRaidMember="/", + IdUsage="", + IdType="", + IdVersion="", + IdLabel="", + IdUUID="", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=True, + HintIgnore=False, + HintAuto=False, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), + "/org/freedesktop/UDisks2/block_devices/mmcblk1p1": BlockFixture( + Device=b"/dev/mmcblk1p1", + PreferredDevice=b"/dev/mmcblk1p1", + Symlinks=[ + b"/dev/disk/by-id/mmc-BJTD4R_0x97cde291-part1", + b"/dev/disk/by-label/hassos-boot", + b"/dev/disk/by-partlabel/hassos-boot", + b"/dev/disk/by-partuuid/48617373-01", + b"/dev/disk/by-path/platform-ffe07000.mmc-part1", + b"/dev/disk/by-uuid/16DD-EED4", + ], + DeviceNumber=45825, + Id="by-id-mmc-BJTD4R_0x97cde291-part1", + Size=25165824, + ReadOnly=False, + Drive="/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291", + MDRaid="/", + MDRaidMember="/", + IdUsage="filesystem", + IdType="vfat", + IdVersion="FAT16", + IdLabel="hassos-boot", + IdUUID="16DD-EED4", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=True, + HintIgnore=False, + HintAuto=False, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), + "/org/freedesktop/UDisks2/block_devices/mmcblk1p2": BlockFixture( + Device=b"/dev/mmcblk1p2", + PreferredDevice=b"/dev/mmcblk1p2", + Symlinks=[ + b"/dev/disk/by-id/mmc-BJTD4R_0x97cde291-part2", + b"/dev/disk/by-partuuid/48617373-02", + b"/dev/disk/by-path/platform-ffe07000.mmc-part2", + ], + DeviceNumber=45826, + Id="by-id-mmc-BJTD4R_0x97cde291-part2", + Size=1024, + ReadOnly=False, + Drive="/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291", + MDRaid="/", + MDRaidMember="/", + IdUsage="", + IdType="", + IdVersion="", + IdLabel="", + IdUUID="", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=True, + HintIgnore=False, + HintAuto=False, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), + "/org/freedesktop/UDisks2/block_devices/mmcblk1p3": BlockFixture( + Device=b"/dev/mmcblk1p3", + PreferredDevice=b"/dev/mmcblk1p3", + Symlinks=[ + b"/dev/disk/by-id/mmc-BJTD4R_0x97cde291-part3", + b"/dev/disk/by-label/hassos-overlay", + b"/dev/disk/by-partuuid/48617373-03", + b"/dev/disk/by-path/platform-ffe07000.mmc-part3", + b"/dev/disk/by-uuid/0cd0d026-8c69-484e-bbf1-8197019fa7df", + ], + DeviceNumber=45827, + Id="by-id-mmc-BJTD4R_0x97cde291-part3", + Size=100663296, + ReadOnly=False, + Drive="/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291", + MDRaid="/", + MDRaidMember="/", + IdUsage="filesystem", + IdType="ext4", + IdVersion="1.0", + IdLabel="hassos-overlay", + IdUUID="0cd0d026-8c69-484e-bbf1-8197019fa7df", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=True, + HintIgnore=False, + HintAuto=False, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), + "/org/freedesktop/UDisks2/block_devices/sda": BlockFixture( + Device=b"/dev/sda", + PreferredDevice=b"/dev/sda", + Symlinks=[ + b"/dev/disk/by-id/usb-SSK_SSK_Storage_DF56419883D56-0:0", + b"/dev/disk/by-path/platform-xhci-hcd.1.auto-usb-0:1.4:1.0-scsi-0:0:0:0", + ], + DeviceNumber=2048, + Id="by-id-usb-SSK_SSK_Storage_DF56419883D56-0:0", + Size=250059350016, + ReadOnly=False, + Drive="/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56", + MDRaid="/", + MDRaidMember="/", + IdUsage="", + IdType="", + IdVersion="", + IdLabel="", + IdUUID="", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=False, + HintIgnore=False, + HintAuto=True, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), + "/org/freedesktop/UDisks2/block_devices/sda1": BlockFixture( + Device=b"/dev/sda1", + PreferredDevice=b"/dev/sda1", + Symlinks=[ + b"/dev/disk/by-id/usb-SSK_SSK_Storage_DF56419883D56-0:0-part1", + b"/dev/disk/by-label/hassos-data", + b"/dev/disk/by-partlabel/hassos-data-external", + b"/dev/disk/by-partuuid/6f3f99f4-4d34-476b-b051-77886da57fa9", + b"/dev/disk/by-path/platform-xhci-hcd.1.auto-usb-0:1.4:1.0-scsi-0:0:0:0-part1", + b"/dev/disk/by-uuid/b82b23cb-0c47-4bbb-acf5-2a2afa8894a2", + ], + DeviceNumber=2049, + Id="by-id-usb-SSK_SSK_Storage_DF56419883D56-0:0-part1", + Size=250058113024, + ReadOnly=False, + Drive="/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56", + MDRaid="/", + MDRaidMember="/", + IdUsage="filesystem", + IdType="ext4", + IdVersion="1.0", + IdLabel="hassos-data", + IdUUID="b82b23cb-0c47-4bbb-acf5-2a2afa8894a2", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=False, + HintIgnore=False, + HintAuto=True, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), + "/org/freedesktop/UDisks2/block_devices/sdb": BlockFixture( + Device=b"/dev/sdb", + PreferredDevice=b"/dev/sdb", + Symlinks=[ + b"/dev/disk/by-id/usb-Generic_Flash_Disk_61BCDDB6-0:0", + b"/dev/disk/by-path/platform-xhci-hcd.1.auto-usb-0:1.2:1.0-scsi-0:0:0:0", + ], + DeviceNumber=2064, + Id="", + Size=8054112256, + ReadOnly=False, + Drive="/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6", + MDRaid="/", + MDRaidMember="/", + IdUsage="", + IdType="", + IdVersion="", + IdLabel="", + IdUUID="", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=False, + HintIgnore=False, + HintAuto=True, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), + "/org/freedesktop/UDisks2/block_devices/sdb1": BlockFixture( + Device=b"/dev/sdb1", + PreferredDevice=b"/dev/sdb1", + Symlinks=[ + b"/dev/disk/by-id/usb-Generic_Flash_Disk_61BCDDB6-0:0-part1", + b"/dev/disk/by-path/platform-xhci-hcd.1.auto-usb-0:1.2:1.0-scsi-0:0:0:0-part1", + b"/dev/disk/by-uuid/2802-1EDE", + ], + DeviceNumber=2065, + Id="by-uuid-2802-1EDE", + Size=67108864, + ReadOnly=False, + Drive="/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6", + MDRaid="/", + MDRaidMember="/", + IdUsage="filesystem", + IdType="vfat", + IdVersion="FAT16", + IdLabel="", + IdUUID="2802-1EDE", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=False, + HintIgnore=False, + HintAuto=True, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), + "/org/freedesktop/UDisks2/block_devices/zram1": BlockFixture( + Device=b"/dev/zram1", + PreferredDevice=b"/dev/zram1", + Symlinks=[], + DeviceNumber=64769, + Id="", + Size=33554432, + ReadOnly=False, + Drive="/", + MDRaid="/", + MDRaidMember="/", + IdUsage="", + IdType="", + IdVersion="", + IdLabel="", + IdUUID="", + Configuration=[], + CryptoBackingDevice="/", + HintPartitionable=True, + HintSystem=True, + HintIgnore=False, + HintAuto=False, + HintName="", + HintIconName="", + HintSymbolicIconName="", + UserspaceMountOptions=[], + ), +} + + +class Block(DBusServiceMock): + """Block mock. + + gdbus introspect --system --dest org.freedesktop.UDisks2 --object-path /org/freedesktop/UDisks2/block_devices/sda + """ + + interface = "org.freedesktop.UDisks2.Block" + + def __init__(self, object_path: str): + """Initialize object.""" + super().__init__() + self.object_path = object_path + self.fixture: BlockFixture = FIXTURES[object_path] + + @dbus_property(access=PropertyAccess.READ) + def Device(self) -> "ay": + """Get Device.""" + return self.fixture.Device + + @dbus_property(access=PropertyAccess.READ) + def PreferredDevice(self) -> "ay": + """Get PreferredDevice.""" + return self.fixture.PreferredDevice + + @dbus_property(access=PropertyAccess.READ) + def Symlinks(self) -> "aay": + """Get Symlinks.""" + return self.fixture.Symlinks + + @dbus_property(access=PropertyAccess.READ) + def DeviceNumber(self) -> "t": + """Get DeviceNumber.""" + return self.fixture.DeviceNumber + + @dbus_property(access=PropertyAccess.READ) + def Id(self) -> "s": + """Get Id.""" + return self.fixture.Id + + @dbus_property(access=PropertyAccess.READ) + def Size(self) -> "t": + """Get Size.""" + return self.fixture.Size + + @dbus_property(access=PropertyAccess.READ) + def ReadOnly(self) -> "b": + """Get ReadOnly.""" + return self.fixture.ReadOnly + + @dbus_property(access=PropertyAccess.READ) + def Drive(self) -> "o": + """Get Drive.""" + return self.fixture.Drive + + @dbus_property(access=PropertyAccess.READ) + def MDRaid(self) -> "o": + """Get MDRaid.""" + return self.fixture.MDRaid + + @dbus_property(access=PropertyAccess.READ) + def MDRaidMember(self) -> "o": + """Get MDRaidMember.""" + return self.fixture.MDRaidMember + + @dbus_property(access=PropertyAccess.READ) + def IdUsage(self) -> "s": + """Get IdUsage.""" + return self.fixture.IdUsage + + @dbus_property(access=PropertyAccess.READ) + def IdType(self) -> "s": + """Get IdType.""" + return self.fixture.IdType + + @dbus_property(access=PropertyAccess.READ) + def IdVersion(self) -> "s": + """Get IdVersion.""" + return self.fixture.IdVersion + + @dbus_property(access=PropertyAccess.READ) + def IdLabel(self) -> "s": + """Get IdLabel.""" + return self.fixture.IdLabel + + @dbus_property(access=PropertyAccess.READ) + def IdUUID(self) -> "s": + """Get IdUUID.""" + return self.fixture.IdUUID + + @dbus_property(access=PropertyAccess.READ) + def Configuration(self) -> "a(sa{sv})": + """Get Configuration.""" + return self.fixture.Configuration + + @dbus_property(access=PropertyAccess.READ) + def CryptoBackingDevice(self) -> "o": + """Get CryptoBackingDevice.""" + return self.fixture.CryptoBackingDevice + + @dbus_property(access=PropertyAccess.READ) + def HintPartitionable(self) -> "b": + """Get HintPartitionable.""" + return self.fixture.HintPartitionable + + @dbus_property(access=PropertyAccess.READ) + def HintSystem(self) -> "b": + """Get HintSystem.""" + return self.fixture.HintSystem + + @dbus_property(access=PropertyAccess.READ) + def HintIgnore(self) -> "b": + """Get HintIgnore.""" + return self.fixture.HintIgnore + + @dbus_property(access=PropertyAccess.READ) + def HintAuto(self) -> "b": + """Get HintAuto.""" + return self.fixture.HintAuto + + @dbus_property(access=PropertyAccess.READ) + def HintName(self) -> "s": + """Get HintName.""" + return self.fixture.HintName + + @dbus_property(access=PropertyAccess.READ) + def HintIconName(self) -> "s": + """Get HintIconName.""" + return self.fixture.HintIconName + + @dbus_property(access=PropertyAccess.READ) + def HintSymbolicIconName(self) -> "s": + """Get HintSymbolicIconName.""" + return self.fixture.HintSymbolicIconName + + @dbus_property(access=PropertyAccess.READ) + def UserspaceMountOptions(self) -> "as": + """Get UserspaceMountOptions.""" + return self.fixture.UserspaceMountOptions + + @dbus_method() + def AddConfigurationItem(self, item: "(sa{sv})", options: "a{sv}") -> None: + """Do AddConfigurationItem method.""" + + @dbus_method() + def RemoveConfigurationItem(self, item: "(sa{sv})", options: "a{sv}") -> None: + """Do RemoveConfigurationItem method.""" + + @dbus_method() + def UpdateConfigurationItem( + self, old_item: "(sa{sv})", new_item: "(sa{sv})", options: "a{sv}" + ) -> None: + """Do UpdateConfigurationItem method.""" + + @dbus_method() + def GetSecretConfiguration(self, options: "a{sv}") -> "a(sa{sv})": + """Do GetSecretConfiguration method.""" + return [] + + @dbus_method() + def Format(self, type_: "s", options: "a{sv}") -> None: + """Do Format method.""" + + @dbus_method() + def OpenForBackup(self, options: "a{sv}") -> "h": + """Do OpenForBackup method.""" + return 100 + + @dbus_method() + def OpenForRestore(self, options: "a{sv}") -> "h": + """Do OpenForRestore method.""" + return 101 + + @dbus_method() + def OpenForBenchmark(self, options: "a{sv}") -> "h": + """Do OpenForBenchmark method.""" + return 102 + + @dbus_method() + def OpenDevice(self, mode: "s", options: "a{sv}") -> "h": + """Do OpenDevice method.""" + return 103 + + @dbus_method() + def Rescan(self, options: "a{sv}") -> None: + """Do Rescan method.""" diff --git a/tests/dbus_service_mocks/udisks2_drive.py b/tests/dbus_service_mocks/udisks2_drive.py new file mode 100644 index 000000000..66723e632 --- /dev/null +++ b/tests/dbus_service_mocks/udisks2_drive.py @@ -0,0 +1,324 @@ +"""Mock of UDisks2 Drive service.""" + +from ctypes import c_int32, c_uint32, c_uint64 +from dataclasses import dataclass + +from dbus_fast import Variant +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.UDisks2" +DEFAULT_OBJECT_PATH = "/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Drive(object_path if object_path else DEFAULT_OBJECT_PATH) + + +# pylint: disable=invalid-name + + +@dataclass(slots=True) +class DriveFixture: + """Drive fixture.""" + + Vendor: str + Model: str + Revision: str + Serial: str + WWN: str + Id: str + Configuration: dict[str, Variant] + Media: str + MediaCompatibility: list[str] + MediaRemovable: bool + MediaAvailable: bool + MediaChangeDetected: bool + Size: c_uint64 + TimeDetected: c_uint64 + TimeMediaDetected: c_uint64 + Optical: bool + OpticalBlank: bool + OpticalNumTracks: c_uint32 + OpticalNumAudioTracks: c_uint32 + OpticalNumDataTracks: c_uint32 + OpticalNumSessions: c_uint32 + RotationRate: c_int32 + ConnectionBus: str + Seat: str + Removable: bool + Ejectable: bool + SortKey: str + CanPowerOff: bool + SiblingId: str + + +FIXTURES: dict[str, DriveFixture] = { + "/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291": DriveFixture( + Vendor="", + Model="BJTD4R", + Revision="", + Serial="0x97cde291", + WWN="", + Id="BJTD4R-0x97cde291", + Configuration={}, + Media="flash_mmc", + MediaCompatibility=["flash_mmc"], + MediaRemovable=False, + MediaAvailable=True, + MediaChangeDetected=True, + Size=31268536320, + TimeDetected=1673981760067475, + TimeMediaDetected=1673981760067475, + Optical=False, + OpticalBlank=False, + OpticalNumTracks=0, + OpticalNumAudioTracks=0, + OpticalNumDataTracks=0, + OpticalNumSessions=0, + RotationRate=0, + ConnectionBus="sdio", + Seat="seat0", + Removable=False, + Ejectable=False, + SortKey="00coldplug/00fixed/mmcblk1", + CanPowerOff=False, + SiblingId="", + ), + "/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6": DriveFixture( + Vendor="Generic", + Model="Flash Disk", + Revision="8.07", + Serial="61BCDDB6", + WWN="", + Id="Generic-Flash-Disk-61BCDDB6", + Configuration={}, + Media="", + MediaCompatibility=[], + MediaRemovable=True, + MediaAvailable=True, + MediaChangeDetected=True, + Size=8054112256, + TimeDetected=1675972756688073, + TimeMediaDetected=1675972756688073, + Optical=False, + OpticalBlank=False, + OpticalNumTracks=0, + OpticalNumAudioTracks=0, + OpticalNumDataTracks=0, + OpticalNumSessions=0, + RotationRate=-1, + ConnectionBus="usb", + Seat="seat0", + Removable=True, + Ejectable=True, + SortKey="01hotplug/1675972756688073", + CanPowerOff=True, + SiblingId="/sys/devices/platform/soc/ffe09000.usb/ff500000.usb/xhci-hcd.1.auto/usb1/1-1/1-1.2/1-1.2:1.0", + ), + "/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56": DriveFixture( + Vendor="SSK", + Model="SSK Storage", + Revision="0206", + Serial="DF56419883D56", + WWN="", + Id="SSK-SSK-Storage-DF56419883D56", + Configuration={}, + Media="", + MediaCompatibility=[], + MediaRemovable=False, + MediaAvailable=True, + MediaChangeDetected=True, + Size=250059350016, + TimeDetected=1675897304240492, + TimeMediaDetected=1675897304240492, + Optical=False, + OpticalBlank=False, + OpticalNumTracks=0, + OpticalNumAudioTracks=0, + OpticalNumDataTracks=0, + OpticalNumSessions=0, + RotationRate=0, + ConnectionBus="usb", + Seat="seat0", + Removable=True, + Ejectable=False, + SortKey="01hotplug/1675897304240492", + CanPowerOff=True, + SiblingId="/sys/devices/platform/soc/ffe09000.usb/ff500000.usb/xhci-hcd.1.auto/usb2/2-1/2-1.4/2-1.4:1.0", + ), +} + + +class Drive(DBusServiceMock): + """Drive mock. + + gdbus introspect --system --dest org.freedesktop.UDisks2 --object-path /org/freedesktop/UDisks2/drives/id + """ + + interface = "org.freedesktop.UDisks2.Drive" + + def __init__(self, object_path: str): + """Initialize object.""" + super().__init__() + self.object_path = object_path + self.fixture: DriveFixture = FIXTURES[object_path] + + @dbus_property(access=PropertyAccess.READ) + def Vendor(self) -> "s": + """Get Vendor.""" + return self.fixture.Vendor + + @dbus_property(access=PropertyAccess.READ) + def Model(self) -> "s": + """Get Model.""" + return self.fixture.Model + + @dbus_property(access=PropertyAccess.READ) + def Revision(self) -> "s": + """Get Revision.""" + return self.fixture.Revision + + @dbus_property(access=PropertyAccess.READ) + def Serial(self) -> "s": + """Get Serial.""" + return self.fixture.Serial + + @dbus_property(access=PropertyAccess.READ) + def WWN(self) -> "s": + """Get WWN.""" + return self.fixture.WWN + + @dbus_property(access=PropertyAccess.READ) + def Id(self) -> "s": + """Get Id.""" + return self.fixture.Id + + @dbus_property(access=PropertyAccess.READ) + def Configuration(self) -> "a{sv}": + """Get Configuration.""" + return self.fixture.Configuration + + @dbus_property(access=PropertyAccess.READ) + def Media(self) -> "s": + """Get Media.""" + return self.fixture.Media + + @dbus_property(access=PropertyAccess.READ) + def MediaCompatibility(self) -> "as": + """Get MediaCompatibility.""" + return self.fixture.MediaCompatibility + + @dbus_property(access=PropertyAccess.READ) + def MediaRemovable(self) -> "b": + """Get MediaRemovable.""" + return self.fixture.MediaRemovable + + @dbus_property(access=PropertyAccess.READ) + def MediaAvailable(self) -> "b": + """Get MediaAvailable.""" + return self.fixture.MediaAvailable + + @dbus_property(access=PropertyAccess.READ) + def MediaChangeDetected(self) -> "b": + """Get MediaChangeDetected.""" + return self.fixture.MediaChangeDetected + + @dbus_property(access=PropertyAccess.READ) + def Size(self) -> "t": + """Get Size.""" + return self.fixture.Size + + @dbus_property(access=PropertyAccess.READ) + def TimeDetected(self) -> "t": + """Get TimeDetected.""" + return self.fixture.TimeDetected + + @dbus_property(access=PropertyAccess.READ) + def TimeMediaDetected(self) -> "t": + """Get TimeMediaDetected.""" + return self.fixture.TimeMediaDetected + + @dbus_property(access=PropertyAccess.READ) + def Optical(self) -> "b": + """Get Optical.""" + return self.fixture.Optical + + @dbus_property(access=PropertyAccess.READ) + def OpticalBlank(self) -> "b": + """Get OpticalBlank.""" + return self.fixture.OpticalBlank + + @dbus_property(access=PropertyAccess.READ) + def OpticalNumTracks(self) -> "u": + """Get OpticalNumTracks.""" + return self.fixture.OpticalNumTracks + + @dbus_property(access=PropertyAccess.READ) + def OpticalNumAudioTracks(self) -> "u": + """Get OpticalNumAudioTracks.""" + return self.fixture.OpticalNumAudioTracks + + @dbus_property(access=PropertyAccess.READ) + def OpticalNumDataTracks(self) -> "u": + """Get OpticalNumDataTracks.""" + return self.fixture.OpticalNumDataTracks + + @dbus_property(access=PropertyAccess.READ) + def OpticalNumSessions(self) -> "u": + """Get OpticalNumSessions.""" + return self.fixture.OpticalNumSessions + + @dbus_property(access=PropertyAccess.READ) + def RotationRate(self) -> "i": + """Get RotationRate.""" + return self.fixture.RotationRate + + @dbus_property(access=PropertyAccess.READ) + def ConnectionBus(self) -> "s": + """Get ConnectionBus.""" + return self.fixture.ConnectionBus + + @dbus_property(access=PropertyAccess.READ) + def Seat(self) -> "s": + """Get Seat.""" + return self.fixture.Seat + + @dbus_property(access=PropertyAccess.READ) + def Removable(self) -> "b": + """Get Removable.""" + return self.fixture.Removable + + @dbus_property(access=PropertyAccess.READ) + def Ejectable(self) -> "b": + """Get Ejectable.""" + return self.fixture.Ejectable + + @dbus_property(access=PropertyAccess.READ) + def SortKey(self) -> "s": + """Get SortKey.""" + return self.fixture.SortKey + + @dbus_property(access=PropertyAccess.READ) + def CanPowerOff(self) -> "b": + """Get CanPowerOff.""" + return self.fixture.CanPowerOff + + @dbus_property(access=PropertyAccess.READ) + def SiblingId(self) -> "s": + """Get SiblingId.""" + return self.fixture.SiblingId + + @dbus_method() + def Eject(self, options: "a{sv}") -> None: + """Do Eject method.""" + + @dbus_method() + def SetConfiguration(self, value: "a{sv}", options: "a{sv}") -> None: + """Do SetConfiguration method.""" + + @dbus_method() + def PowerOff(self, options: "a{sv}") -> None: + """Do PowerOff method.""" diff --git a/tests/dbus_service_mocks/udisks2_filesystem.py b/tests/dbus_service_mocks/udisks2_filesystem.py new file mode 100644 index 000000000..ef8acf81e --- /dev/null +++ b/tests/dbus_service_mocks/udisks2_filesystem.py @@ -0,0 +1,118 @@ +"""Mock of UDisks2 Filesystem service.""" + +from ctypes import c_uint64 +from dataclasses import dataclass + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.UDisks2" +DEFAULT_OBJECT_PATH = "/org/freedesktop/UDisks2/block_devices/sda1" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Filesystem(object_path if object_path else DEFAULT_OBJECT_PATH) + + +# pylint: disable=invalid-name + + +@dataclass(slots=True) +class FilesystemFixture: + """Filesystem fixture.""" + + MountPoints: list[bytes] + Size: c_uint64 + + +FIXTURES: dict[str, FilesystemFixture] = { + "/org/freedesktop/UDisks2/block_devices/mmcblk1p1": FilesystemFixture( + MountPoints=[b"/mnt/boot"], Size=0 + ), + "/org/freedesktop/UDisks2/block_devices/mmcblk1p3": FilesystemFixture( + MountPoints=[ + b"/etc/NetworkManager/system-connections", + b"/etc/dropbear", + b"/etc/hostname", + b"/etc/hosts", + b"/etc/modprobe.d", + b"/etc/modules-load.d", + b"/etc/systemd/timesyncd.conf", + b"/etc/udev/rules.d", + b"/mnt/overlay", + b"/root/.docker", + b"/root/.ssh", + b"/var/lib/NetworkManager", + b"/var/lib/bluetooth", + b"/var/lib/systemd", + ], + Size=100663296, + ), + "/org/freedesktop/UDisks2/block_devices/sda1": FilesystemFixture( + MountPoints=[], Size=250058113024 + ), + "/org/freedesktop/UDisks2/block_devices/sdb1": FilesystemFixture( + MountPoints=[b"/mnt/data/supervisor/media/ext"], Size=67108864 + ), + "/org/freedesktop/UDisks2/block_devices/zram1": FilesystemFixture( + MountPoints=[b"/var"], Size=0 + ), +} + + +class Filesystem(DBusServiceMock): + """Filesystem mock. + + gdbus introspect --system --dest org.freedesktop.UDisks2 --object-path /org/freedesktop/UDisks2/block_devices/sda1 + """ + + interface = "org.freedesktop.UDisks2.Filesystem" + + def __init__(self, object_path: str): + """Initialize object.""" + super().__init__() + self.object_path = object_path + self.fixture: FilesystemFixture = FIXTURES[object_path] + + @dbus_property(access=PropertyAccess.READ) + def MountPoints(self) -> "aay": + """Get MountPoints.""" + return self.fixture.MountPoints + + @dbus_property(access=PropertyAccess.READ) + def Size(self) -> "t": + """Get Size.""" + return self.fixture.Size + + @dbus_method() + def SetLabel(self, label: "s", options: "a{sv}") -> None: + """Do SetLabel method.""" + + @dbus_method() + def Mount(self, options: "a{sv}") -> "s": + """Do Mount method.""" + return "/run/media/dev/hassos_data" + + @dbus_method() + def Unmount(self, options: "a{sv}") -> None: + """Do Unmount method.""" + + @dbus_method() + def Resize(self, size: "t", options: "a{sv}") -> None: + """Do Resize method.""" + + @dbus_method() + def Check(self, options: "a{sv}") -> "b": + """Do Check method.""" + return True + + @dbus_method() + def Repair(self, options: "a{sv}") -> "b": + """Do Repair method.""" + return True + + @dbus_method() + def TakeOwnership(self, options: "a{sv}") -> None: + """Do TakeOwnership method.""" diff --git a/tests/dbus_service_mocks/udisks2_loop.py b/tests/dbus_service_mocks/udisks2_loop.py new file mode 100644 index 000000000..6ba275d8d --- /dev/null +++ b/tests/dbus_service_mocks/udisks2_loop.py @@ -0,0 +1,48 @@ +"""Mock of UDisks2 Loop service.""" + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.UDisks2" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Loop() + + +# pylint: disable=invalid-name + + +class Loop(DBusServiceMock): + """Loop mock. + + gdbus introspect --system --dest org.freedesktop.UDisks2 --object-path /org/freedesktop/UDisks2/block_devices/loop0 + """ + + interface = "org.freedesktop.UDisks2.Loop" + object_path = "/org/freedesktop/UDisks2/block_devices/loop0" + + @dbus_property(access=PropertyAccess.READ) + def BackingFile(self) -> "ay": + """Get BackingFile.""" + return b"" + + @dbus_property(access=PropertyAccess.READ) + def Autoclear(self) -> "b": + """Get Autoclear.""" + return False + + @dbus_property(access=PropertyAccess.READ) + def SetupByUID(self) -> "u": + """Get SetupByUID.""" + return 0 + + @dbus_method() + def Delete(self, options: "a{sv}") -> None: + """Do Delete method.""" + + @dbus_method() + def SetAutoClear(self, value: "b", options: "a{sv}") -> None: + """Do SetAutoClear method.""" diff --git a/tests/dbus_service_mocks/udisks2_manager.py b/tests/dbus_service_mocks/udisks2_manager.py new file mode 100644 index 000000000..b00605278 --- /dev/null +++ b/tests/dbus_service_mocks/udisks2_manager.py @@ -0,0 +1,102 @@ +"""Mock of UDisks2 Manager service.""" + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.UDisks2" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return UDisks2Manager() + + +# pylint: disable=invalid-name + + +class UDisks2Manager(DBusServiceMock): + """UDisks2 Manager mock. + + gdbus introspect --system --dest org.freedesktop.UDisks2 --object-path /org/freedesktop/UDisks2/Manager + """ + + interface = "org.freedesktop.UDisks2.Manager" + object_path = "/org/freedesktop/UDisks2/Manager" + + @dbus_property(access=PropertyAccess.READ) + def Version(self) -> "s": + """Get Version.""" + return "2.9.2" + + @dbus_property(access=PropertyAccess.READ) + def SupportedFilesystems(self) -> "as": + """Get SupportedFilesystems.""" + return ["ext4", "vfat", "ntfs", "exfat", "swap"] + + @dbus_property(access=PropertyAccess.READ) + def SupportedEncryptionTypes(self) -> "as": + """Get SupportedEncryptionTypes.""" + return ["luks1", "luks2"] + + @dbus_property(access=PropertyAccess.READ) + def DefaultEncryptionType(self) -> "s": + """Get DefaultEncryptionType.""" + return "luks1" + + @dbus_method() + def CanFormat(self, type_: "s") -> "(bs)": + """Do CanFormat method.""" + return [False, "mkfs.ntfs"] + + @dbus_method() + def CanResize(self, type_: "s") -> "(bts)": + """Do CanResize method.""" + return [False, 6, "ntfsresize"] + + @dbus_method() + def CanCheck(self, type_: "s") -> "(bs)": + """Do CanCheck method.""" + return [False, "ntfsfix"] + + @dbus_method() + def CanRepair(self, type_: "s") -> "(bs)": + """Do CanRepair method.""" + return [False, "ntfsfix"] + + @dbus_method() + def LoopSetup(self, fd: "h", options: "a{sv}") -> "o": + """Do LoopSetup method.""" + return "/org/freedesktop/UDisks2/block_devices/loop0" + + @dbus_method() + def MDRaidCreate( + self, blocks: "ao", level: "s", name: "s", chunk: "t", options: "a{sv}" + ) -> "o": + """Do MDRaidCreate method.""" + return "/org/freedesktop/UDisks2/block_devices/sdb" + + @dbus_method() + def EnableModule(self, name: "s", enable: "b") -> None: + """Do EnableModule method.""" + + @dbus_method() + def GetBlockDevices(self, options: "a{sv}") -> "ao": + """Do GetBlockDevices method.""" + return [ + "/org/freedesktop/UDisks2/block_devices/loop0", + "/org/freedesktop/UDisks2/block_devices/mmcblk1", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p1", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p2", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p3", + "/org/freedesktop/UDisks2/block_devices/sda", + "/org/freedesktop/UDisks2/block_devices/sda1", + "/org/freedesktop/UDisks2/block_devices/sdb", + "/org/freedesktop/UDisks2/block_devices/sdb1", + "/org/freedesktop/UDisks2/block_devices/zram1", + ] + + @dbus_method() + def ResolveDevice(self, devspec: "a{sv}", options: "a{sv}") -> "ao": + """Do ResolveDevice method.""" + return ["/org/freedesktop/UDisks2/block_devices/sda1"] diff --git a/tests/dbus_service_mocks/udisks2_partition.py b/tests/dbus_service_mocks/udisks2_partition.py new file mode 100644 index 000000000..d1d0ffa01 --- /dev/null +++ b/tests/dbus_service_mocks/udisks2_partition.py @@ -0,0 +1,184 @@ +"""Mock of UDisks2 Partition service.""" + +from ctypes import c_uint32, c_uint64 +from dataclasses import dataclass + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.UDisks2" +DEFAULT_OBJECT_PATH = "/org/freedesktop/UDisks2/block_devices/sda1" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return Partition(object_path if object_path else DEFAULT_OBJECT_PATH) + + +# pylint: disable=invalid-name + + +@dataclass(slots=True) +class PartitionFixture: + """Partition fixture.""" + + Number: c_uint32 + Type: str + Flags: c_uint64 + Offset: c_uint64 + Size: c_uint64 + Name: str + UUID: str + Table: str + IsContainer: bool + IsContained: bool + + +FIXTURES: dict[str, PartitionFixture] = { + "/org/freedesktop/UDisks2/block_devices/mmcblk1p1": PartitionFixture( + Number=1, + Type="0x0c", + Flags=128, + Offset=8388608, + Size=25165824, + Name="", + UUID="48617373-01", + Table="/org/freedesktop/UDisks2/block_devices/mmcblk1", + IsContainer=False, + IsContained=False, + ), + "/org/freedesktop/UDisks2/block_devices/mmcblk1p2": PartitionFixture( + Number=2, + Type="0x05", + Flags=0, + Offset=33554432, + Size=600834048, + Name="", + UUID="48617373-02", + Table="/org/freedesktop/UDisks2/block_devices/mmcblk1", + IsContainer=True, + IsContained=False, + ), + "/org/freedesktop/UDisks2/block_devices/mmcblk1p3": PartitionFixture( + Number=3, + Type="0x83", + Flags=0, + Offset=635437056, + Size=100663296, + Name="", + UUID="48617373-03", + Table="/org/freedesktop/UDisks2/block_devices/mmcblk1", + IsContainer=False, + IsContained=False, + ), + "/org/freedesktop/UDisks2/block_devices/sda1": PartitionFixture( + Number=1, + Type="0fc63daf-8483-4772-8e79-3d69d8477de4", + Flags=0, + Offset=1048576, + Size=250058113024, + Name="hassos-data-external", + UUID="6f3f99f4-4d34-476b-b051-77886da57fa9", + Table="/org/freedesktop/UDisks2/block_devices/sda", + IsContainer=False, + IsContained=False, + ), + "/org/freedesktop/UDisks2/block_devices/sdb1": PartitionFixture( + Number=1, + Type="0x0c", + Flags=128, + Offset=1048576, + Size=67108864, + Name="", + UUID="", + Table="/org/freedesktop/UDisks2/block_devices/sdb", + IsContainer=False, + IsContained=False, + ), +} + + +class Partition(DBusServiceMock): + """Block mock. + + gdbus introspect --system --dest org.freedesktop.UDisks2 --object-path /org/freedesktop/UDisks2/block_devices/sda1 + """ + + interface = "org.freedesktop.UDisks2.Partition" + + def __init__(self, object_path: str): + """Initialize object.""" + super().__init__() + self.object_path = object_path + self.fixture: PartitionFixture = FIXTURES[object_path] + + @dbus_property(access=PropertyAccess.READ) + def Number(self) -> "u": + """Get Number.""" + return self.fixture.Number + + @dbus_property(access=PropertyAccess.READ) + def Type(self) -> "s": + """Get Type.""" + return self.fixture.Type + + @dbus_property(access=PropertyAccess.READ) + def Flags(self) -> "t": + """Get Flags.""" + return self.fixture.Flags + + @dbus_property(access=PropertyAccess.READ) + def Offset(self) -> "t": + """Get Offset.""" + return self.fixture.Offset + + @dbus_property(access=PropertyAccess.READ) + def Size(self) -> "t": + """Get Size.""" + return self.fixture.Size + + @dbus_property(access=PropertyAccess.READ) + def Name(self) -> "s": + """Get Name.""" + return self.fixture.Name + + @dbus_property(access=PropertyAccess.READ) + def UUID(self) -> "s": + """Get UUID.""" + return self.fixture.UUID + + @dbus_property(access=PropertyAccess.READ) + def Table(self) -> "o": + """Get Table.""" + return self.fixture.Table + + @dbus_property(access=PropertyAccess.READ) + def IsContainer(self) -> "b": + """Get IsContainer.""" + return self.fixture.IsContainer + + @dbus_property(access=PropertyAccess.READ) + def IsContained(self) -> "b": + """Get IsContained.""" + return self.fixture.IsContained + + @dbus_method() + def SetType(self, type_: "s", options: "a{sv}") -> None: + """Do SetType method.""" + + @dbus_method() + def SetName(self, name: "s", options: "a{sv}") -> None: + """Do SetName method.""" + + @dbus_method() + def SetFlags(self, flags: "t", options: "a{sv}") -> None: + """Do SetFlags method.""" + + @dbus_method() + def Resize(self, size: "t", options: "a{sv}") -> None: + """Do Resize method.""" + + @dbus_method() + def Delete(self, options: "a{sv}") -> None: + """Do Delete method.""" diff --git a/tests/dbus_service_mocks/udisks2_partition_table.py b/tests/dbus_service_mocks/udisks2_partition_table.py new file mode 100644 index 000000000..9091c2a5b --- /dev/null +++ b/tests/dbus_service_mocks/udisks2_partition_table.py @@ -0,0 +1,90 @@ +"""Mock of UDisks2 Partition Table service.""" + +from dataclasses import dataclass + +from dbus_fast.service import PropertyAccess, dbus_property + +from .base import DBusServiceMock, dbus_method + +BUS_NAME = "org.freedesktop.UDisks2" +DEFAULT_OBJECT_PATH = "/org/freedesktop/UDisks2/block_devices/sda" + + +def setup(object_path: str | None = None) -> DBusServiceMock: + """Create dbus mock object.""" + return PartitionTable(object_path if object_path else DEFAULT_OBJECT_PATH) + + +# pylint: disable=invalid-name + + +@dataclass(slots=True) +class PartitionTableFixture: + """PartitionTable fixture.""" + + Partitions: list[str] + Type: str + + +FIXTURES: dict[str, PartitionTableFixture] = { + "/org/freedesktop/UDisks2/block_devices/mmcblk1": PartitionTableFixture( + Partitions=[ + "/org/freedesktop/UDisks2/block_devices/mmcblk1p1", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p2", + "/org/freedesktop/UDisks2/block_devices/mmcblk1p3", + ], + Type="dos", + ), + "/org/freedesktop/UDisks2/block_devices/sda": PartitionTableFixture( + Partitions=["/org/freedesktop/UDisks2/block_devices/sda1"], Type="gpt" + ), + "/org/freedesktop/UDisks2/block_devices/sdb": PartitionTableFixture( + Partitions=["/org/freedesktop/UDisks2/block_devices/sdb1"], Type="gpt" + ), +} + + +class PartitionTable(DBusServiceMock): + """PartitionTable mock. + + gdbus introspect --system --dest org.freedesktop.UDisks2 --object-path /org/freedesktop/UDisks2/block_devices/sda + """ + + interface = "org.freedesktop.UDisks2.PartitionTable" + + def __init__(self, object_path: str): + """Initialize object.""" + super().__init__() + self.object_path = object_path + self.fixture: PartitionTableFixture = FIXTURES[object_path] + + @dbus_property(access=PropertyAccess.READ) + def Partitions(self) -> "ao": + """Get Partitions.""" + return self.fixture.Partitions + + @dbus_property(access=PropertyAccess.READ) + def Type(self) -> "s": + """Get Type.""" + return self.fixture.Type + + @dbus_method() + def CreatePartition( + self, offset: "t", size: "t", type_: "s", name: "s", options: "a{sv}" + ) -> "o": + """Do CreatePartition method.""" + return "/org/freedesktop/UDisks2/block_devices/sda2" + + @dbus_method() + def CreatePartitionAndFormat( + self, + offset: "t", + size: "t", + type_: "s", + name: "s", + options: "a{sv}", + format_type_: "s", + format_options: "a{sv}", + ) -> "o": + """Do CreatePartitionAndFormat method.""" + return "/org/freedesktop/UDisks2/block_devices/sda2"