diff --git a/supervisor/api/__init__.py b/supervisor/api/__init__.py index fe4ab15a2..1b195209e 100644 --- a/supervisor/api/__init__.py +++ b/supervisor/api/__init__.py @@ -146,6 +146,7 @@ class RestAPI(CoreSysAttributes): web.post("/os/update", api_os.update), web.post("/os/config/sync", api_os.config_sync), web.post("/os/datadisk/move", api_os.migrate_data), + web.get("/os/datadisk/list", api_os.list_data), ] ) diff --git a/supervisor/api/os.py b/supervisor/api/os.py index 8965ea005..ae1b4939d 100644 --- a/supervisor/api/os.py +++ b/supervisor/api/os.py @@ -10,6 +10,7 @@ import voluptuous as vol from ..const import ( ATTR_BOARD, ATTR_BOOT, + ATTR_DEVICES, ATTR_UPDATE_AVAILABLE, ATTR_VERSION, ATTR_VERSION_LATEST, @@ -59,3 +60,10 @@ class APIOS(CoreSysAttributes): body = await api_validate(SCHEMA_DISK, request) await asyncio.shield(self.sys_os.datadisk.migrate_disk(body[ATTR_DEVICE])) + + @api_process + async def list_data(self, request: web.Request) -> Dict[str, Any]: + """Return possible data targets.""" + return { + ATTR_DEVICES: self.sys_os.datadisk.available_disks, + } diff --git a/supervisor/bootstrap.py b/supervisor/bootstrap.py index be75daa7b..d55b78df3 100644 --- a/supervisor/bootstrap.py +++ b/supervisor/bootstrap.py @@ -37,7 +37,7 @@ from .core import Core from .coresys import CoreSys from .dbus.manager import DBusManager from .discovery import Discovery -from .hardware.module import HardwareManager +from .hardware.manager import HardwareManager from .homeassistant.module import HomeAssistant from .host.manager import HostManager from .ingress import Ingress diff --git a/supervisor/coresys.py b/supervisor/coresys.py index 9d7a8696f..7e269fa27 100644 --- a/supervisor/coresys.py +++ b/supervisor/coresys.py @@ -25,7 +25,7 @@ if TYPE_CHECKING: from .core import Core from .dbus.manager import DBusManager from .discovery import Discovery - from .hardware.module import HardwareManager + from .hardware.manager import HardwareManager from .os.manager import OSManager from .homeassistant.module import HomeAssistant from .host.manager import HostManager diff --git a/supervisor/hardware/data.py b/supervisor/hardware/data.py index 957b5b2f1..d7867a705 100644 --- a/supervisor/hardware/data.py +++ b/supervisor/hardware/data.py @@ -1,8 +1,11 @@ """Data representation of Hardware.""" +from __future__ import annotations + from pathlib import Path from typing import Dict, List, Optional import attr +import pyudev @attr.s(slots=True, frozen=True) @@ -13,16 +16,18 @@ class Device: path: Path = attr.ib(eq=False) sysfs: Path = attr.ib(eq=True) subsystem: str = attr.ib(eq=False) + parent: Optional[Path] = attr.ib(eq=False) links: List[Path] = attr.ib(eq=False) attributes: Dict[str, str] = attr.ib(eq=False) + children: List[Path] = attr.ib(eq=False) @property - def cgroups_major(self) -> int: + def major(self) -> int: """Return Major cgroups.""" return int(self.attributes.get("MAJOR", 0)) @property - def cgroups_minor(self) -> int: + def minor(self) -> int: """Return Major cgroups.""" return int(self.attributes.get("MINOR", 0)) @@ -34,3 +39,17 @@ class Device: continue return link return None + + @staticmethod + def import_udev(udevice: pyudev.Device) -> Device: + """Remap a pyudev object into a Device.""" + return Device( + udevice.sys_name, + Path(udevice.device_node), + Path(udevice.sys_path), + udevice.subsystem, + None if not udevice.parent else Path(udevice.parent.sys_path), + [Path(node) for node in udevice.device_links], + {attr: udevice.properties[attr] for attr in udevice.properties}, + [Path(node.sys_path) for node in udevice.children], + ) diff --git a/supervisor/hardware/disk.py b/supervisor/hardware/disk.py index 71fec8e01..5c213b509 100644 --- a/supervisor/hardware/disk.py +++ b/supervisor/hardware/disk.py @@ -4,6 +4,8 @@ from pathlib import Path import shutil from typing import Union +from supervisor.exceptions import HardwareNotFound + from ..coresys import CoreSys, CoreSysAttributes from .const import UdevSubsystem from .data import Device @@ -22,13 +24,29 @@ class HwDisk(CoreSysAttributes): """Init hardware object.""" self.coresys = coresys - def is_system_partition(self, device: Device) -> bool: - """Return true if this is a system disk/partition.""" + def is_used_by_system(self, device: Device) -> bool: + """Return true if this is a system partition.""" if device.subsystem != UdevSubsystem.DISK: return False - if device.attributes.get("ID_FS_LABEL", "").startswith("hassos"): + # Root + if device.minor == 0: + for child in device.children: + try: + device = self.sys_hardware.get_by_path(child) + except HardwareNotFound: + continue + if device.subsystem == UdevSubsystem.DISK: + if device.attributes.get("ID_FS_LABEL", "").startswith("hassos"): + return True + return False + + # Partition + if device.minor > 0 and device.attributes.get("ID_FS_LABEL", "").startswith( + "hassos" + ): return True + return False def get_disk_total_space(self, path: Union[str, Path]) -> float: diff --git a/supervisor/hardware/module.py b/supervisor/hardware/manager.py similarity index 91% rename from supervisor/hardware/module.py rename to supervisor/hardware/manager.py index a8818bc68..079ca01bc 100644 --- a/supervisor/hardware/module.py +++ b/supervisor/hardware/manager.py @@ -107,15 +107,7 @@ class HardwareManager(CoreSysAttributes): # Skip devices without mapping if not device.device_node or self.helper.hide_virtual_device(device): continue - - self._devices[device.sys_name] = Device( - device.sys_name, - Path(device.device_node), - Path(device.sys_path), - device.subsystem, - [Path(node) for node in device.device_links], - {attr: device.properties[attr] for attr in device.properties}, - ) + self._devices[device.sys_name] = Device.import_udev(device) async def load(self) -> None: """Load hardware backend.""" diff --git a/supervisor/hardware/monitor.py b/supervisor/hardware/monitor.py index 1705570e3..39077f9a0 100644 --- a/supervisor/hardware/monitor.py +++ b/supervisor/hardware/monitor.py @@ -107,14 +107,7 @@ class HwMonitor(CoreSysAttributes): ) return - device = Device( - udev.sys_name, - Path(udev.device_node), - Path(udev.sys_path), - udev.subsystem, - [Path(node) for node in udev.device_links], - {attr: udev.properties[attr] for attr in udev.properties}, - ) + device = Device.import_udev(udev) self.sys_hardware.update_device(device) # If it's a new device - process actions diff --git a/supervisor/hardware/policy.py b/supervisor/hardware/policy.py index 2eaf98517..dc7da28c8 100644 --- a/supervisor/hardware/policy.py +++ b/supervisor/hardware/policy.py @@ -72,7 +72,7 @@ class HwPolicy(CoreSysAttributes): def is_match_cgroup(self, group: PolicyGroup, device: Device) -> bool: """Return true if device is in cgroup Policy.""" - return device.cgroups_major in _CGROUPS.get(group, []) + return device.major in _CGROUPS.get(group, []) def get_cgroups_rules(self, group: PolicyGroup) -> List[str]: """Generate cgroups rules for a policy group.""" @@ -81,10 +81,10 @@ class HwPolicy(CoreSysAttributes): # Lookup dynamic device groups from host if group in _CGROUPS_DYNAMIC_MAJOR: majors = { - device.cgroups_major + device.major for device in self.sys_hardware.devices if device.subsystem in _CGROUPS_DYNAMIC_MAJOR[group] - and device.cgroups_major not in _CGROUPS[group] + and device.major not in _CGROUPS[group] } cgroups.extend([f"c {dev}:* rwm" for dev in majors]) @@ -93,7 +93,7 @@ class HwPolicy(CoreSysAttributes): for device in self.sys_hardware.devices: if ( device.subsystem not in _CGROUPS_DYNAMIC_MINOR[group] - or device.cgroups_major in _CGROUPS[group] + or device.major in _CGROUPS[group] ): continue cgroups.append(self.get_cgroups_rule(device)) @@ -103,7 +103,7 @@ class HwPolicy(CoreSysAttributes): def get_cgroups_rule(self, device: Device) -> str: """Generate a cgroups rule for given device.""" cgroup_type = "c" if device.subsystem != UdevSubsystem.DISK else "b" - return f"{cgroup_type} {device.cgroups_major}:{device.cgroups_minor} rwm" + return f"{cgroup_type} {device.major}:{device.minor} rwm" def get_full_access(self) -> str: """Get full access to all devices.""" @@ -111,7 +111,7 @@ class HwPolicy(CoreSysAttributes): def allowed_for_access(self, device: Device) -> bool: """Return True if allow to access to this device.""" - if self.sys_hardware.disk.is_system_partition(device): + if self.sys_hardware.disk.is_used_by_system(device): return False return True diff --git a/supervisor/os/data_disk.py b/supervisor/os/data_disk.py index 9ac2eb436..6299babe1 100644 --- a/supervisor/os/data_disk.py +++ b/supervisor/os/data_disk.py @@ -1,7 +1,7 @@ """Home Assistant Operating-System DataDisk.""" import logging from pathlib import Path -from typing import Optional +from typing import List, Optional from awesomeversion import AwesomeVersion @@ -33,6 +33,23 @@ class DataDisk(CoreSysAttributes): """Return Path to used Disk for data.""" return self.sys_dbus.agent.datadisk.current_device + @property + def available_disks(self) -> List[Path]: + """Return a list of possible new disk locations.""" + device_paths: List[Path] = [] + for device in self.sys_hardware.devices: + # Filter devices out which can't be a target + if ( + device.subsystem != UdevSubsystem.DISK + or device.attributes.get("DEVTYPE") != "disk" + or device.minor != 0 + or self.sys_hardware.disk.is_used_by_system(device) + ): + continue + device_paths.append(device.path) + + return device_paths + @Job(conditions=[JobCondition.OS_AGENT]) async def load(self) -> None: """Load DataDisk feature.""" @@ -55,11 +72,11 @@ class DataDisk(CoreSysAttributes): f"'{new_disk!s}' don't exists on the host!", _LOGGER.error ) from None - if device.subsystem != UdevSubsystem.DISK: + if device.subsystem != UdevSubsystem.DISK or device.minor != 0: raise HassOSDataDiskError( f"'{new_disk!s}' is not a harddisk!", _LOGGER.error ) - if self.sys_hardware.disk.is_system_partition(device): + if self.sys_hardware.disk.is_used_by_system(device): raise HassOSDataDiskError( f"'{new_disk}' is a system disk and can't be used!", _LOGGER.error ) diff --git a/tests/addons/test_options.py b/tests/addons/test_options.py index 8d9a00bd7..46a6fc706 100644 --- a/tests/addons/test_options.py +++ b/tests/addons/test_options.py @@ -136,25 +136,40 @@ def test_simple_device_schema(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/002"), "tty", + None, [], {"ID_VENDOR": "xy"}, + [], ), Device( "ttyUSB0", Path("/dev/ttyUSB0"), Path("/sys/bus/usb/001"), "tty", + None, [Path("/dev/ttyS1"), Path("/dev/serial/by-id/xyx")], {"ID_VENDOR": "xy"}, + [], + ), + Device( + "ttyS0", + Path("/dev/ttyS0"), + Path("/sys/bus/usb/003"), + "tty", + None, + [], + {}, + [], ), - Device("ttyS0", Path("/dev/ttyS0"), Path("/sys/bus/usb/003"), "tty", [], {}), Device( "video1", Path("/dev/video1"), Path("/sys/bus/usb/004"), "misc", + None, [], {"ID_VENDOR": "xy"}, + [], ), ): coresys.hardware.update_device(device) @@ -297,25 +312,40 @@ def test_ui_simple_device_schema(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/002"), "tty", + None, [], {"ID_VENDOR": "xy"}, + [], ), Device( "ttyUSB0", Path("/dev/ttyUSB0"), Path("/sys/bus/usb/001"), "tty", + None, [Path("/dev/ttyS1"), Path("/dev/serial/by-id/xyx")], {"ID_VENDOR": "xy"}, + [], + ), + Device( + "ttyS0", + Path("/dev/ttyS0"), + Path("/sys/bus/usb/003"), + "tty", + None, + [], + {}, + [], ), - Device("ttyS0", Path("/dev/ttyS0"), Path("/sys/bus/usb/003"), "tty", [], {}), Device( "video1", Path("/dev/video1"), Path("/sys/bus/usb/004"), "misc", + None, [], {"ID_VENDOR": "xy"}, + [], ), ): coresys.hardware.update_device(device) @@ -348,25 +378,40 @@ def test_ui_simple_device_schema_no_filter(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/002"), "tty", + None, [], {"ID_VENDOR": "xy"}, + [], ), Device( "ttyUSB0", Path("/dev/ttyUSB0"), Path("/sys/bus/usb/001"), "tty", + None, [Path("/dev/ttyS1"), Path("/dev/serial/by-id/xyx")], {"ID_VENDOR": "xy"}, + [], + ), + Device( + "ttyS0", + Path("/dev/ttyS0"), + Path("/sys/bus/usb/003"), + "tty", + None, + [], + {}, + [], ), - Device("ttyS0", Path("/dev/ttyS0"), Path("/sys/bus/usb/003"), "tty", [], {}), Device( "video1", Path("/dev/video1"), Path("/sys/bus/usb/004"), "misc", + None, [], {"ID_VENDOR": "xy"}, + [], ), ): coresys.hardware.update_device(device) diff --git a/tests/api/test_hardware.py b/tests/api/test_hardware.py index f2deb4978..116ca376f 100644 --- a/tests/api/test_hardware.py +++ b/tests/api/test_hardware.py @@ -24,8 +24,10 @@ async def test_api_hardware_info_device(api_client, coresys): Path("/dev/sda"), Path("/sys/bus/usb/000"), "sound", + None, [Path("/dev/serial/by-id/test")], {"ID_NAME": "xy"}, + [], ) ) diff --git a/tests/api/test_os.py b/tests/api/test_os.py index 036ec4557..cacc694ac 100644 --- a/tests/api/test_os.py +++ b/tests/api/test_os.py @@ -1,7 +1,10 @@ """Test OS API.""" +from pathlib import Path + import pytest from supervisor.coresys import CoreSys +from supervisor.hardware.data import Device # pylint: disable=protected-access @@ -36,8 +39,8 @@ async def test_api_os_info_with_agent(api_client, coresys: CoreSys): @pytest.mark.asyncio -async def test_api_os_move_data(api_client, coresys: CoreSys): - """Test docker info api.""" +async def test_api_os_datadisk_move(api_client, coresys: CoreSys): + """Test datadisk move without exists disk.""" await coresys.dbus.agent.connect() await coresys.dbus.agent.update() coresys.os._available = True @@ -46,3 +49,40 @@ async def test_api_os_move_data(api_client, coresys: CoreSys): result = await resp.json() assert result["message"] == "'/dev/sdaaaa' don't exists on the host!" + + +@pytest.mark.asyncio +async def test_api_os_datadisk_list(api_client, coresys: CoreSys): + """Test datadisk list function.""" + await coresys.dbus.agent.connect() + await coresys.dbus.agent.update() + + coresys.hardware.update_device( + Device( + "sda", + Path("/dev/sda"), + Path("/sys/bus/usb/000"), + "block", + None, + [Path("/dev/serial/by-id/test")], + {"ID_NAME": "xy", "MINOR": "0", "DEVTYPE": "disk"}, + [], + ) + ) + coresys.hardware.update_device( + Device( + "sda1", + Path("/dev/sda1"), + Path("/sys/bus/usb/000/1"), + "block", + None, + [Path("/dev/serial/by-id/test1")], + {"ID_NAME": "xy", "MINOR": "1", "DEVTYPE": "partition"}, + [], + ) + ) + + resp = await api_client.get("/os/datadisk/list") + result = await resp.json() + + assert result["data"]["devices"] == ["/dev/sda"] diff --git a/tests/hardware/test_data.py b/tests/hardware/test_data.py index 15b104a8c..daac0adbf 100644 --- a/tests/hardware/test_data.py +++ b/tests/hardware/test_data.py @@ -13,10 +13,12 @@ def test_device_property(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/001"), "tty", + None, [Path("/dev/serial/by-id/fixed-device")], {"MAJOR": "5", "MINOR": "10"}, + [], ) assert device.by_id == device.links[0] - assert device.cgroups_major == 5 - assert device.cgroups_minor == 10 + assert device.major == 5 + assert device.minor == 10 diff --git a/tests/hardware/test_disk.py b/tests/hardware/test_disk.py index 0de8d204d..84e8e0ef6 100644 --- a/tests/hardware/test_disk.py +++ b/tests/hardware/test_disk.py @@ -3,32 +3,51 @@ from pathlib import Path from unittest.mock import patch +from supervisor.coresys import CoreSys from supervisor.hardware.data import Device -def test_system_partition(coresys): - """Test if it is a system partition.""" +def test_system_partition_disk(coresys: CoreSys): + """Test if it is a system disk/partition.""" disk = Device( - "sda0", - Path("/dev/sda0"), + "sda1", + Path("/dev/sda1"), Path("/sys/bus/usb/001"), "block", + None, [], {"MAJOR": "5", "MINOR": "10"}, + [], ) - assert not coresys.hardware.disk.is_system_partition(disk) + assert not coresys.hardware.disk.is_used_by_system(disk) disk = Device( - "sda0", - Path("/dev/sda0"), + "sda1", + Path("/dev/sda1"), Path("/sys/bus/usb/001"), "block", + None, [], {"MAJOR": "5", "MINOR": "10", "ID_FS_LABEL": "hassos-overlay"}, + [], ) - assert coresys.hardware.disk.is_system_partition(disk) + assert coresys.hardware.disk.is_used_by_system(disk) + + coresys.hardware.update_device(disk) + disk_root = Device( + "sda", + Path("/dev/sda"), + Path("/sys/bus/usb/001"), + "block", + None, + [], + {"MAJOR": "5", "MINOR": "0"}, + [Path("/dev/sda1")], + ) + + assert coresys.hardware.disk.is_used_by_system(disk_root) def test_free_space(coresys): diff --git a/tests/hardware/test_helper.py b/tests/hardware/test_helper.py index ed82e42e4..c5c2359de 100644 --- a/tests/hardware/test_helper.py +++ b/tests/hardware/test_helper.py @@ -16,8 +16,10 @@ def test_have_audio(coresys): Path("/dev/sda"), Path("/sys/bus/usb/000"), "sound", + None, [], {"ID_NAME": "xy"}, + [], ) ) @@ -34,8 +36,10 @@ def test_have_usb(coresys): Path("/dev/sda"), Path("/sys/bus/usb/000"), "usb", + None, [], {"ID_NAME": "xy"}, + [], ) ) @@ -52,8 +56,10 @@ def test_have_gpio(coresys): Path("/dev/sda"), Path("/sys/bus/usb/000"), "gpio", + None, [], {"ID_NAME": "xy"}, + [], ) ) diff --git a/tests/hardware/test_module.py b/tests/hardware/test_module.py index b93ee5906..92f7ad0e7 100644 --- a/tests/hardware/test_module.py +++ b/tests/hardware/test_module.py @@ -25,25 +25,40 @@ def test_device_path_lookup(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/001"), "tty", + None, [], {"ID_VENDOR": "xy"}, + [], ), Device( "ttyUSB0", Path("/dev/ttyUSB0"), Path("/sys/bus/usb/000"), "tty", + None, [Path("/dev/ttyS1"), Path("/dev/serial/by-id/xyx")], {"ID_VENDOR": "xy"}, + [], + ), + Device( + "ttyS0", + Path("/dev/ttyS0"), + Path("/sys/bus/usb/002"), + "tty", + None, + [], + {}, + [], ), - Device("ttyS0", Path("/dev/ttyS0"), Path("/sys/bus/usb/002"), "tty", [], {}), Device( "video1", Path("/dev/video1"), Path("/sys/bus/usb/003"), "misc", + None, [], {"ID_VENDOR": "xy"}, + [], ), ): coresys.hardware.update_device(device) @@ -66,25 +81,40 @@ def test_device_filter(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/000"), "tty", + None, [], {"ID_VENDOR": "xy"}, + [], ), Device( "ttyUSB0", Path("/dev/ttyUSB0"), Path("/sys/bus/usb/001"), "tty", + None, [Path("/dev/ttyS1"), Path("/dev/serial/by-id/xyx")], {"ID_VENDOR": "xy"}, + [], + ), + Device( + "ttyS0", + Path("/dev/ttyS0"), + Path("/sys/bus/usb/002"), + "tty", + None, + [], + {}, + [], ), - Device("ttyS0", Path("/dev/ttyS0"), Path("/sys/bus/usb/002"), "tty", [], {}), Device( "video1", Path("/dev/video1"), Path("/sys/bus/usb/003"), "misc", + None, [], {"ID_VENDOR": "xy"}, + [], ), ): coresys.hardware.update_device(device) diff --git a/tests/hardware/test_policy.py b/tests/hardware/test_policy.py index f964cb894..13d5d0154 100644 --- a/tests/hardware/test_policy.py +++ b/tests/hardware/test_policy.py @@ -14,8 +14,10 @@ def test_device_policy(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/001"), "tty", + None, [], {"MAJOR": "5", "MINOR": "10"}, + [], ) assert coresys.hardware.policy.get_cgroups_rule(device) == "c 5:10 rwm" @@ -25,8 +27,10 @@ def test_device_policy(coresys): Path("/dev/sda0"), Path("/sys/bus/usb/001"), "block", + None, [], {"MAJOR": "5", "MINOR": "10"}, + [], ) assert coresys.hardware.policy.get_cgroups_rule(disk) == "b 5:10 rwm" @@ -48,8 +52,10 @@ def test_device_in_policy(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/001"), "tty", + None, [], {"MAJOR": "204", "MINOR": "10"}, + [], ) assert coresys.hardware.policy.is_match_cgroup(PolicyGroup.UART, device) @@ -64,8 +70,10 @@ def test_allowed_access(coresys): Path("/dev/sda0"), Path("/sys/bus/usb/001"), "block", + None, [], {"MAJOR": "5", "MINOR": "10", "ID_FS_LABEL": "hassos-overlay"}, + [], ) assert not coresys.hardware.policy.allowed_for_access(disk) @@ -75,8 +83,10 @@ def test_allowed_access(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/001"), "tty", + None, [], {"MAJOR": "204", "MINOR": "10"}, + [], ) assert coresys.hardware.policy.allowed_for_access(device) @@ -90,32 +100,40 @@ def test_dynamic_group_alloc_minor(coresys): Path("/dev/ttyACM0"), Path("/sys/bus/usb/001"), "tty", + None, [], {"MAJOR": "204", "MINOR": "10"}, + [], ), Device( "ttyUSB0", Path("/dev/ttyUSB0"), Path("/sys/bus/usb/000"), "tty", + None, [Path("/dev/ttyS1"), Path("/dev/serial/by-id/xyx")], {"MAJOR": "188", "MINOR": "10"}, + [], ), Device( "ttyS0", Path("/dev/ttyS0"), Path("/sys/bus/usb/002"), "tty", + None, [], {"MAJOR": "4", "MINOR": "65"}, + [], ), Device( "video1", Path("/dev/video1"), Path("/sys/bus/usb/003"), "misc", + None, [], {"MAJOR": "38", "MINOR": "10"}, + [], ), ): coresys.hardware.update_device(device) @@ -136,32 +154,40 @@ def test_dynamic_group_alloc_major(coresys): Path("/dev/gpio16"), Path("/sys/bus/usb/001"), "gpio", + None, [], {"MAJOR": "254", "MINOR": "10"}, + [], ), Device( "gpiomem", Path("/dev/gpiomem"), Path("/sys/bus/usb/000"), "gpiomem", + None, [Path("/dev/ttyS1"), Path("/dev/serial/by-id/xyx")], {"MAJOR": "239", "MINOR": "10"}, + [], ), Device( "ttyS0", Path("/dev/ttyS0"), Path("/sys/bus/usb/002"), "tty", + None, [], {"MAJOR": "4", "MINOR": "65"}, + [], ), Device( "video1", Path("/dev/video1"), Path("/sys/bus/usb/003"), "misc", + None, [], {"MAJOR": "38", "MINOR": "10"}, + [], ), ): coresys.hardware.update_device(device) diff --git a/tests/os/test_data_disk.py b/tests/os/test_data_disk.py new file mode 100644 index 000000000..04658637f --- /dev/null +++ b/tests/os/test_data_disk.py @@ -0,0 +1,64 @@ +"""Test OS API.""" +from pathlib import Path, PosixPath + +import pytest + +from supervisor.coresys import CoreSys +from supervisor.exceptions import HassOSDataDiskError +from supervisor.hardware.data import Device + +# pylint: disable=protected-access + + +@pytest.mark.asyncio +async def tests_datadisk_current(coresys: CoreSys): + """Test current datadisk.""" + await coresys.dbus.agent.connect() + await coresys.dbus.agent.update() + + assert coresys.os.datadisk.disk_used == PosixPath("/dev/sda") + + +@pytest.mark.asyncio +async def test_datadisk_move(coresys: CoreSys): + """Test datadisk moved without exists device.""" + await coresys.dbus.agent.connect() + await coresys.dbus.agent.update() + coresys.os._available = True + + with pytest.raises(HassOSDataDiskError): + await coresys.os.datadisk.migrate_disk(Path("/dev/sdaaaa")) + + +@pytest.mark.asyncio +async def test_datadisk_list(coresys: CoreSys): + """Test docker info api.""" + await coresys.dbus.agent.connect() + await coresys.dbus.agent.update() + + coresys.hardware.update_device( + Device( + "sda", + Path("/dev/sda"), + Path("/sys/bus/usb/000"), + "block", + None, + [Path("/dev/serial/by-id/test")], + {"ID_NAME": "xy", "MINOR": "0", "DEVTYPE": "disk"}, + [], + ) + ) + coresys.hardware.update_device( + Device( + "sda1", + Path("/dev/sda1"), + Path("/sys/bus/usb/000/1"), + "block", + None, + [Path("/dev/serial/by-id/test1")], + {"ID_NAME": "xy", "MINOR": "1", "DEVTYPE": "partition"}, + [], + ) + ) + + assert coresys.os.datadisk.available_disks == [PosixPath("/dev/sda")]