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

Return list of possible data disk targets (#3133)

Browse Source 
* Return list of possible data disk targets

* fix path

* fix tests

* Add test

* Fix tests

* Add tests

* Add more tests

* Remove debug

* Address comments

* more clear
This commit is contained in:
Pascal Vizeli 2021-09-21 14:51:58 +02:00 committed by GitHub
parent 4f97013df4
commit 04f36e92e1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 332 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
supervisor/api/__init__.py
View File

@ -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),
]
)

8
supervisor/api/os.py
View File

@ -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,
}

2
supervisor/bootstrap.py
View File

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

2
supervisor/coresys.py
View File

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

23
supervisor/hardware/data.py
View File

@ -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],
)

24
supervisor/hardware/disk.py
View File

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

10
supervisor/hardware/module.py → supervisor/hardware/manager.py
View File

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

9
supervisor/hardware/monitor.py
View File

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

12
supervisor/hardware/policy.py
View File

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

23
supervisor/os/data_disk.py
View File

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

51
tests/addons/test_options.py
View File

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

2
tests/api/test_hardware.py
View File

@ -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"},
[],
)
)

44
tests/api/test_os.py
View File

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

6
tests/hardware/test_data.py
View File

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

35
tests/hardware/test_disk.py
View File

@ -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):

6
tests/hardware/test_helper.py
View File

@ -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"},
[],
)
)

34
tests/hardware/test_module.py
View File

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

26
tests/hardware/test_policy.py
View File

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

64
tests/os/test_data_disk.py Normal file
View File

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