feat: TYPE_LABEL = "Storage" replaces storage_device = True flag

The TYPE_LABEL attribute on the base class hierarchy is now the
canonical way for drivers to declare their device category. The
storage_device sentinel flag is removed — classification is determined
by reading TYPE_LABEL via getattr() in the backend.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Christian Manivong
2026-06-24 11:38:36 +02:00
co-authored by Claude Sonnet 4.6
parent 07b68cd73f
commit 46c4c13f3d
7 changed files with 48 additions and 27 deletions
+3 -3
View File
@@ -14,9 +14,9 @@ Usage::
driver = OpenMediaVaultDriver("192.168.1.10", "admin", "secret") driver = OpenMediaVaultDriver("192.168.1.10", "admin", "secret")
driver.open() driver.open()
print(driver.get_facts()) facts = driver.get_facts()
print(driver.get_disks()) disks = driver.get_disks()
print(driver.get_shares()) shares = driver.get_shares()
driver.close() driver.close()
""" """
+27 -24
View File
@@ -16,9 +16,11 @@ All optional_args from LinuxDriver are accepted (pkg_manager, sudo_password, …
Tested against OpenMediaVault 6.x and 7.x (Debian Bullseye / Bookworm base). Tested against OpenMediaVault 6.x and 7.x (Debian Bullseye / Bookworm base).
""" """
from __future__ import annotations
import json as _json import json as _json
import re import re
from typing import Any, Dict, List from typing import Any
from napalm_device_types.models import ( from napalm_device_types.models import (
DiskPoolDict, DiskPoolDict,
@@ -45,12 +47,12 @@ class OpenMediaVaultDriver(LinuxDriver):
the full systemd service list). the full systemd service list).
""" """
TYPE_LABEL = "Storage"
driver_name = "openmediavault" driver_name = "openmediavault"
storage_device = True # signals netOrk to classify this driver as "storage"
# ── Facts override ───────────────────────────────────────────────────────── # ── Facts override ─────────────────────────────────────────────────────────
def get_facts(self) -> Dict[str, Any]: def get_facts(self) -> dict[str, Any]:
"""Return device facts, overriding OMV version from dpkg.""" """Return device facts, overriding OMV version from dpkg."""
facts = super().get_facts() facts = super().get_facts()
try: try:
@@ -65,7 +67,7 @@ class OpenMediaVaultDriver(LinuxDriver):
# ── Warnings ────────────────────────────────────────────────────────────── # ── Warnings ──────────────────────────────────────────────────────────────
def get_device_warnings(self) -> List[Dict[str, Any]]: def get_device_warnings(self) -> list[dict[str, Any]]:
"""Return warnings, ignoring packages held back by apt-get upgrade. """Return warnings, ignoring packages held back by apt-get upgrade.
On OMV, kernel and OMV-specific packages are routinely held back On OMV, kernel and OMV-specific packages are routinely held back
@@ -74,7 +76,7 @@ class OpenMediaVaultDriver(LinuxDriver):
We simulate the upgrade first and only warn if there are packages We simulate the upgrade first and only warn if there are packages
that a standard apt-get upgrade would actually install. that a standard apt-get upgrade would actually install.
""" """
warnings: List[Dict[str, Any]] = [] warnings: list[dict[str, Any]] = []
try: try:
sim = self._sudo( sim = self._sudo(
"DEBIAN_FRONTEND=noninteractive apt-get upgrade --simulate 2>&1", "DEBIAN_FRONTEND=noninteractive apt-get upgrade --simulate 2>&1",
@@ -105,7 +107,7 @@ class OpenMediaVaultDriver(LinuxDriver):
# ── Physical disks ───────────────────────────────────────────────────────── # ── Physical disks ─────────────────────────────────────────────────────────
def get_disks(self) -> List[PhysicalDiskDict]: def get_disks(self) -> list[PhysicalDiskDict]:
"""Return physical disk inventory. """Return physical disk inventory.
Uses ``lsblk -J`` for device enumeration and ``smartctl -iA`` Uses ``lsblk -J`` for device enumeration and ``smartctl -iA``
@@ -141,7 +143,7 @@ class OpenMediaVaultDriver(LinuxDriver):
}, },
] ]
""" """
result: List[PhysicalDiskDict] = [] result: list[PhysicalDiskDict] = []
try: try:
raw = self._send( raw = self._send(
@@ -226,7 +228,7 @@ class OpenMediaVaultDriver(LinuxDriver):
# ── md RAID pools ────────────────────────────────────────────────────────── # ── md RAID pools ──────────────────────────────────────────────────────────
def get_disk_pools(self) -> Dict[str, DiskPoolDict]: def get_disk_pools(self) -> dict[str, DiskPoolDict]:
"""Return md RAID array status from ``/proc/mdstat`` and ``mdadm --detail``. """Return md RAID array status from ``/proc/mdstat`` and ``mdadm --detail``.
Keys are array names (e.g. ``"md1"``). Each value contains: Keys are array names (e.g. ``"md1"``). Each value contains:
@@ -261,7 +263,7 @@ class OpenMediaVaultDriver(LinuxDriver):
}, },
} }
""" """
result: Dict[str, DiskPoolDict] = {} result: dict[str, DiskPoolDict] = {}
try: try:
mdstat = self._send("cat /proc/mdstat 2>/dev/null") mdstat = self._send("cat /proc/mdstat 2>/dev/null")
@@ -310,7 +312,7 @@ class OpenMediaVaultDriver(LinuxDriver):
# ── Logical volumes / filesystems ────────────────────────────────────────── # ── Logical volumes / filesystems ──────────────────────────────────────────
def get_volumes(self) -> Dict[str, LogicalVolumeDict]: def get_volumes(self) -> dict[str, LogicalVolumeDict]:
"""Return mounted filesystem info from ``df``. """Return mounted filesystem info from ``df``.
Only real block devices (``/dev/…``) are included; tmpfs, devtmpfs, Only real block devices (``/dev/…``) are included; tmpfs, devtmpfs,
@@ -351,7 +353,7 @@ class OpenMediaVaultDriver(LinuxDriver):
}, },
} }
""" """
result: Dict[str, LogicalVolumeDict] = {} result: dict[str, LogicalVolumeDict] = {}
try: try:
df_out = self._send("df -Pk 2>/dev/null") df_out = self._send("df -Pk 2>/dev/null")
@@ -400,7 +402,7 @@ class OpenMediaVaultDriver(LinuxDriver):
# ── NAS shares ──────────────────────────────────────────────────────────── # ── NAS shares ────────────────────────────────────────────────────────────
def get_shares(self) -> Dict[str, NASShareDict]: def get_shares(self) -> dict[str, NASShareDict]:
"""Return all configured NAS shares from the OMV config database. """Return all configured NAS shares from the OMV config database.
Covers SMB/CIFS, NFS, and FTP shares. Keys are prefixed with the Covers SMB/CIFS, NFS, and FTP shares. Keys are prefixed with the
@@ -442,7 +444,7 @@ class OpenMediaVaultDriver(LinuxDriver):
}, },
} }
""" """
shares: Dict[str, NASShareDict] = {} shares: dict[str, NASShareDict] = {}
shares.update(self._smb_shares()) shares.update(self._smb_shares())
shares.update(self._nfs_shares()) shares.update(self._nfs_shares())
shares.update(self._ftp_shares()) shares.update(self._ftp_shares())
@@ -450,7 +452,7 @@ class OpenMediaVaultDriver(LinuxDriver):
# ── Storage services ────────────────────────────────────────────────────── # ── Storage services ──────────────────────────────────────────────────────
def get_storage_services(self) -> Dict[str, StorageServiceDict]: def get_storage_services(self) -> dict[str, StorageServiceDict]:
"""Return status of key NAS services (SMB, NFS, FTP, SSH, rsync). """Return status of key NAS services (SMB, NFS, FTP, SSH, rsync).
This method is separate from ``get_services()`` (inherited from This method is separate from ``get_services()`` (inherited from
@@ -482,7 +484,7 @@ class OpenMediaVaultDriver(LinuxDriver):
("ssh", "ssh", 22), ("ssh", "ssh", 22),
("rsync", "rsync", 873), ("rsync", "rsync", 873),
] ]
result: Dict[str, StorageServiceDict] = {} result: dict[str, StorageServiceDict] = {}
for key, unit, port in _SVC_MAP: for key, unit, port in _SVC_MAP:
try: try:
running = ( running = (
@@ -502,16 +504,17 @@ class OpenMediaVaultDriver(LinuxDriver):
# ── Not implemented ──────────────────────────────────────────────────────── # ── Not implemented ────────────────────────────────────────────────────────
def get_volume_snapshots(self, volume: str = "") -> List[VolumeSnapshotDict]: def get_volume_snapshots(self, volume: str = "") -> list[VolumeSnapshotDict]:
"""Return volume snapshots (not supported on OpenMediaVault)."""
raise NotImplementedError( raise NotImplementedError(
"OpenMediaVault does not expose volume snapshots via this driver" "OpenMediaVault does not expose volume snapshots via this driver"
) )
def get_replication_jobs(self) -> List[ReplicationJobDict]: def get_replication_jobs(self) -> list[ReplicationJobDict]:
"""Return replication jobs (not supported on OpenMediaVault).""" """Return replication jobs (not supported on OpenMediaVault)."""
raise NotImplementedError raise NotImplementedError
def get_quotas(self) -> List[StorageQuotaDict]: def get_quotas(self) -> list[StorageQuotaDict]:
"""Return storage quotas (not supported on OpenMediaVault).""" """Return storage quotas (not supported on OpenMediaVault)."""
raise NotImplementedError raise NotImplementedError
@@ -558,8 +561,8 @@ class OpenMediaVaultDriver(LinuxDriver):
pass pass
return "" return ""
def _smb_shares(self) -> Dict[str, NASShareDict]: def _smb_shares(self) -> dict[str, NASShareDict]:
out: Dict[str, NASShareDict] = {} out: dict[str, NASShareDict] = {}
for s in self._omv_read("conf.service.smb.shares"): for s in self._omv_read("conf.service.smb.shares"):
name = s.get("name", "") name = s.get("name", "")
if not name: if not name:
@@ -577,8 +580,8 @@ class OpenMediaVaultDriver(LinuxDriver):
) )
return out return out
def _nfs_shares(self) -> Dict[str, NASShareDict]: def _nfs_shares(self) -> dict[str, NASShareDict]:
out: Dict[str, NASShareDict] = {} out: dict[str, NASShareDict] = {}
for s in self._omv_read("conf.service.nfs.shares"): for s in self._omv_read("conf.service.nfs.shares"):
path = self._resolve_shared_folder(s.get("sharedfolderref", "")) path = self._resolve_shared_folder(s.get("sharedfolderref", ""))
name = path.split("/")[-1] or s.get("sharedfolderref", "") name = path.split("/")[-1] or s.get("sharedfolderref", "")
@@ -597,8 +600,8 @@ class OpenMediaVaultDriver(LinuxDriver):
) )
return out return out
def _ftp_shares(self) -> Dict[str, NASShareDict]: def _ftp_shares(self) -> dict[str, NASShareDict]:
out: Dict[str, NASShareDict] = {} out: dict[str, NASShareDict] = {}
for s in self._omv_read("conf.service.ftp.shares"): for s in self._omv_read("conf.service.ftp.shares"):
path = self._resolve_shared_folder(s.get("sharedfolderref", "")) path = self._resolve_shared_folder(s.get("sharedfolderref", ""))
name = path.split("/")[-1] or s.get("sharedfolderref", "") name = path.split("/")[-1] or s.get("sharedfolderref", "")
Binary file not shown.
+18
View File
@@ -0,0 +1,18 @@
"""Tests for TYPE_LABEL on OpenMediaVaultDriver."""
from napalm_openmediavault.openmediavault import OpenMediaVaultDriver
def test_type_label_is_storage():
assert OpenMediaVaultDriver.TYPE_LABEL == "Storage"
def test_type_label_overrides_linux():
from napalm_linux.linux import LinuxDriver
assert LinuxDriver.TYPE_LABEL == "Linux"
assert OpenMediaVaultDriver.TYPE_LABEL != LinuxDriver.TYPE_LABEL
def test_storage_device_flag_removed():
assert not hasattr(OpenMediaVaultDriver, "storage_device"), \
"storage_device is superseded by TYPE_LABEL = 'Storage'"