jeans/supervisor
jeans/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-07-19 15:16:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Use session dbus for udisks2 tests (#4194)

Browse Source 
* Use session dbus for udisks2 tests

* Don't redefine type
This commit is contained in:
Mike Degatano 2023-03-17 15:56:34 -04:00 committed by GitHub
parent 0ddce4d9bc
commit d9b86fa2ab
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 1817 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
tests/common.py
View File

@ -108,12 +108,17 @@ def exists_fixture(filename: str) -> bool:
async def mock_dbus_services( async def mock_dbus_services(
to_mock: dict[str, list[str] | str | None], bus: MessageBus 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. """Mock specified dbus services on bus.
to_mock is dictionary where the key is a dbus service to mock (module must exist 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 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). 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] = {} services: dict[str, list[DBusServiceMock] | DBusServiceMock] = {}
requested_names: set[str] = set() requested_names: set[str] = set()
@ -127,10 +132,10 @@ async def mock_dbus_services(
requested_names.add(service_module.BUS_NAME) requested_names.add(service_module.BUS_NAME)
if isinstance(to_mock[module], list): if isinstance(to_mock[module], list):
services[module] = [ services[module] = {
service_module.setup(obj_path).export(bus) obj_path: service_module.setup(obj_path).export(bus)
for obj_path in to_mock[module] for obj_path in to_mock[module]
] }
else: else:
services[module] = service_module.setup(to_mock[module]).export(bus) services[module] = service_module.setup(to_mock[module]).export(bus)

64
tests/conftest.py
View File

@ -147,12 +147,11 @@ async def dbus_session_bus(dbus_session) -> MessageBus:
@pytest.fixture @pytest.fixture
async def dbus_services( async def dbus_services(
request: pytest.FixtureRequest, dbus_session: MessageBus request: pytest.FixtureRequest, dbus_session: MessageBus
) -> dict[str, DBusServiceMock | list[DBusServiceMock]]: ) -> dict[str, dict[str, DBusServiceMock] | DBusServiceMock]:
"""Mock specified dbus services on session bus. """Mock specified dbus services on session bus.
Should be used indirectly. Provide a dictionary where the key a dbus service to mock Wrapper on mock_dbus_services intended to be used indirectly.
(module must exist in dbus_service_mocks). Value is the object path for the mocked service. Request param passed to it as to_mock and output returned as fixture value.
Can also be a list of object paths or None (if the mocked service defines the object path).
""" """
with patch("supervisor.dbus.manager.MessageBus.connect", return_value=dbus_session): with patch("supervisor.dbus.manager.MessageBus.connect", return_value=dbus_session):
yield await mock_dbus_services(request.param, 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") @pytest.fixture(name="network_manager_services")
async def fixture_network_manager_services( async def fixture_network_manager_services(
dbus_session_bus: MessageBus, dbus_session_bus: MessageBus,
) -> dict[str, DBusServiceMock]: ) -> dict[str, DBusServiceMock | dict[str, DBusServiceMock]]:
"""Mock all services network manager connects to.""" """Mock all services network manager connects to."""
yield await mock_dbus_services( yield await mock_dbus_services(
{ {
@ -357,14 +356,65 @@ async def fixture_network_manager_services(
@pytest.fixture @pytest.fixture
async def network_manager( 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: ) -> NetworkManager:
"""Mock NetworkManager.""" """Mock Network Manager."""
nm_obj = NetworkManager() nm_obj = NetworkManager()
await nm_obj.connect(dbus_session_bus) await nm_obj.connect(dbus_session_bus)
yield nm_obj 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( async def mock_dbus_interface(
dbus: DBus, dbus_bus: MessageBus, instance: DBusInterface dbus: DBus, dbus_bus: MessageBus, instance: DBusInterface
) -> DBusInterface: ) -> DBusInterface:

2
tests/dbus/network/setting/test_init.py
View File

@ -18,7 +18,7 @@ from tests.dbus_service_mocks.network_connection_settings import (
@pytest.fixture(name="connection_settings_service", autouse=True) @pytest.fixture(name="connection_settings_service", autouse=True)
async def fixture_connection_settings_service( async def fixture_connection_settings_service(
network_manager_services: dict[str, DBusServiceMock] network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]]
) -> ConnectionSettingsService: ) -> ConnectionSettingsService:
"""Mock Connection Settings service.""" """Mock Connection Settings service."""
yield network_manager_services["network_connection_settings"] yield network_manager_services["network_connection_settings"]

2
tests/dbus/network/test_connection.py
View File

@ -16,7 +16,7 @@ from tests.dbus_service_mocks.network_active_connection import (
@pytest.fixture(name="active_connection_service", autouse=True) @pytest.fixture(name="active_connection_service", autouse=True)
async def fixture_active_connection_service( async def fixture_active_connection_service(
network_manager_services: dict[str, DBusServiceMock] network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]]
) -> ActiveConnectionService: ) -> ActiveConnectionService:
"""Mock Active Connection service.""" """Mock Active Connection service."""
yield network_manager_services["network_active_connection"] yield network_manager_services["network_active_connection"]

12
tests/dbus/network/test_interface.py
View File

@ -17,18 +17,22 @@ from tests.dbus_service_mocks.network_device import Device as DeviceService
@pytest.fixture(name="device_eth0_service") @pytest.fixture(name="device_eth0_service")
async def fixture_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: ) -> DeviceService:
"""Mock Device eth0 service.""" """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") @pytest.fixture(name="device_wlan0_service")
async def fixture_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: ) -> DeviceService:
"""Mock Device wlan0 service.""" """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") @pytest.fixture(name="device_unmanaged_service")

2
tests/dbus/network/test_network_manager.py
View File

@ -22,7 +22,7 @@ from tests.dbus_service_mocks.network_manager import (
@pytest.fixture(name="network_manager_service", autouse=True) @pytest.fixture(name="network_manager_service", autouse=True)
async def fixture_network_manager_service( async def fixture_network_manager_service(
network_manager_services: dict[str, DBusServiceMock], network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]],
) -> NetworkManagerService: ) -> NetworkManagerService:
"""Mock NetworkManager dbus service.""" """Mock NetworkManager dbus service."""
yield network_manager_services["network_manager"] yield network_manager_services["network_manager"]

2
tests/dbus/network/test_wireless.py
View File

@ -15,7 +15,7 @@ from tests.dbus_service_mocks.network_device_wireless import (
@pytest.fixture(name="device_wireless_service", autouse=True) @pytest.fixture(name="device_wireless_service", autouse=True)
async def fixture_device_wireless_service( async def fixture_device_wireless_service(
network_manager_services: dict[str, DBusServiceMock] network_manager_services: dict[str, DBusServiceMock | dict[str, DBusServiceMock]]
) -> DeviceWirelessService: ) -> DeviceWirelessService:
"""Mock Device Wireless service.""" """Mock Device Wireless service."""
yield network_manager_services["network_device_wireless"] yield network_manager_services["network_device_wireless"]

64
tests/dbus/udisks2/test_block.py
View File

@ -1,8 +1,8 @@
"""Test UDisks2 Block Device interface.""" """Test UDisks2 Block Device interface."""
import asyncio
from pathlib import Path from pathlib import Path
from dbus_fast import Variant
from dbus_fast.aio.message_bus import MessageBus from dbus_fast.aio.message_bus import MessageBus
import pytest 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.const import FormatType, PartitionTableType
from supervisor.dbus.udisks2.data import FormatOptions 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.""" """Test block device info."""
sda = UDisks2Block("/org/freedesktop/UDisks2/block_devices/sda") sda = UDisks2Block("/org/freedesktop/UDisks2/block_devices/sda")
sda1 = UDisks2Block( 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.symlinks is None
assert sda1.filesystem is None assert sda1.filesystem is None
await sda.connect(dbus_bus) await sda.connect(dbus_session_bus)
await sda1.connect(dbus_bus) await sda1.connect(dbus_session_bus)
assert sda.drive == "/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56" assert sda.drive == "/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56"
assert sda.device == Path("/dev/sda") 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.partition_table is None
assert sda1.filesystem.mount_points == [] assert sda1.filesystem.mount_points == []
fire_property_change_signal(sda, {"IdLabel": "test"}) block_sda_service.emit_properties_changed({"IdLabel": "test"})
await asyncio.sleep(0) await block_sda_service.ping()
assert sda.id_label == "test" assert sda.id_label == "test"
with pytest.raises(AssertionError): block_sda_service.emit_properties_changed({}, ["IdLabel"])
fire_property_change_signal(sda1) 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.""" """Test formatting block device."""
sda = UDisks2Block("/org/freedesktop/UDisks2/block_devices/sda") 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")) await sda.format(FormatType.GPT, FormatOptions(label="test"))
assert dbus == [ assert block_sda_service.Format.calls == [
"/org/freedesktop/UDisks2/block_devices/sda-org.freedesktop.UDisks2.Block.Format" (
"gpt",
{
"label": Variant("s", "test"),
"auth.no_user_interaction": Variant("b", True),
},
)
] ]

61
tests/dbus/udisks2/test_drive.py
View File

@ -1,16 +1,50 @@
"""Test UDisks2 Drive.""" """Test UDisks2 Drive."""
import asyncio
from datetime import datetime, timezone from datetime import datetime, timezone
from dbus_fast import Variant
from dbus_fast.aio.message_bus import MessageBus from dbus_fast.aio.message_bus import MessageBus
import pytest
from supervisor.dbus.udisks2.drive import UDisks2Drive 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.""" """Test drive info."""
ssk = UDisks2Drive("/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56") 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.time_detected is None
assert ssk.ejectable is None assert ssk.ejectable is None
await ssk.connect(dbus_bus) await ssk.connect(dbus_session_bus)
assert ssk.vendor == "SSK" assert ssk.vendor == "SSK"
assert ssk.model == "SSK Storage" 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.time_detected == datetime(2023, 2, 8, 23, 1, 44, 240492, timezone.utc)
assert ssk.ejectable is False assert ssk.ejectable is False
fire_property_change_signal(ssk, {"Ejectable": True}) drive_ssk_storage_service.emit_properties_changed({"Ejectable": True})
await asyncio.sleep(0) await drive_ssk_storage_service.ping()
assert ssk.ejectable is True 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.""" """Test eject."""
flash = UDisks2Drive("/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6") flash = UDisks2Drive("/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6")
await flash.connect(dbus_bus) await flash.connect(dbus_session_bus)
await flash.eject() await flash.eject()
assert dbus == [ assert drive_flash_disk_service.Eject.calls == [
"/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6-org.freedesktop.UDisks2.Drive.Eject" ({"auth.no_user_interaction": Variant("b", True)},)
] ]

103
tests/dbus/udisks2/test_filesystem.py
View File

@ -1,26 +1,59 @@
"""Test UDisks2 Filesystem.""" """Test UDisks2 Filesystem."""
import asyncio
from pathlib import Path from pathlib import Path
from dbus_fast import Variant
from dbus_fast.aio.message_bus import MessageBus from dbus_fast.aio.message_bus import MessageBus
import pytest import pytest
from supervisor.dbus.udisks2.data import MountOptions, UnmountOptions from supervisor.dbus.udisks2.data import MountOptions, UnmountOptions
from supervisor.dbus.udisks2.filesystem import UDisks2Filesystem 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") @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.""" """Return connected sda1 filesystem object."""
filesystem = UDisks2Filesystem("/org/freedesktop/UDisks2/block_devices/sda1") filesystem = UDisks2Filesystem("/org/freedesktop/UDisks2/block_devices/sda1")
await filesystem.connect(dbus_bus) await filesystem.connect(dbus_session_bus)
return filesystem 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.""" """Test filesystem info."""
sda1 = UDisks2Filesystem("/org/freedesktop/UDisks2/block_devices/sda1") sda1 = UDisks2Filesystem("/org/freedesktop/UDisks2/block_devices/sda1")
sdb1 = UDisks2Filesystem( sdb1 = UDisks2Filesystem(
@ -32,53 +65,69 @@ async def test_filesystem_info(dbus: list[str], dbus_bus: MessageBus):
assert sdb1.size is None assert sdb1.size is None
assert sdb1.mount_points is None assert sdb1.mount_points is None
await sda1.connect(dbus_bus) await sda1.connect(dbus_session_bus)
await sdb1.connect(dbus_bus) await sdb1.connect(dbus_session_bus)
assert sda1.size == 250058113024 assert sda1.size == 250058113024
assert sda1.mount_points == [] assert sda1.mount_points == []
assert sdb1.size == 67108864 assert sdb1.size == 67108864
assert sdb1.mount_points == [Path("/mnt/data/supervisor/media/ext")] assert sdb1.mount_points == [Path("/mnt/data/supervisor/media/ext")]
fire_property_change_signal( filesystem_sda1_service.emit_properties_changed({"MountPoints": [b"/mnt/media"]})
sda1, {"MountPoints": [bytearray("/mnt/media", encoding="utf-8")]} await filesystem_sda1_service.ping()
)
await asyncio.sleep(0)
assert sda1.mount_points == [Path("/mnt/media")] assert sda1.mount_points == [Path("/mnt/media")]
with pytest.raises(AssertionError): filesystem_sda1_service.emit_properties_changed({}, ["MountPoints"])
fire_property_change_signal( await filesystem_sda1_service.ping()
sdb1, {"MountPoints": [bytearray("/mnt/media", encoding="utf-8")]} 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.""" """Test mount."""
assert await sda1.mount(MountOptions(fstype="gpt")) == "/run/media/dev/hassos_data" assert await sda1.mount(MountOptions(fstype="gpt")) == "/run/media/dev/hassos_data"
assert dbus == [ assert filesystem_sda1_service.Mount.calls == [
"/org/freedesktop/UDisks2/block_devices/sda1-org.freedesktop.UDisks2.Filesystem.Mount" (
{
"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.""" """Test unmount."""
await sda1.unmount(UnmountOptions(force=True)) await sda1.unmount(UnmountOptions(force=True))
assert dbus == [ assert filesystem_sda1_service.Unmount.calls == [
"/org/freedesktop/UDisks2/block_devices/sda1-org.freedesktop.UDisks2.Filesystem.Unmount" ({"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.""" """Test check."""
assert await sda1.check() is True assert await sda1.check() is True
assert dbus == [ assert filesystem_sda1_service.Check.calls == [
"/org/freedesktop/UDisks2/block_devices/sda1-org.freedesktop.UDisks2.Filesystem.Check" ({"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.""" """Test repair."""
assert await sda1.repair() is True assert await sda1.repair() is True
assert dbus == [ assert filesystem_sda1_service.Repair.calls == [
"/org/freedesktop/UDisks2/block_devices/sda1-org.freedesktop.UDisks2.Filesystem.Repair" ({"auth.no_user_interaction": Variant("b", True)},)
] ]

121
tests/dbus/udisks2/test_manager.py
View File

@ -1,33 +1,48 @@
"""Test UDisks2 Manager interface.""" """Test UDisks2 Manager interface."""
import asyncio
from awesomeversion import AwesomeVersion from awesomeversion import AwesomeVersion
from dbus_fast import Variant
from dbus_fast.aio.message_bus import MessageBus
import pytest import pytest
from supervisor.coresys import CoreSys from supervisor.dbus.udisks2 import UDisks2
from supervisor.dbus.udisks2.data import DeviceSpecification from supervisor.dbus.udisks2.data import DeviceSpecification
from supervisor.exceptions import DBusNotConnectedError, DBusObjectError 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.""" """Test udisks2 manager dbus connection."""
dbus.clear() udisks2_manager_service.GetBlockDevices.calls.clear()
assert coresys.dbus.udisks2.supported_filesystems is None 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", "ext4",
"vfat", "vfat",
"ntfs", "ntfs",
"exfat", "exfat",
"swap", "swap",
] ]
assert coresys.dbus.udisks2.version == AwesomeVersion("2.9.2") assert udisks2.version == AwesomeVersion("2.9.2")
assert {block.object_path for block in coresys.dbus.udisks2.block_devices} == { assert {block.object_path for block in udisks2.block_devices} == {
"/org/freedesktop/UDisks2/block_devices/loop0", "/org/freedesktop/UDisks2/block_devices/loop0",
"/org/freedesktop/UDisks2/block_devices/mmcblk1", "/org/freedesktop/UDisks2/block_devices/mmcblk1",
"/org/freedesktop/UDisks2/block_devices/mmcblk1p1", "/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/sdb1",
"/org/freedesktop/UDisks2/block_devices/zram1", "/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/BJTD4R_0x97cde291",
"/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56", "/org/freedesktop/UDisks2/drives/SSK_SSK_Storage_DF56419883D56",
"/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6", "/org/freedesktop/UDisks2/drives/Generic_Flash_Disk_61BCDDB6",
} }
assert dbus == [ assert udisks2_manager_service.GetBlockDevices.calls == [
"/org/freedesktop/UDisks2/Manager-org.freedesktop.UDisks2.Manager.GetBlockDevices" ({"auth.no_user_interaction": Variant("b", True)},)
] ]
dbus.clear() udisks2_manager_service.GetBlockDevices.calls.clear()
fire_property_change_signal( udisks2_manager_service.emit_properties_changed({"SupportedFilesystems": ["ext4"]})
coresys.dbus.udisks2, {"SupportedFilesystems": ["ext4"]} await udisks2_manager_service.ping()
) assert udisks2.supported_filesystems == ["ext4"]
await asyncio.sleep(0) assert udisks2_manager_service.GetBlockDevices.calls == []
assert coresys.dbus.udisks2.supported_filesystems == ["ext4"]
assert dbus == [] 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.""" """Test get block device by object path."""
udisks2 = UDisks2()
with pytest.raises(DBusNotConnectedError): with pytest.raises(DBusNotConnectedError):
coresys.dbus.udisks2.get_block_device( udisks2.get_block_device("/org/freedesktop/UDisks2/block_devices/sda1")
"/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" "/org/freedesktop/UDisks2/block_devices/sda1"
) )
assert block_device.id_label == "hassos-data" assert block_device.id_label == "hassos-data"
with pytest.raises(DBusObjectError): 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.""" """Test get drive by object path."""
udisks2 = UDisks2()
with pytest.raises(DBusNotConnectedError): with pytest.raises(DBusNotConnectedError):
coresys.dbus.udisks2.get_drive( udisks2.get_drive("/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291")
"/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( drive = udisks2.get_drive("/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291")
"/org/freedesktop/UDisks2/drives/BJTD4R_0x97cde291"
)
assert drive.id == "BJTD4R-0x97cde291" assert drive.id == "BJTD4R-0x97cde291"
with pytest.raises(DBusObjectError): 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.""" """Test resolve device."""
udisks2 = UDisks2()
with pytest.raises(DBusNotConnectedError): 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 udisks2.resolve_device(DeviceSpecification(path="/dev/sda1"))
devices = await coresys.dbus.udisks2.resolve_device(
DeviceSpecification(path="/dev/sda1")
)
assert len(devices) == 1 assert len(devices) == 1
assert devices[0].id_label == "hassos-data" assert devices[0].id_label == "hassos-data"
assert dbus == [ assert udisks2_manager_service.ResolveDevice.calls == [
"/org/freedesktop/UDisks2/Manager-org.freedesktop.UDisks2.Manager.ResolveDevice" (
{"path": Variant("s", "/dev/sda1")},
{"auth.no_user_interaction": Variant("b", True)},
)
] ]

86
tests/dbus/udisks2/test_partition_table.py
View File

@ -1,7 +1,6 @@
"""Test UDisks2 Partition Table.""" """Test UDisks2 Partition Table."""
import asyncio from dbus_fast import Variant
from dbus_fast.aio.message_bus import MessageBus from dbus_fast.aio.message_bus import MessageBus
import pytest import pytest
@ -9,10 +8,43 @@ from supervisor.dbus.udisks2.const import PartitionTableType
from supervisor.dbus.udisks2.data import CreatePartitionOptions from supervisor.dbus.udisks2.data import CreatePartitionOptions
from supervisor.dbus.udisks2.partition_table import UDisks2PartitionTable 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.""" """Test partition table info."""
sda = UDisks2PartitionTable("/org/freedesktop/UDisks2/block_devices/sda") sda = UDisks2PartitionTable("/org/freedesktop/UDisks2/block_devices/sda")
sdb = UDisks2PartitionTable( 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.type is None
assert sdb.partitions is None assert sdb.partitions is None
await sda.connect(dbus_bus) await sda.connect(dbus_session_bus)
await sdb.connect(dbus_bus) await sdb.connect(dbus_session_bus)
assert sda.type == PartitionTableType.GPT assert sda.type == PartitionTableType.GPT
assert sda.partitions == ["/org/freedesktop/UDisks2/block_devices/sda1"] assert sda.partitions == ["/org/freedesktop/UDisks2/block_devices/sda1"]
assert sdb.type == PartitionTableType.GPT assert sdb.type == PartitionTableType.GPT
assert sdb.partitions == ["/org/freedesktop/UDisks2/block_devices/sdb1"] assert sdb.partitions == ["/org/freedesktop/UDisks2/block_devices/sdb1"]
fire_property_change_signal( partition_table_sda_service.emit_properties_changed(
sda,
{ {
"Partitions": [ "Partitions": [
"/org/freedesktop/UDisks2/block_devices/sda1", "/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 == [ assert sda.partitions == [
"/org/freedesktop/UDisks2/block_devices/sda1", "/org/freedesktop/UDisks2/block_devices/sda1",
"/org/freedesktop/UDisks2/block_devices/sda2", "/org/freedesktop/UDisks2/block_devices/sda2",
] ]
with pytest.raises(AssertionError): partition_table_sda_service.emit_properties_changed({}, ["Partitions"])
fire_property_change_signal( await partition_table_sda_service.ping()
sdb, 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(
{ {
"MountPoints": [ "Partitions": [
"/org/freedesktop/UDisks2/block_devices/sdb", "/org/freedesktop/UDisks2/block_devices/sdb1",
"/org/freedesktop/UDisks2/block_devices/sdb", "/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.""" """Test create partition."""
sda = UDisks2PartitionTable("/org/freedesktop/UDisks2/block_devices/sda") sda = UDisks2PartitionTable("/org/freedesktop/UDisks2/block_devices/sda")
await sda.connect(dbus_bus) await sda.connect(dbus_session_bus)
assert ( assert (
await sda.create_partition( 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" == "/org/freedesktop/UDisks2/block_devices/sda2"
) )
assert dbus == [ assert partition_table_sda_service.CreatePartition.calls == [
"/org/freedesktop/UDisks2/block_devices/sda-org.freedesktop.UDisks2.PartitionTable.CreatePartition" (
0,
1000000,
"dos",
"hassos-data",
{
"partition-type": Variant("s", "primary"),
"auth.no_user_interaction": Variant("b", True),
},
)
] ]

549
tests/dbus_service_mocks/udisks2_block.py Normal file
View File

@ -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."""

324
tests/dbus_service_mocks/udisks2_drive.py Normal file
View File

@ -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."""

118
tests/dbus_service_mocks/udisks2_filesystem.py Normal file
View File

@ -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."""

48
tests/dbus_service_mocks/udisks2_loop.py Normal file
View File

@ -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."""

102
tests/dbus_service_mocks/udisks2_manager.py Normal file
View File

@ -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"]

184
tests/dbus_service_mocks/udisks2_partition.py Normal file
View File

@ -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."""

90
tests/dbus_service_mocks/udisks2_partition_table.py Normal file
View File

@ -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"