napalm_openmediavault/openmediavault.py

@@ -1,4 +1,3 @@
-# -*- coding: utf-8 -*-
 # Licensed under the Apache License, Version 2.0
 
 """NAPALM driver for OpenMediaVault NAS systems.
@@ -22,7 +21,7 @@ import json as _json
 import re
 from typing import Any
 
-from napalm_device_types import FingerprintRule
+from napalm_device_types import FingerprintRule, StorageDriver
 from napalm_device_types.models import (
     DiskPoolDict,
     LogicalVolumeDict,
@@ -33,7 +32,6 @@ from napalm_device_types.models import (
     StorageServiceDict,
     VolumeSnapshotDict,
 )
-from napalm_device_types import StorageDriver
 from napalm_linux.linux import LinuxDriver
 
 
@@ -76,9 +74,7 @@ class OpenMediaVaultDriver(StorageDriver, LinuxDriver):
         """Return device facts, overriding OMV version from dpkg."""
         facts = super().get_facts()
         try:
-            ver = self._send(
-                "dpkg-query -W -f='${Version}' openmediavault 2>/dev/null"
-            ).strip()
+            ver = self._send("dpkg-query -W -f='${Version}' openmediavault 2>/dev/null").strip()
             if ver:
                 facts["os_version"] = f"OMV {ver}"
         except Exception:
@@ -159,9 +155,7 @@ class OpenMediaVaultDriver(StorageDriver, LinuxDriver):
             rpm = 0
 
             try:
-                smart = self._sudo(
-                    f"smartctl -iA /dev/{name} 2>/dev/null", read_timeout=15
-                )
+                smart = self._sudo(f"smartctl -iA /dev/{name} 2>/dev/null", read_timeout=15)
                 if "PASSED" in smart:
                     health = "healthy"
                 elif "FAILED" in smart:
@@ -180,18 +174,20 @@ class OpenMediaVaultDriver(StorageDriver, LinuxDriver):
             except Exception:
                 pass
 
-            result.append(PhysicalDiskDict(
-                slot=f"/dev/{name}",
-                model=model,
-                serial=serial,
-                vendor=vendor,
-                type=disk_type,
-                size=size,
-                rpm=rpm,
-                temperature=temperature,
-                health=health,
-                pool=self._disk_to_pool(name),
-            ))
+            result.append(
+                PhysicalDiskDict(
+                    slot=f"/dev/{name}",
+                    model=model,
+                    serial=serial,
+                    vendor=vendor,
+                    type=disk_type,
+                    size=size,
+                    rpm=rpm,
+                    temperature=temperature,
+                    health=health,
+                    pool=self._disk_to_pool(name),
+                )
+            )
 
         return result
 
@@ -200,7 +196,7 @@ class OpenMediaVaultDriver(StorageDriver, LinuxDriver):
         try:
             mdstat = self._send("cat /proc/mdstat 2>/dev/null")
             for line in mdstat.splitlines():
-                if disk_name in line and re.match(r'^md\d+', line):
+                if disk_name in line and re.match(r"^md\d+", line):
                     return line.split(":")[0].strip()
         except Exception:
             pass
@@ -251,19 +247,17 @@ class OpenMediaVaultDriver(StorageDriver, LinuxDriver):
             return result
 
         for line in mdstat.splitlines():
-            m = re.match(r'^(md\d+)\s*:\s+(\S+)\s+(\S+)\s+(.*)', line)
+            m = re.match(r"^(md\d+)\s*:\s+(\S+)\s+(\S+)\s+(.*)", line)
             if not m:
                 continue
 
             md_name, state_word, level_raw, members_raw = m.groups()
-            member_disks = re.findall(r'(\w+)\[\d+\]', members_raw)
+            member_disks = re.findall(r"(\w+)\[\d+\]", members_raw)
             status = "online" if state_word == "active" else "offline"
             total = 0
 
             try:
-                detail = self._sudo(
-                    f"mdadm --detail /dev/{md_name} 2>/dev/null", read_timeout=10
-                )
+                detail = self._sudo(f"mdadm --detail /dev/{md_name} 2>/dev/null", read_timeout=10)
                 size_m = re.search(r"Array Size\s*:\s*(\d+)\s*\(", detail)
                 if size_m:
                     total = int(size_m.group(1)) * 1024  # kB → bytes
@@ -359,10 +353,13 @@ class OpenMediaVaultDriver(StorageDriver, LinuxDriver):
             key = mountpoint.lstrip("/").replace("/", "-") or "root"
             pool = device.removeprefix("/dev/")
 
-            readonly = bool(re.search(
-                rf'^{re.escape(device)}\s+\S+\s+\S+\s+(?:[^,\s]*,)*ro(?:,|\s|$)',
-                mounts, re.MULTILINE,
-            ))
+            readonly = bool(
+                re.search(
+                    rf"^{re.escape(device)}\s+\S+\s+\S+\s+(?:[^,\s]*,)*ro(?:,|\s|$)",
+                    mounts,
+                    re.MULTILINE,
+                )
+            )
 
             result[key] = LogicalVolumeDict(
                 name=key,
@@ -458,11 +455,11 @@ class OpenMediaVaultDriver(StorageDriver, LinuxDriver):
         """
         _SVC_MAP = [
             # (key, systemd unit, default port)
-            ("smb",   "smbd",       445),
-            ("nfs",   "nfs-server", 2049),
-            ("ftp",   "proftpd",    21),
-            ("ssh",   "ssh",        22),
-            ("rsync", "rsync",      873),
+            ("smb", "smbd", 445),
+            ("nfs", "nfs-server", 2049),
+            ("ftp", "proftpd", 21),
+            ("ssh", "ssh", 22),
+            ("rsync", "rsync", 873),
         ]
         result: dict[str, StorageServiceDict] = {}
         for key, unit, port in _SVC_MAP:
@@ -478,7 +475,11 @@ class OpenMediaVaultDriver(StorageDriver, LinuxDriver):
             except Exception:
                 running = enabled = False
             result[key] = StorageServiceDict(
-                name=key, enabled=enabled, running=running, port=port, version="",
+                name=key,
+                enabled=enabled,
+                running=running,
+                port=port,
+                version="",
             )
         return result
 
@@ -486,9 +487,7 @@ class OpenMediaVaultDriver(StorageDriver, LinuxDriver):
 
     def get_volume_snapshots(self, volume: str = "") -> list[VolumeSnapshotDict]:
         """Return volume snapshots (not supported on OpenMediaVault)."""
-        raise NotImplementedError(
-            "OpenMediaVault does not expose volume snapshots via this driver"
-        )
+        raise NotImplementedError("OpenMediaVault does not expose volume snapshots via this driver")
 
     def get_replication_jobs(self) -> list[ReplicationJobDict]:
         """Return replication jobs (not supported on OpenMediaVault)."""

pyproject.toml

@@ -41,6 +41,17 @@ Repository = "https://github.com/chrismanivong/napalm-openmediavault"
 where = ["."]
 include = ["napalm_openmediavault*"]
 
+[tool.ruff]
+line-length = 100
+target-version = "py310"
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "UP", "B"]
+
+[tool.ruff.lint.per-file-ignores]
+# Recorded device output keeps its lines.
+"tests/**" = ["E501"]
+
 [tool.mypy]
 python_version = "3.9"
 strict = true

tests/test_omv_type_label.py

@@ -9,10 +9,12 @@ def test_type_label_is_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"), \
+    assert not hasattr(OpenMediaVaultDriver, "storage_device"), (
         "storage_device is superseded by TYPE_LABEL = 'Storage'"
+    )