.gitea/workflows/ci.yml

@@ -0,0 +1,15 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+    tags: ["v*"]
+  pull_request:
+
+jobs:
+  driver:
+    uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
+    with:
+      standard-ref: main
+    secrets:
+      PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}

.gitignore

@@ -5,3 +5,6 @@ build/
 dist/
 .venv/
 .pytest_cache/
+.coverage
+.mypy_cache/
+.ruff_cache/

CHANGELOG.md

@@ -7,6 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.6.0] - 2026-10-23
+
+The first release on PyPI.
+
+### Changed
+- The driver follows the NAPALM driver standard
+  (https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
+  tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
+  a build check, and a release tag publishes to PyPI. Python 3.10 stays the floor; the
+  package now ships `py.typed`, and the license is declared as the SPDX string
+  `Apache-2.0`. napalm-device-types now comes from PyPI (4.21.1 or later).
+- Requires redfish 3.4.1 and pyghmi 1.5.1: older redfish releases return no answer
+  with `retries=0`, and pyghmi 1.5.0 misses a dependency it imports.
+- A failed IPMI logout on close is now logged at debug level.
+
 ## [0.5.0] - 2026-10-17
 
 ### Added

README.md

@@ -78,8 +78,6 @@ These follow the other drivers in the `NAPALM` organisation:
 - Parsing lives in pure functions, tested on recorded responses under
   `tests/fixtures/<vendor>-<model>-<firmware>/`, with a `COMMANDS.txt` (here: the request
   paths) listing what produced each file.
-- `napalm-device-types` is not on PyPI: install it from
-  `git+https://git.netork.io/NAPALM/napalm-device-types.git` before this package.
 
 ## Usage
 
@@ -87,10 +85,11 @@ These follow the other drivers in the `NAPALM` organisation:
 import napalm
 
 driver = napalm.get_network_driver("redfish")
-with driver("bmc01.example.net", "netork", "secret",
-            optional_args={"verify_ssl": "/etc/netork/bmc-ca.pem"}) as bmc:
-    bmc.get_facts()           # the BMC: firmware as os_version, its ports
-    bmc.get_power_state()     # the server: {"state": "On", "actions": [...]}
+with driver(
+    "bmc01.example.net", "netork", "secret", optional_args={"verify_ssl": "/etc/netork/bmc-ca.pem"}
+) as bmc:
+    bmc.get_facts()  # the BMC: firmware as os_version, its ports
+    bmc.get_power_state()  # the server: {"state": "On", "actions": [...]}
     bmc.get_hardware_inventory()
 ```
 
@@ -103,11 +102,12 @@ to basic auth only if the service cannot create sessions), `retries` (1).
 
 ## Development
 
+The rules are in [STANDARDS.md](https://git.netork.io/NAPALM/driver-standard/src/branch/main/STANDARDS.md).
+
 ```bash
 python -m venv .venv && . .venv/bin/activate
-pip install git+https://git.netork.io/NAPALM/napalm-device-types.git
-pip install -e '.[test]'
-pytest
+pip install -e '.[dev]'
+pytest --cov
 ```
 
 Fixtures from the DMTF mockups live under `tests/fixtures/dmtf-<mockup>/`, unmodified, with

docs/role.md

@@ -18,24 +18,30 @@ package updates and service control for a BMC.
 ```python
 class ManagementControllerDriver(DeviceTypeDriver):
     """Contract, not code."""
-    ROLE: str = "management_controller"   # key exposed as device_class
+
+    ROLE: str = "management_controller"  # key exposed as device_class
     TYPE_LABEL: str = "Management Controller"
-    USES_SSH: bool = False                # Redfish is HTTPS; IPMI is UDP
+    USES_SSH: bool = False  # Redfish is HTTPS; IPMI is UDP
 
     if TYPE_CHECKING:
         # The managed machine
         def get_power_state(self) -> PowerStateDict: ...
-        def set_power_state(self, action: PowerAction) -> None: ...   # On, GracefulShutdown,
-                                                                      # ForceOff, GracefulRestart,
-                                                                      # ForceRestart, PowerCycle
-        def get_hardware_inventory(self) -> HardwareInventoryDict: ...  # CPUs, DIMMs, drives,
-                                                                        # PSUs, NICs, serials
+        def set_power_state(self, action: PowerAction) -> None:
+            ...  # On, GracefulShutdown,
+            # ForceOff, GracefulRestart,
+            # ForceRestart, PowerCycle
+
+        def get_hardware_inventory(self) -> HardwareInventoryDict:
+            ...  # CPUs, DIMMs, drives,
+            # PSUs, NICs, serials
+
         def get_event_log(self, limit: int = 500) -> List[EventLogEntryDict]: ...  # SEL
         # The controller itself
-        def get_firmware_inventory(self) -> List[FirmwareDict]: ...    # BMC, BIOS, others
-        def get_management_protocols(self) -> ManagementProtocolsDict: ...  # IPMI, SSH,
-                                                                            # HTTP(S), SNMP,
-                                                                            # KVM, virtual media
+        def get_firmware_inventory(self) -> List[FirmwareDict]: ...  # BMC, BIOS, others
+        def get_management_protocols(self) -> ManagementProtocolsDict:
+            ...  # IPMI, SSH,
+            # HTTP(S), SNMP,
+            # KVM, virtual media
 ```
 
 Points to settle in napalm-device-types:

napalm_redfish/bmc.py

@@ -108,8 +108,7 @@ class BmcDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver):
             except (RedfishError, ConnectionException) as exc:
                 if not has_ipmi:
                     raise
-                LOG.info("%s: %s failed over Redfish (%s); asking IPMI", self.hostname, name,
-                         exc)
+                LOG.info("%s: %s failed over Redfish (%s); asking IPMI", self.hostname, name, exc)
         if not has_ipmi:
             raise NotImplementedError(f"{self.hostname}: {name} needs Redfish")
         return getattr(self._ipmi_conn(), name)(*args)

napalm_redfish/driver.py

@@ -190,8 +190,9 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
         except RedfishLicenseError as exc:
             self._license_refused = exc
         except RedfishError as exc:
-            raise ConnectionException(f"Cannot connect to Redfish at {self.hostname}: {exc}") \
-                from exc
+            raise ConnectionException(
+                f"Cannot connect to Redfish at {self.hostname}: {exc}"
+            ) from exc
 
     def close(self) -> None:
         self.transport.close()
@@ -271,26 +272,37 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
         return parse_network_protocol(raw or {})
 
     def _manager_interfaces(self) -> list[_JsonDict]:
-        return [parse_ethernet_interface(raw)
-                for raw in self._members_of(self._manager_raw(), "EthernetInterfaces")]
+        return [
+            parse_ethernet_interface(raw)
+            for raw in self._members_of(self._manager_raw(), "EthernetInterfaces")
+        ]
 
     # --- NAPALM getters ----------------------------------------------------------------
 
     def get_facts(self) -> _JsonDict:
         if self._license_refused is not None:
-            return {"uptime": -1.0, "vendor": "", "model": "", "hostname": self.hostname,
-                    "fqdn": self.hostname, "os_version": "", "serial_number": "",
-                    "interface_list": []}
+            return {
+                "uptime": -1.0,
+                "vendor": "",
+                "model": "",
+                "hostname": self.hostname,
+                "fqdn": self.hostname,
+                "os_version": "",
+                "serial_number": "",
+                "interface_list": [],
+            }
         system = parse_system(self._system_raw())
         manager = parse_manager(self._manager_raw())
         interfaces = self._manager_interfaces()
         protocol = self._network_protocol()
-        hostname = (protocol["hostname"]
-                    or next((i["hostname"] for i in interfaces if i["hostname"]), None)
-                    or self.hostname)
-        fqdn = (protocol["fqdn"]
-                or next((i["fqdn"] for i in interfaces if i["fqdn"]), None)
-                or hostname)
+        hostname = (
+            protocol["hostname"]
+            or next((i["hostname"] for i in interfaces if i["hostname"]), None)
+            or self.hostname
+        )
+        fqdn = (
+            protocol["fqdn"] or next((i["fqdn"] for i in interfaces if i["fqdn"]), None) or hostname
+        )
         return {
             "uptime": -1.0,  # Redfish has no uptime
             "vendor": system["manufacturer"] or "",
@@ -327,11 +339,13 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
             fans, temperature = legacy["fans"], legacy["temperature"]
         elif thermal:
             subsystem = parse_thermal_subsystem(self.transport.get(thermal["path"]))
-            parsed_fans = [parse_fan(raw) for raw in self.transport.members(
-                subsystem["fans_path"])] if subsystem["fans_path"] else []
+            parsed_fans = (
+                [parse_fan(raw) for raw in self.transport.members(subsystem["fans_path"])]
+                if subsystem["fans_path"]
+                else []
+            )
             if sources["sensors"]:
-                sensors = [parse_sensor(raw)
-                           for raw in self.transport.members(sources["sensors"])]
+                sensors = [parse_sensor(raw) for raw in self.transport.members(sources["sensors"])]
                 temperature = build_environment(sensors=sensors)["temperature"]
             elif subsystem["thermal_metrics_path"]:
                 metrics = self.transport.get_optional(subsystem["thermal_metrics_path"])
@@ -342,12 +356,13 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
         if supply and supply["schema"] == "Power":
             power = parse_power(self.transport.get(supply["path"]))["power"]
         elif supply:
-            psus = [parse_power_supply(raw, self.transport.follow(raw, "Metrics"))
-                    for raw in self._power_supplies_raw(supply["path"])]
+            psus = [
+                parse_power_supply(raw, self.transport.follow(raw, "Metrics"))
+                for raw in self._power_supplies_raw(supply["path"])
+            ]
             power = build_environment(power_supplies=psus)["power"]
 
-        return {"fans": fans, "temperature": temperature, "power": power,
-                "cpu": {}, "memory": {}}
+        return {"fans": fans, "temperature": temperature, "power": power, "cpu": {}, "memory": {}}
 
     def _power_supplies_raw(self, subsystem_path: str) -> list[_JsonDict]:
         subsystem = parse_power_subsystem(self.transport.get(subsystem_path))
@@ -358,8 +373,7 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
         service = self._service("AccountService")
         if service is None:
             return {}
-        return to_napalm_users(parse_account(raw) for raw in self._members_of(service,
-                                                                               "Accounts"))
+        return to_napalm_users(parse_account(raw) for raw in self._members_of(service, "Accounts"))
 
     def get_ntp_servers(self) -> _JsonDict:
         return {server: {} for server in self._network_protocol()["ntp"]["servers"]}
@@ -387,15 +401,23 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
         """
         refused = self._license_refused
         if refused is not None:
-            return [{"code": "bmc_license_required",
-                     "meta": {"licenses": refused.licenses, "message": refused.detail}}]
+            return [
+                {
+                    "code": "bmc_license_required",
+                    "meta": {"licenses": refused.licenses, "message": refused.detail},
+                }
+            ]
         self._probe_withheld()
         if not self._withheld:
             return []
         licenses = sorted({name for names in self._withheld.values() for name in names.split()})
         parts = [part for part in _WITHHELD_ORDER if part in self._withheld]
-        return [{"code": "bmc_license_partial",
-                 "meta": {"licenses": " ".join(licenses), "parts": parts}}]
+        return [
+            {
+                "code": "bmc_license_partial",
+                "meta": {"licenses": " ".join(licenses), "parts": parts},
+            }
+        ]
 
     def get_licenses(self) -> list[_JsonDict]:
         """The licenses the BMC asks for and does not have (``status`` "missing").
@@ -407,8 +429,10 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
         """
         refused = self._license_refused
         if refused is not None:
-            return [{"name": name, "status": "missing", "detail": refused.detail}
-                    for name in refused.licenses.split() or [""]]
+            return [
+                {"name": name, "status": "missing", "detail": refused.detail}
+                for name in refused.licenses.split() or [""]
+            ]
         self._probe_withheld()
         withheld: dict[str, list[str]] = {}
         for part in _WITHHELD_ORDER:
@@ -416,8 +440,10 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
                 continue
             for name in self._withheld[part].split() or [""]:
                 withheld.setdefault(name, []).append(part)
-        return [{"name": name, "status": "missing", "withholds": parts}
-                for name, parts in withheld.items()]
+        return [
+            {"name": name, "status": "missing", "withholds": parts}
+            for name, parts in withheld.items()
+        ]
 
     def _withhold(self, part: str, exc: RedfishFeatureLicenseError) -> None:
         LOG.info("%s: %s withheld by the BMC: %s", self.hostname, part, exc)
@@ -460,8 +486,7 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
 
     def get_power_state(self) -> _JsonDict:
         system = self.transport.get(self._system_path())
-        return {"state": parse_power_state(system) or "",
-                "actions": self._accepted_actions(system)}
+        return {"state": parse_power_state(system) or "", "actions": self._accepted_actions(system)}
 
     def set_power_state(self, action: str) -> None:
         if action not in POWER_ACTIONS:
@@ -475,7 +500,8 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
             raise RuntimeError(
                 f"{self.hostname}: {action} not accepted while the server is "
                 f"{parse_power_state(system) or 'in an unknown state'} "
-                f"(accepted: {', '.join(accepted) or 'none'})")
+                f"(accepted: {', '.join(accepted) or 'none'})"
+            )
         try:
             self.transport.post(target, {"ResetType": action})
         except RedfishError as exc:
@@ -490,11 +516,13 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
             raise RuntimeError(f"{self.hostname}: the BMC offers no reset action")
         body: _JsonDict = {}
         if reset["allowed"] is not None:
-            reset_type = next((t for t in ("GracefulRestart", "ForceRestart")
-                               if t in reset["allowed"]), None)
+            reset_type = next(
+                (t for t in ("GracefulRestart", "ForceRestart") if t in reset["allowed"]), None
+            )
             if reset_type is None:
-                raise RuntimeError(f"{self.hostname}: BMC reset types "
-                                   f"{reset['allowed']} include no restart")
+                raise RuntimeError(
+                    f"{self.hostname}: BMC reset types {reset['allowed']} include no restart"
+                )
             body["ResetType"] = reset_type
         try:
             self.transport.post(reset["target"], body)
@@ -524,9 +552,15 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
             cpu = parse_processor(raw)
             if cpu["processor_type"] not in (None, "CPU"):
                 continue  # GPUs, FPGAs, accelerators
-            cpus.append({"socket": cpu["socket"] or cpu["id"] or "", "model": cpu["model"] or "",
-                         "cores": cpu["cores"], "threads": cpu["threads"],
-                         "status": _status(cpu)})
+            cpus.append(
+                {
+                    "socket": cpu["socket"] or cpu["id"] or "",
+                    "model": cpu["model"] or "",
+                    "cores": cpu["cores"],
+                    "threads": cpu["threads"],
+                    "status": _status(cpu),
+                }
+            )
         return cpus
 
     def _dimms(self, system: _JsonDict) -> list[_JsonDict]:
@@ -534,15 +568,17 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
         for raw in self._members_of(system, "Memory"):
             dimm = parse_memory(raw)
             size = dimm["capacity_mib"]
-            dimms.append({
-                "slot": dimm["location"] or dimm["id"] or "",
-                "size_bytes": size * 1024 * 1024 if size else None,
-                "type": dimm["memory_device_type"] or "",
-                "speed_mts": dimm["speed_mhz"],  # Redfish: MHz or MT/s "as appropriate"
-                "part_number": dimm["part_number"] or "",
-                "serial_number": dimm["serial_number"] or "",
-                "status": _status(dimm),
-            })
+            dimms.append(
+                {
+                    "slot": dimm["location"] or dimm["id"] or "",
+                    "size_bytes": size * 1024 * 1024 if size else None,
+                    "type": dimm["memory_device_type"] or "",
+                    "speed_mts": dimm["speed_mhz"],  # Redfish: MHz or MT/s "as appropriate"
+                    "part_number": dimm["part_number"] or "",
+                    "serial_number": dimm["serial_number"] or "",
+                    "status": _status(dimm),
+                }
+            )
         return dimms
 
     def _storage_drives(self, system: _JsonDict) -> list[_JsonDict] | None:
@@ -572,15 +608,18 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
             parsed = []
             for storage in self._members_of(system, "SimpleStorage"):
                 parsed.extend(parse_simple_storage(storage)["drives"])
-        return [{
-            "slot": d["name"] or d["id"] or "",
-            "model": d["model"] or "",
-            "serial_number": d["serial_number"] or "",
-            "size_bytes": d["capacity_bytes"],
-            "media": d["media_type"] or "",
-            "protocol": d["protocol"] or "",
-            "status": _status(d),
-        } for d in parsed]
+        return [
+            {
+                "slot": d["name"] or d["id"] or "",
+                "model": d["model"] or "",
+                "serial_number": d["serial_number"] or "",
+                "size_bytes": d["capacity_bytes"],
+                "media": d["media_type"] or "",
+                "protocol": d["protocol"] or "",
+                "status": _status(d),
+            }
+            for d in parsed
+        ]
 
     def _psus(self) -> list[_JsonDict]:
         source = environment_sources(self._chassis_raw())["power"]
@@ -595,13 +634,15 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
             if not isinstance(raw, dict):
                 continue
             psu = parse_power_supply(raw)
-            psus.append({
-                "slot": psu["name"],
-                "model": psu["model"] or "",
-                "serial_number": psu["serial_number"] or "",
-                "capacity_watts": int(psu["capacity"]) if psu["capacity"] >= 0 else None,
-                "status": _status(psu),
-            })
+            psus.append(
+                {
+                    "slot": psu["name"],
+                    "model": psu["model"] or "",
+                    "serial_number": psu["serial_number"] or "",
+                    "capacity_watts": int(psu["capacity"]) if psu["capacity"] >= 0 else None,
+                    "status": _status(psu),
+                }
+            )
         return psus
 
     def _nics(self, system: _JsonDict) -> list[_JsonDict]:
@@ -610,13 +651,16 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
             nic = parse_ethernet_interface(raw)
             speed = nic["interface"]["speed"]
             link_up = {"LinkUp": True, "LinkDown": False, "NoLink": False}.get(
-                nic["link_status"] or "")
-            nics.append({
-                "name": nic["name"] or "",
-                "mac_address": nic["interface"]["mac_address"].lower(),
-                "speed_mbps": int(speed) if speed >= 0 else None,
-                "link_up": link_up,
-            })
+                nic["link_status"] or ""
+            )
+            nics.append(
+                {
+                    "name": nic["name"] or "",
+                    "mac_address": nic["interface"]["mac_address"].lower(),
+                    "speed_mbps": int(speed) if speed >= 0 else None,
+                    "link_up": link_up,
+                }
+            )
         return nics
 
     # --- role: event log, firmware, protocols --------------------------------------------
@@ -640,20 +684,24 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
                 try:
                     raw_entries = self.transport.members(service["entries_path"])
                 except RedfishNotFound:
-                    LOG.warning("%s: log %s has no entries collection", self.hostname,
-                                service["id"])
+                    LOG.warning(
+                        "%s: log %s has no entries collection", self.hostname, service["id"]
+                    )
                     continue
                 for raw in raw_entries:
                     entry = parse_log_entry(raw)
                     severity = entry["severity"]
-                    entries.append({
-                        "id": odata_id(raw) or f"{service['id']}/{entry['id']}",
-                        "timestamp": _timestamp(entry["created"]),
-                        "severity": severity if severity in _SEVERITIES else "OK",
-                        "message": entry["message"] or "",
-                    })
-        entries.sort(key=lambda e: (e["timestamp"] is not None, e["timestamp"] or 0.0),
-                     reverse=True)
+                    entries.append(
+                        {
+                            "id": odata_id(raw) or f"{service['id']}/{entry['id']}",
+                            "timestamp": _timestamp(entry["created"]),
+                            "severity": severity if severity in _SEVERITIES else "OK",
+                            "message": entry["message"] or "",
+                        }
+                    )
+        entries.sort(
+            key=lambda e: (e["timestamp"] is not None, e["timestamp"] or 0.0), reverse=True
+        )
         return entries[:limit]
 
     def _update_service(self) -> _JsonDict | None:
@@ -673,11 +721,20 @@ class RedfishDriver(HealthMetricsMixin, LicenseMixin, ManagementControllerDriver
                 return []
             bmc = parse_manager(self._manager_raw())["firmware_version"]
             bios = self._system_raw().get("BiosVersion")
-            return [{"name": name, "version": version, "updateable": None}
-                    for name, version in (("BMC", bmc), ("BIOS", bios)) if version]
+            return [
+                {"name": name, "version": version, "updateable": None}
+                for name, version in (("BMC", bmc), ("BIOS", bios))
+                if version
+            ]
         items = [parse_firmware(raw) for raw in self._members_of(service, "FirmwareInventory")]
-        return [{"name": fw["name"] or fw["id"] or "", "version": fw["version"] or "",
-                 "updateable": fw["updateable"]} for fw in items]
+        return [
+            {
+                "name": fw["name"] or fw["id"] or "",
+                "version": fw["version"] or "",
+                "updateable": fw["updateable"],
+            }
+            for fw in items
+        ]
 
     def get_management_protocols(self) -> _JsonDict:
         """What the BMC listens on. A protocol the BMC does not report counts as off."""

napalm_redfish/ipmi.py

@@ -8,14 +8,19 @@ sensors, the power state and the system event log, and switches the server's pow
 
 from __future__ import annotations
 
+import logging
 from typing import Any
 
 from napalm.base.exceptions import ConnectionException
 from napalm_device_types import HealthMetricsMixin, ManagementControllerDriver
 from napalm_device_types.models import POWER_ACTIONS
-from pyghmi.exceptions import IpmiException
-from pyghmi.ipmi.command import Command
-from pyghmi.ipmi.private.session import Session
+
+# IPMI is this driver's protocol: the fallback for BMCs without Redfish (see above).
+from pyghmi.exceptions import IpmiException  # nosec B415
+from pyghmi.ipmi.command import Command  # nosec B415
+from pyghmi.ipmi.private.session import Session  # nosec B415
+
+LOG = logging.getLogger(__name__)
 
 _JsonDict = dict[str, Any]
 
@@ -85,8 +90,7 @@ class IpmiDriver(HealthMetricsMixin, ManagementControllerDriver):
                 port=self.port,
             )
         except Exception as exc:
-            raise ConnectionException(f"Cannot connect to IPMI at {self.hostname}: {exc}") \
-                from exc
+            raise ConnectionException(f"Cannot connect to IPMI at {self.hostname}: {exc}") from exc
 
     def close(self) -> None:
         ipmi, self._ipmi = self._ipmi, None
@@ -94,7 +98,7 @@ class IpmiDriver(HealthMetricsMixin, ManagementControllerDriver):
             try:
                 ipmi.ipmi_session.logout()
             except Exception:  # a session left open times out on the BMC
-                pass
+                LOG.debug("%s: ignored", self.hostname, exc_info=True)
         self._forget_session()
 
     def _forget_session(self) -> None:
@@ -219,7 +223,8 @@ class IpmiDriver(HealthMetricsMixin, ManagementControllerDriver):
         if action not in accepted:
             raise RuntimeError(
                 f"{self.hostname}: {action} not accepted while the server is "
-                f"{state or 'in an unknown state'} (accepted: {', '.join(accepted) or 'none'})")
+                f"{state or 'in an unknown state'} (accepted: {', '.join(accepted) or 'none'})"
+            )
         try:
             self._cmd().set_power(_CHASSIS_CONTROL[action], wait=False)
         except IpmiException as exc:
@@ -235,7 +240,8 @@ class IpmiDriver(HealthMetricsMixin, ManagementControllerDriver):
                 "serial_number": "",
                 "capacity_watts": None,
                 "status": _severity(int(reading.health or 0))
-                if _supply_present(reading) else "Absent",
+                if _supply_present(reading)
+                else "Absent",
             }
             for reading in self._sensors()
             if reading.type == "Power Supply"
@@ -259,14 +265,17 @@ class IpmiDriver(HealthMetricsMixin, ManagementControllerDriver):
             component = event.get("component") or ""
             text = event.get("event") or ""
             timecode = event.get("timecode")
-            entries.append({
-                "id": str(event.get("record_id", "")),
-                "timestamp": float(timecode) if timecode else None,
-                "severity": _severity(int(event.get("severity") or 0)),
-                "message": f"{component}: {text}" if component and text else component or text,
-            })
-        entries.sort(key=lambda e: (e["timestamp"] is not None, e["timestamp"] or 0.0),
-                     reverse=True)
+            entries.append(
+                {
+                    "id": str(event.get("record_id", "")),
+                    "timestamp": float(timecode) if timecode else None,
+                    "severity": _severity(int(event.get("severity") or 0)),
+                    "message": f"{component}: {text}" if component and text else component or text,
+                }
+            )
+        entries.sort(
+            key=lambda e: (e["timestamp"] is not None, e["timestamp"] or 0.0), reverse=True
+        )
         return entries[:limit]
 
     def get_firmware_inventory(self) -> list[_JsonDict]:

napalm_redfish/parsers/accounts.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Any, Iterable, Mapping
+from collections.abc import Iterable, Mapping
+from typing import Any
 
 from .common import boolean, text
 
@@ -34,7 +35,8 @@ def to_napalm_users(accounts: Iterable[Mapping[str, Any] | None]) -> dict[str, d
             continue
         users[acc["username"]] = {
             "level": ROLE_LEVELS.get(acc.get("role_id") or "", 0),
-            "password": "",
+            # NAPALM's get_users field; Redfish never returns a password, so it stays empty.
+            "password": "",  # nosec B105
             "sshkeys": [],
         }
     return users

napalm_redfish/parsers/common.py

@@ -2,21 +2,24 @@
 
 from __future__ import annotations
 
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 
 # DMI defaults that board vendors leave in place (Supermicro boards ship most of them).
-PLACEHOLDERS = frozenset({
-    "to be filled by o.e.m.",
-    "default string",
-    "not specified",
-    "not applicable",
-    "n/a",
-    "none",
-    "0123456789",
-    "system serial number",
-    "system product name",
-    "system manufacturer",
-})
+PLACEHOLDERS = frozenset(
+    {
+        "to be filled by o.e.m.",
+        "default string",
+        "not specified",
+        "not applicable",
+        "n/a",
+        "none",
+        "0123456789",
+        "system serial number",
+        "system product name",
+        "system manufacturer",
+    }
+)
 
 
 def text(value: Any) -> str | None:

napalm_redfish/parsers/environment.py

@@ -16,12 +16,20 @@ left out.
 
 from __future__ import annotations
 
-from typing import Any, Iterable, Mapping
+from collections.abc import Iterable, Mapping
+from typing import Any
 
 from .common import link, number, present, status, text, unique_name
 
-_SENSOR_TYPES_BY_UNITS = {"Cel": "Temperature", "V": "Voltage", "W": "Power", "A": "Current",
-                          "RPM": "Rotational", "%": "Percent", "Hz": "Frequency"}
+_SENSOR_TYPES_BY_UNITS = {
+    "Cel": "Temperature",
+    "V": "Voltage",
+    "W": "Power",
+    "A": "Current",
+    "RPM": "Rotational",
+    "%": "Percent",
+    "Hz": "Frequency",
+}
 
 
 def environment_sources(chassis: Mapping[str, Any]) -> dict[str, Any]:
@@ -138,8 +146,11 @@ def parse_power(power: Mapping[str, Any]) -> dict[str, Any]:
         if not psu["present"]:
             continue
         name = unique_name(supplies, psu["name"], item.get("MemberId", index))
-        supplies[name] = {"status": psu["status"], "capacity": psu["capacity"],
-                          "output": psu["output"]}
+        supplies[name] = {
+            "status": psu["status"],
+            "capacity": psu["capacity"],
+            "output": psu["output"],
+        }
 
     voltages: dict[str, Any] = {}
     for index, item in _members(power, "Voltages"):
@@ -204,8 +215,9 @@ def parse_fan(fan: Mapping[str, Any]) -> dict[str, Any]:
     }
 
 
-def parse_power_supply(supply: Mapping[str, Any],
-                       metrics: Mapping[str, Any] | None = None) -> dict[str, Any]:
+def parse_power_supply(
+    supply: Mapping[str, Any], metrics: Mapping[str, Any] | None = None
+) -> dict[str, Any]:
     """One member of ``PowerSubsystem/PowerSupplies``, with its ``Metrics`` if read.
 
     An item of the legacy ``Power.PowerSupplies`` array parses too (``MemberId`` as id,
@@ -273,16 +285,24 @@ def parse_thermal_metrics(metrics: Mapping[str, Any]) -> dict[str, Any]:
     return {"temperature": temperature}
 
 
-def build_environment(*, fans: Iterable[Mapping[str, Any]] = (),
-                      power_supplies: Iterable[Mapping[str, Any]] = (),
-                      sensors: Iterable[Mapping[str, Any]] = ()) -> dict[str, Any]:
+def build_environment(
+    *,
+    fans: Iterable[Mapping[str, Any]] = (),
+    power_supplies: Iterable[Mapping[str, Any]] = (),
+    sensors: Iterable[Mapping[str, Any]] = (),
+) -> dict[str, Any]:
     """Combine parsed fans, power supplies and sensors into the shape of the legacy parsers.
 
     ``consumed_watts`` comes from the chassis total power sensor
     (``ReadingType`` Power, ``ElectricalContext`` Total, ``PhysicalContext`` Chassis).
     """
-    env: dict[str, Any] = {"fans": {}, "temperature": {}, "power": {}, "voltages": {},
-                           "consumed_watts": None}
+    env: dict[str, Any] = {
+        "fans": {},
+        "temperature": {},
+        "power": {},
+        "voltages": {},
+        "consumed_watts": None,
+    }
     for fan in fans:
         if fan["present"]:
             name = unique_name(env["fans"], fan["name"], fan["id"])
@@ -290,8 +310,11 @@ def build_environment(*, fans: Iterable[Mapping[str, Any]] = (),
     for psu in power_supplies:
         if psu["present"]:
             name = unique_name(env["power"], psu["name"], psu["id"])
-            env["power"][name] = {"status": psu["status"], "capacity": psu["capacity"],
-                                  "output": psu["output"]}
+            env["power"][name] = {
+                "status": psu["status"],
+                "capacity": psu["capacity"],
+                "output": psu["output"],
+            }
     for sensor in sensors:
         reading = sensor["reading"]
         if reading is None:
@@ -303,7 +326,11 @@ def build_environment(*, fans: Iterable[Mapping[str, Any]] = (),
         elif sensor["reading_type"] == "Voltage":
             name = unique_name(env["voltages"], sensor["name"], sensor["id"])
             env["voltages"][name] = {"voltage": reading, **flags}
-        elif (sensor["reading_type"] == "Power" and sensor["electrical_context"] == "Total"
-              and sensor["physical_context"] == "Chassis" and env["consumed_watts"] is None):
+        elif (
+            sensor["reading_type"] == "Power"
+            and sensor["electrical_context"] == "Total"
+            and sensor["physical_context"] == "Chassis"
+            and env["consumed_watts"] is None
+        ):
             env["consumed_watts"] = reading
     return env

napalm_redfish/parsers/firmware.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 
 from .common import boolean, link_list, status, text
 

napalm_redfish/parsers/inventory.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 
 from .common import boolean, ident, integer, link, link_list, location_label, present, status, text
 
@@ -39,8 +40,9 @@ def parse_memory(dimm: Mapping[str, Any]) -> dict[str, Any]:
     return {
         "id": text(dimm.get("Id")),
         "name": text(dimm.get("Name")),
-        "location": (location_label(dimm) or text(dimm.get("DeviceLocator"))
-                     or text(dimm.get("Name"))),
+        "location": (
+            location_label(dimm) or text(dimm.get("DeviceLocator")) or text(dimm.get("Name"))
+        ),
         "capacity_mib": capacity,
         "memory_device_type": text(dimm.get("MemoryDeviceType")),
         "base_module_type": text(dimm.get("BaseModuleType")),
@@ -62,7 +64,8 @@ def _controller(ctrl: Mapping[str, Any]) -> dict[str, Any]:
         "model": ident(ctrl.get("Model")),
         "serial_number": ident(ctrl.get("SerialNumber")),
         "firmware_version": text(ctrl.get("FirmwareVersion")),
-        "speed_gbps": ctrl.get("SpeedGbps") if isinstance(ctrl.get("SpeedGbps"), (int, float))
+        "speed_gbps": ctrl.get("SpeedGbps")
+        if isinstance(ctrl.get("SpeedGbps"), (int, float))
         else None,
         "state": state,
         "health": health,
@@ -83,7 +86,8 @@ def parse_storage(storage: Mapping[str, Any]) -> dict[str, Any]:
         "state": state,
         "health": health,
         "controllers": [_controller(c) for c in controllers if isinstance(c, Mapping)]
-        if isinstance(controllers, list) else [],
+        if isinstance(controllers, list)
+        else [],
         "controllers_path": link(storage, "Controllers"),
         "drive_paths": link_list(storage, "Drives"),
     }
@@ -117,8 +121,11 @@ def parse_simple_storage(storage: Mapping[str, Any]) -> dict[str, Any]:
     """
     state, health = status(storage)
     devices = storage.get("Devices")
-    drives = [parse_drive(dev) for dev in devices if isinstance(dev, Mapping)] \
-        if isinstance(devices, list) else []
+    drives = (
+        [parse_drive(dev) for dev in devices if isinstance(dev, Mapping)]
+        if isinstance(devices, list)
+        else []
+    )
     return {
         "id": text(storage.get("Id")),
         "name": text(storage.get("Name")),

napalm_redfish/parsers/logs.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 
 from .common import boolean, integer, link, status, text
 

napalm_redfish/parsers/manager.py

@@ -3,7 +3,8 @@
 from __future__ import annotations
 
 import ipaddress
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 
 from .common import boolean, integer, link_list, links, number, status, text
 
@@ -132,8 +133,9 @@ def parse_network_protocol(proto: Mapping[str, Any]) -> dict[str, Any]:
     for community in snmp.get("CommunityStrings") or []:
         if isinstance(community, Mapping):
             # The community string itself is never kept, even if a BMC returns it.
-            communities.append({"name": text(community.get("Name")),
-                                "access": text(community.get("AccessMode"))})
+            communities.append(
+                {"name": text(community.get("Name")), "access": text(community.get("AccessMode"))}
+            )
 
     return {
         "hostname": text(proto.get("HostName")),

napalm_redfish/parsers/service_root.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 
 from .common import link, links, text
 

napalm_redfish/parsers/system.py

@@ -2,7 +2,8 @@
 
 from __future__ import annotations
 
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 
 from .common import ident, integer, link_list, links, number, status, text
 
@@ -10,10 +11,12 @@ from .common import ident, integer, link_list, links, number, status, text
 def parse_system(system: Mapping[str, Any]) -> dict[str, Any]:
     """Identity, summaries and links of the managed server."""
     state, health = status(system)
-    cpus = system.get("ProcessorSummary") if isinstance(system.get("ProcessorSummary"),
-                                                        Mapping) else {}
-    memory = system.get("MemorySummary") if isinstance(system.get("MemorySummary"),
-                                                       Mapping) else {}
+    cpus = (
+        system.get("ProcessorSummary")
+        if isinstance(system.get("ProcessorSummary"), Mapping)
+        else {}
+    )
+    memory = system.get("MemorySummary") if isinstance(system.get("MemorySummary"), Mapping) else {}
     return {
         "id": text(system.get("Id")),
         "name": text(system.get("Name")),
@@ -44,8 +47,9 @@ def parse_power_state(system: Mapping[str, Any]) -> str | None:
     return text(system.get("PowerState"))
 
 
-def parse_reset_action(resource: Mapping[str, Any],
-                       action: str = "#ComputerSystem.Reset") -> dict[str, Any]:
+def parse_reset_action(
+    resource: Mapping[str, Any], action: str = "#ComputerSystem.Reset"
+) -> dict[str, Any]:
     """Target and allowed ``ResetType`` values of a reset action.
 
     ``allowed`` is ``None`` when the service does not list the values inline. Then

napalm_redfish/transport.py

@@ -20,7 +20,8 @@ import ipaddress
 import logging
 import os
 import re
-from typing import Any, Mapping
+from collections.abc import Mapping
+from typing import Any
 from urllib.parse import urlsplit
 
 import requests
@@ -80,8 +81,14 @@ class RedfishFeatureLicenseError(RedfishAuthError):
     refuses ``Systems/1/Storage`` and ``UpdateService`` for want of DCMS.
     """
 
-    def __init__(self, message: str, *, licenses: str = "", status: int | None = None,
-                 path: str | None = None):
+    def __init__(
+        self,
+        message: str,
+        *,
+        licenses: str = "",
+        status: int | None = None,
+        path: str | None = None,
+    ):
         super().__init__(message, status=status, path=path)
         #: The licenses the service names, e.g. "DCMS"; empty when it names none.
         self.licenses = licenses
@@ -270,8 +277,9 @@ class RedfishTransport:
                 return "session"
             if self.auth == "session":
                 raise RedfishAuthError(f"{self.base_url}: session login returned no X-Auth-Token")
-            LOG.info("%s: session login returned no token, falling back to basic auth",
-                     self.base_url)
+            LOG.info(
+                "%s: session login returned no token, falling back to basic auth", self.base_url
+            )
         self._call(client.login, auth=AuthMethod.BASIC)
         return "basic"
 
@@ -281,8 +289,9 @@ class RedfishTransport:
         except InvalidCredentialsError as exc:
             raise RedfishAuthError(f"{self.base_url}: invalid credentials", status=401) from exc
         except RedfishPasswordChangeRequiredError as exc:
-            raise RedfishAuthError(f"{self.base_url}: the account must change its password "
-                                   "before it can be used") from exc
+            raise RedfishAuthError(
+                f"{self.base_url}: the account must change its password before it can be used"
+            ) from exc
         except SessionCreationError as exc:
             raise _SessionUnsupported(str(exc).strip()) from exc
         except (RetriesExhaustedError, ServerDownOrUnreachableError) as exc:
@@ -300,7 +309,7 @@ class RedfishTransport:
             LOG.warning("%s: could not delete the Redfish session: %s", self.base_url, exc)
             client._session.close()
 
-    def __enter__(self) -> "RedfishTransport":
+    def __enter__(self) -> RedfishTransport:
         self.open()
         return self
 
@@ -340,20 +349,25 @@ class RedfishTransport:
             data = resp.dict
         except Exception as exc:
             if status < 400:
-                raise RedfishError(f"{method} {path}: response is not JSON",
-                                   status=status, path=path) from exc
+                raise RedfishError(
+                    f"{method} {path}: response is not JSON", status=status, path=path
+                ) from exc
             data = {}
         if status < 400:
             if not isinstance(data, dict):
-                raise RedfishError(f"{method} {path}: response is not a JSON object",
-                                   status=status, path=path)
+                raise RedfishError(
+                    f"{method} {path}: response is not a JSON object", status=status, path=path
+                )
             return data
         detail = _error_message(data)
         message = f"{method} {path}: HTTP {status}" + (f": {detail}" if detail else "")
         if status == 403 and any(i.endswith(LICENSE_REFUSED) for i in _message_ids(data)):
             raise RedfishFeatureLicenseError(
-                message, licenses=" ".join(" ".join(_message_args(data)).split()),
-                status=status, path=path)
+                message,
+                licenses=" ".join(" ".join(_message_args(data)).split()),
+                status=status,
+                path=path,
+            )
         if status in (401, 403):
             raise RedfishAuthError(message, status=status, path=path)
         if status == 404:

pyproject.toml

@@ -1,33 +1,36 @@
 [build-system]
-requires = ["setuptools>=64"]
+requires = ["setuptools>=77"]
 build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-redfish"
-version = "0.5.0"
+version = "0.6.0"
 description = "NAPALM driver for server baseboard management controllers over DMTF Redfish"
 readme = "README.md"
-license = { file = "LICENSE" }
+license = "Apache-2.0"
 requires-python = ">=3.10"
 dependencies = [
     "napalm>=5.0",
-    # Not on PyPI: install from git before this package (see README).
-    "napalm-device-types>=4.15",
-    "redfish>=3.2",
-    "pyghmi>=1.5",
+    "napalm-device-types>=4.21.1",
+    # Before 3.4.1, max_retry=0 (retries=0) gets no response back at all.
+    "redfish>=3.4.1",
+    # 1.5.0 imports python-dateutil without declaring it.
+    "pyghmi>=1.5.1",
 ]
 classifiers = [
-    "License :: OSI Approved :: Apache Software License",
     "Programming Language :: Python :: 3",
     "Topic :: System :: Hardware",
     "Topic :: System :: Networking",
 ]
 
 [project.optional-dependencies]
-test = [
-    "pytest>=7",
-    "responses>=0.23",
-]
+# responses mocks the BMC's HTTP answers in the tests.
+dev = ["pytest>=8", "pytest-cov>=5", "mypy>=1.11", "ruff>=0.6", "responses>=0.23"]
+
+[project.urls]
+Repository = "https://git.netork.io/NAPALM/napalm-redfish"
+Issues = "https://git.netork.io/NAPALM/napalm-redfish/issues"
+Changelog = "https://git.netork.io/NAPALM/napalm-redfish/src/branch/main/CHANGELOG.md"
 
 [project.entry-points."napalm.drivers"]
 redfish = "napalm_redfish:RedfishDriver"
@@ -37,5 +40,51 @@ bmc = "napalm_redfish.bmc:BmcDriver"
 [tool.setuptools.packages.find]
 include = ["napalm_redfish*"]
 
+[tool.setuptools.package-data]
+napalm_redfish = ["py.typed"]
+
+[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.10"
+strict = true
+# napalm itself is untyped, so every driver subclasses Any.
+disallow_subclassing_any = false
+
+[[tool.mypy.overrides]]
+module = ["napalm.*", "redfish.*", "pyghmi.*", "requests.*"]
+ignore_missing_imports = true
+
+[[tool.mypy.overrides]]
+# Not yet under strict mypy (quality baseline); may only shrink.
+module = [
+    "napalm_redfish.bmc",
+    "napalm_redfish.driver",
+    "napalm_redfish.ipmi",
+    "napalm_redfish.parsers.common",
+    "napalm_redfish.parsers.environment",
+    "napalm_redfish.parsers.manager",
+    "napalm_redfish.parsers.system",
+    "napalm_redfish.transport",
+]
+ignore_errors = true
+
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+# The tests import their helpers as tests.conftest, also under a bare pytest.
+pythonpath = ["."]
+
+[tool.coverage.run]
+source = ["napalm_redfish"]
+
+[tool.coverage.report]
+fail_under = 94

tests/conftest.py

@@ -17,7 +17,7 @@ def fixture_file(mockup: str, path: str) -> Path:
     path = path.split("#", 1)[0].split("?", 1)[0].rstrip("/")
     if not path.startswith(PREFIX):
         raise ValueError(f"not a Redfish path: {path}")
-    relative = path[len(PREFIX):].lstrip("/")
+    relative = path[len(PREFIX) :].lstrip("/")
     return FIXTURES / mockup / relative / "index.json"
 
 
@@ -64,8 +64,14 @@ class MockupServer:
         for method in (responses.GET, responses.POST, responses.DELETE):
             rsps.add_callback(method, anything, callback=self._handle)
 
-    def override(self, method: str, path: str, status: int, body: dict | str | None = None,
-                 headers: dict | None = None):
+    def override(
+        self,
+        method: str,
+        path: str,
+        status: int,
+        body: dict | str | None = None,
+        headers: dict | None = None,
+    ):
         self.overrides[(method, path)] = (status, body if body is not None else {}, headers or {})
 
     @staticmethod
@@ -82,8 +88,11 @@ class MockupServer:
             if key in self.overrides:
                 return self._reply(*self.overrides[key])
         if request.method == "POST" and parts.path == SESSIONS:
-            return self._reply(201, {"@odata.id": SESSION_PATH, "Id": "0123456789abcdef"},
-                               {"X-Auth-Token": TOKEN, "Location": BASE_URL + SESSION_PATH})
+            return self._reply(
+                201,
+                {"@odata.id": SESSION_PATH, "Id": "0123456789abcdef"},
+                {"X-Auth-Token": TOKEN, "Location": BASE_URL + SESSION_PATH},
+            )
         if request.method == "DELETE" and parts.path == SESSION_PATH:
             return self._reply(204, {})
         if request.method == "GET":
@@ -91,13 +100,23 @@ class MockupServer:
                 return self._reply(200, load(self.mockup, parts.path))
             except (FileNotFoundError, ValueError):
                 pass
-        return self._reply(404, {"error": {"code": "Base.1.0.ResourceMissingAtURI",
-                                           "message": f"{parts.path} not found"}})
+        return self._reply(
+            404,
+            {
+                "error": {
+                    "code": "Base.1.0.ResourceMissingAtURI",
+                    "message": f"{parts.path} not found",
+                }
+            },
+        )
 
     def calls(self, method: str | None = None, path: str | None = None):
-        return [c for c in self.rsps.calls
-                if (method is None or c.request.method == method)
-                and (path is None or urlsplit(c.request.url).path == path)]
+        return [
+            c
+            for c in self.rsps.calls
+            if (method is None or c.request.method == method)
+            and (path is None or urlsplit(c.request.url).path == path)
+        ]
 
 
 @pytest.fixture

tests/parsers/test_accounts.py

@@ -34,14 +34,39 @@ def test_account_defaults():
 
 
 def test_to_napalm_users(rackmount):
-    accounts = [parse_account(rackmount(f"/redfish/v1/AccountService/Accounts/{i}"))
-                for i in (1, 2)]
-    accounts.append({"id": "3", "username": "netork", "role_id": "ReadOnly", "enabled": True,
-                     "locked": False, "account_types": []})
-    accounts.append({"id": "4", "username": "ops", "role_id": "Operator", "enabled": False,
-                     "locked": False, "account_types": []})
-    accounts.append({"id": "5", "username": "custom", "role_id": "OemSpecial", "enabled": True,
-                     "locked": False, "account_types": []})
+    accounts = [
+        parse_account(rackmount(f"/redfish/v1/AccountService/Accounts/{i}")) for i in (1, 2)
+    ]
+    accounts.append(
+        {
+            "id": "3",
+            "username": "netork",
+            "role_id": "ReadOnly",
+            "enabled": True,
+            "locked": False,
+            "account_types": [],
+        }
+    )
+    accounts.append(
+        {
+            "id": "4",
+            "username": "ops",
+            "role_id": "Operator",
+            "enabled": False,
+            "locked": False,
+            "account_types": [],
+        }
+    )
+    accounts.append(
+        {
+            "id": "5",
+            "username": "custom",
+            "role_id": "OemSpecial",
+            "enabled": True,
+            "locked": False,
+            "account_types": [],
+        }
+    )
     accounts.append(None)
     assert to_napalm_users(accounts) == {
         "Administrator": {"level": 15, "password": "", "sshkeys": []},

tests/parsers/test_environment.py

@@ -39,10 +39,12 @@ def test_sources_legacy_only(localstorage):
 
 
 def test_sources_mixed_and_missing():
-    src = environment_sources({
-        "ThermalSubsystem": {"@odata.id": "/redfish/v1/Chassis/X/ThermalSubsystem"},
-        "Power": {"@odata.id": "/redfish/v1/Chassis/X/Power"},
-    })
+    src = environment_sources(
+        {
+            "ThermalSubsystem": {"@odata.id": "/redfish/v1/Chassis/X/ThermalSubsystem"},
+            "Power": {"@odata.id": "/redfish/v1/Chassis/X/Power"},
+        }
+    )
     assert src["thermal"]["schema"] == "ThermalSubsystem"
     assert src["power"]["schema"] == "Power"
     assert environment_sources({}) == {"thermal": None, "power": None, "sensors": None}
@@ -68,42 +70,70 @@ def test_thermal(rackmount):
     "reading, health, alert, critical",
     [
         (41, "OK", False, False),
-        (42, "OK", True, False),     # at UpperThresholdNonCritical
-        (45, "OK", True, True),      # at UpperThresholdCritical
-        (49, "OK", True, True),      # above UpperThresholdFatal
-        (4, "OK", True, True),       # below LowerThresholdCritical
-        (9, "OK", True, False),      # below LowerThresholdNonCritical
+        (42, "OK", True, False),  # at UpperThresholdNonCritical
+        (45, "OK", True, True),  # at UpperThresholdCritical
+        (49, "OK", True, True),  # above UpperThresholdFatal
+        (4, "OK", True, True),  # below LowerThresholdCritical
+        (9, "OK", True, False),  # below LowerThresholdNonCritical
         (30, "Warning", True, False),
         (30, "Critical", True, True),
     ],
 )
 def test_thermal_thresholds(reading, health, alert, critical):
-    env = parse_thermal({"Temperatures": [{
-        "Name": "T", "ReadingCelsius": reading, "Status": {"State": "Enabled", "Health": health},
-        "UpperThresholdNonCritical": 42, "UpperThresholdCritical": 45, "UpperThresholdFatal": 48,
-        "LowerThresholdNonCritical": 10, "LowerThresholdCritical": 5,
-    }]})
+    env = parse_thermal(
+        {
+            "Temperatures": [
+                {
+                    "Name": "T",
+                    "ReadingCelsius": reading,
+                    "Status": {"State": "Enabled", "Health": health},
+                    "UpperThresholdNonCritical": 42,
+                    "UpperThresholdCritical": 45,
+                    "UpperThresholdFatal": 48,
+                    "LowerThresholdNonCritical": 10,
+                    "LowerThresholdCritical": 5,
+                }
+            ]
+        }
+    )
     assert env["temperature"]["T"] == {
-        "temperature": float(reading), "is_alert": alert, "is_critical": critical}
+        "temperature": float(reading),
+        "is_alert": alert,
+        "is_critical": critical,
+    }
 
 
 def test_thermal_old_schema_names_and_duplicates():
     # Thermal 1.0 uses FanName; Supermicro repeats names across members.
-    env = parse_thermal({
-        "Fans": [
-            {"MemberId": "0", "FanName": "FAN1", "Reading": 3000, "Status": {"State": "Enabled",
-                                                                                "Health": "OK"}},
-            {"MemberId": "1", "FanName": "FAN2", "Status": {"State": "Absent"}},
-            {"MemberId": "2", "FanName": "FAN3", "Reading": 0,
-             "Status": {"State": "Enabled", "Health": "Critical"}},
-            {"MemberId": "3", "FanName": "FAN1", "Reading": 2900,
-             "Status": {"State": "Enabled", "Health": "OK"}},
-        ],
-        "Temperatures": [
-            {"MemberId": "0", "Name": "System Temp", "ReadingCelsius": 30},
-            {"MemberId": "1", "Name": "System Temp", "ReadingCelsius": 31},
-        ],
-    })
+    env = parse_thermal(
+        {
+            "Fans": [
+                {
+                    "MemberId": "0",
+                    "FanName": "FAN1",
+                    "Reading": 3000,
+                    "Status": {"State": "Enabled", "Health": "OK"},
+                },
+                {"MemberId": "1", "FanName": "FAN2", "Status": {"State": "Absent"}},
+                {
+                    "MemberId": "2",
+                    "FanName": "FAN3",
+                    "Reading": 0,
+                    "Status": {"State": "Enabled", "Health": "Critical"},
+                },
+                {
+                    "MemberId": "3",
+                    "FanName": "FAN1",
+                    "Reading": 2900,
+                    "Status": {"State": "Enabled", "Health": "OK"},
+                },
+            ],
+            "Temperatures": [
+                {"MemberId": "0", "Name": "System Temp", "ReadingCelsius": 30},
+                {"MemberId": "1", "Name": "System Temp", "ReadingCelsius": 31},
+            ],
+        }
+    )
     assert env["fans"] == {
         "FAN1": {"status": True},
         "FAN3": {"status": False},
@@ -126,10 +156,14 @@ def test_power(rackmount):
 
 
 def test_power_unknown_values_are_minus_one():
-    env = parse_power({"PowerSupplies": [
-        {"MemberId": "0", "Name": "PS1", "Status": {"State": "Enabled", "Health": "OK"}},
-        {"MemberId": "1", "Name": "PS2", "Status": {"State": "Absent"}},
-    ]})
+    env = parse_power(
+        {
+            "PowerSupplies": [
+                {"MemberId": "0", "Name": "PS1", "Status": {"State": "Enabled", "Health": "OK"}},
+                {"MemberId": "1", "Name": "PS2", "Status": {"State": "Absent"}},
+            ]
+        }
+    )
     assert env["power"] == {"PS1": {"status": True, "capacity": -1.0, "output": -1.0}}
     assert env["consumed_watts"] is None
     assert env["voltages"] == {}
@@ -211,8 +245,8 @@ def test_sensor_temperature_by_units(rackmount):
     assert s["units"] == "Cel"
     assert s["physical_context"] == "CPU"
     assert s["health"] == "Warning"
-    assert s["is_alert"] is True       # 44 >= UpperCaution 42
-    assert s["is_critical"] is False   # 44 < UpperCritical 45
+    assert s["is_alert"] is True  # 44 >= UpperCaution 42
+    assert s["is_critical"] is False  # 44 < UpperCritical 45
 
 
 def test_sensor_power(rackmount):
@@ -234,28 +268,37 @@ def test_sensor_voltage_and_fan(rackmount):
 
 
 def test_sensor_without_reading():
-    s = parse_sensor({"Id": "X", "Name": "X", "ReadingUnits": "Cel",
-                      "Status": {"State": "Disabled"}})
+    s = parse_sensor(
+        {"Id": "X", "Name": "X", "ReadingUnits": "Cel", "Status": {"State": "Disabled"}}
+    )
     assert s["reading"] is None
     assert s["is_alert"] is False
 
 
 def test_thermal_metrics(rackmount):
     env = parse_thermal_metrics(rackmount(CHASSIS + "/ThermalSubsystem/ThermalMetrics"))
-    assert env == {"temperature": {
-        "Intake": {"temperature": 24.8, "is_alert": False, "is_critical": False},
-        "Exhaust": {"temperature": 40.5, "is_alert": False, "is_critical": False},
-    }}
+    assert env == {
+        "temperature": {
+            "Intake": {"temperature": 24.8, "is_alert": False, "is_critical": False},
+            "Exhaust": {"temperature": 40.5, "is_alert": False, "is_critical": False},
+        }
+    }
 
 
 def test_build_environment_from_new_resources(rackmount):
-    fans = [parse_fan(rackmount(f"{CHASSIS}/ThermalSubsystem/Fans/{f}"))
-            for f in ("Bay1", "Bay2", "CPU1", "CPU2")]
+    fans = [
+        parse_fan(rackmount(f"{CHASSIS}/ThermalSubsystem/Fans/{f}"))
+        for f in ("Bay1", "Bay2", "CPU1", "CPU2")
+    ]
     bay1 = f"{CHASSIS}/PowerSubsystem/PowerSupplies/Bay1"
-    psus = [parse_power_supply(rackmount(bay1), rackmount(bay1 + "/Metrics")),
-            parse_power_supply(rackmount(f"{CHASSIS}/PowerSubsystem/PowerSupplies/Bay2"))]
-    sensors = [parse_sensor(rackmount(f"{CHASSIS}/Sensors/{s}"))
-               for s in ("CPU1Temp", "IntakeTemp", "CPUFan1", "TotalPower", "PS1InputVoltage")]
+    psus = [
+        parse_power_supply(rackmount(bay1), rackmount(bay1 + "/Metrics")),
+        parse_power_supply(rackmount(f"{CHASSIS}/PowerSubsystem/PowerSupplies/Bay2")),
+    ]
+    sensors = [
+        parse_sensor(rackmount(f"{CHASSIS}/Sensors/{s}"))
+        for s in ("CPU1Temp", "IntakeTemp", "CPUFan1", "TotalPower", "PS1InputVoltage")
+    ]
     env = build_environment(fans=fans, power_supplies=psus, sensors=sensors)
     assert env["fans"] == {
         "Fan Bay 1": {"status": True},
@@ -269,18 +312,29 @@ def test_build_environment_from_new_resources(rackmount):
     assert env["temperature"] == {
         "CPU #1 Temperature": {"temperature": 44.0, "is_alert": True, "is_critical": False},
         "Front Panel Intake Temperature": {
-            "temperature": 24.8, "is_alert": False, "is_critical": False},
+            "temperature": 24.8,
+            "is_alert": False,
+            "is_critical": False,
+        },
     }
     assert env["consumed_watts"] == 374.0
     assert env["voltages"] == {
         "Power Supply #1 Input Voltage": {
-            "voltage": 119.27, "is_alert": False, "is_critical": False},
+            "voltage": 119.27,
+            "is_alert": False,
+            "is_critical": False,
+        },
     }
 
 
 def test_build_environment_empty():
     assert build_environment() == {
-        "fans": {}, "temperature": {}, "power": {}, "voltages": {}, "consumed_watts": None}
+        "fans": {},
+        "temperature": {},
+        "power": {},
+        "voltages": {},
+        "consumed_watts": None,
+    }
 
 
 def test_power_supply_legacy_item(rackmount):

tests/parsers/test_inventory.py

@@ -83,12 +83,20 @@ def test_memory_absent(rackmount):
 
 def test_memory_older_schema():
     # Memory 1.0-1.x: DeviceLocator instead of Location, speed in OperatingSpeedMhz.
-    dimm = parse_memory({
-        "Id": "1", "Name": "Memory 1", "DeviceLocator": "P1-DIMMA1", "CapacityMiB": 16384,
-        "OperatingSpeedMhz": 2133, "Manufacturer": "Samsung ", "PartNumber": "M393A2G40DB0-CPB",
-        "SerialNumber": "12345678", "MemoryDeviceType": "DDR4",
-        "Status": {"State": "Enabled", "Health": "OK"},
-    })
+    dimm = parse_memory(
+        {
+            "Id": "1",
+            "Name": "Memory 1",
+            "DeviceLocator": "P1-DIMMA1",
+            "CapacityMiB": 16384,
+            "OperatingSpeedMhz": 2133,
+            "Manufacturer": "Samsung ",
+            "PartNumber": "M393A2G40DB0-CPB",
+            "SerialNumber": "12345678",
+            "MemoryDeviceType": "DDR4",
+            "Status": {"State": "Enabled", "Health": "OK"},
+        }
+    )
     assert dimm["location"] == "P1-DIMMA1"
     assert dimm["speed_mhz"] == 2133
     assert dimm["manufacturer"] == "Samsung"
@@ -110,16 +118,18 @@ def test_storage(localstorage):
     assert st["name"] == "Local Storage Controller"
     assert st["state"] == "Enabled"
     assert st["health"] == "OK"
-    assert st["controllers"] == [{
-        "name": "Contoso Integrated RAID",
-        "manufacturer": "Contoso",
-        "model": "12Gbs Integrated RAID",
-        "serial_number": "2M220100SL",
-        "firmware_version": "1.0.0.7",
-        "speed_gbps": 12,
-        "state": "Enabled",
-        "health": "OK",
-    }]
+    assert st["controllers"] == [
+        {
+            "name": "Contoso Integrated RAID",
+            "manufacturer": "Contoso",
+            "model": "12Gbs Integrated RAID",
+            "serial_number": "2M220100SL",
+            "firmware_version": "1.0.0.7",
+            "speed_gbps": 12,
+            "state": "Enabled",
+            "health": "OK",
+        }
+    ]
     assert st["drive_paths"] == [
         "/redfish/v1/Chassis/1U/Drives/35D38F11ACEF7BD3",
         "/redfish/v1/Chassis/1U/Drives/3F5A8C54207B7233",
@@ -163,7 +173,11 @@ def test_simple_storage(rackmount):
     assert st["name"] == "Simple Storage Controller"
     assert st["health"] == "OK"
     assert [d["name"] for d in st["drives"]] == [
-        "SATA Bay 1", "SATA Bay 2", "SATA Bay 3", "SATA Bay 4"]
+        "SATA Bay 1",
+        "SATA Bay 2",
+        "SATA Bay 3",
+        "SATA Bay 4",
+    ]
     bay1, bay2, bay3, _ = st["drives"]
     assert bay1["model"] == "3000GT8"
     assert bay1["capacity_bytes"] == 8000000000000

tests/parsers/test_manager.py

@@ -88,29 +88,33 @@ def test_ethernet_link_state(props, is_up, is_enabled):
 
 def test_ethernet_unset_addresses_and_formats():
     # Supermicro and others report unset addresses as 0.0.0.0 or "::" and MACs in lower case.
-    eth = parse_ethernet_interface({
-        "Id": "1",
-        "MACAddress": "0c:c4:7a:aa:bb:cc",
-        "SpeedMbps": None,
-        "IPv4Addresses": [
-            {"Address": "0.0.0.0", "SubnetMask": "0.0.0.0"},
-            {"Address": "10.1.2.3", "SubnetMask": "255.255.255.0"},
-            {"Address": "10.1.2.4", "SubnetMask": "garbage"},
-            {"Address": None},
-        ],
-        "IPv6Addresses": [
-            {"Address": "::", "PrefixLength": 0},
-            {"Address": "2001:DB8::5", "PrefixLength": 64},
-            {"Address": "2001:db8::6"},
-        ],
-    })
+    eth = parse_ethernet_interface(
+        {
+            "Id": "1",
+            "MACAddress": "0c:c4:7a:aa:bb:cc",
+            "SpeedMbps": None,
+            "IPv4Addresses": [
+                {"Address": "0.0.0.0", "SubnetMask": "0.0.0.0"},
+                {"Address": "10.1.2.3", "SubnetMask": "255.255.255.0"},
+                {"Address": "10.1.2.4", "SubnetMask": "garbage"},
+                {"Address": None},
+            ],
+            "IPv6Addresses": [
+                {"Address": "::", "PrefixLength": 0},
+                {"Address": "2001:DB8::5", "PrefixLength": 64},
+                {"Address": "2001:db8::6"},
+            ],
+        }
+    )
     assert eth["interface"]["mac_address"] == "0C:C4:7A:AA:BB:CC"
     assert eth["interface"]["speed"] == -1.0
     assert eth["interface"]["mtu"] == 0
     assert eth["interface"]["description"] == ""
     assert eth["ipv4"] == {"10.1.2.3": {"prefix_length": 24}}
-    assert eth["ipv6"] == {"2001:db8::5": {"prefix_length": 64},
-                           "2001:db8::6": {"prefix_length": 128}}
+    assert eth["ipv6"] == {
+        "2001:db8::5": {"prefix_length": 64},
+        "2001:db8::6": {"prefix_length": 128},
+    }
 
 
 # --- NetworkProtocol -----------------------------------------------------------------------
@@ -133,7 +137,11 @@ def test_network_protocol(rackmount):
     }
     assert np["ntp"] == {"enabled": None, "servers": []}
     assert np["snmp"] == {
-        "enabled": True, "port": 161, "v1": True, "v2c": False, "v3": True,
+        "enabled": True,
+        "port": 161,
+        "v1": True,
+        "v2c": False,
+        "v3": True,
         "communities": [
             {"name": "Read-only community", "access": "Limited"},
             {"name": "Full access community", "access": "Full"},
@@ -149,8 +157,14 @@ def test_network_protocol_ntp(telemetry):
     }
     assert np["protocols"]["NTP"] == {"enabled": True, "port": None}
     # Older SNMP schema: only ProtocolEnabled and Port.
-    assert np["snmp"] == {"enabled": True, "port": 161, "v1": None, "v2c": None, "v3": None,
-                          "communities": []}
+    assert np["snmp"] == {
+        "enabled": True,
+        "port": 161,
+        "v1": None,
+        "v2c": None,
+        "v3": None,
+        "communities": [],
+    }
 
 
 def test_network_protocol_port_without_flag():
@@ -162,5 +176,11 @@ def test_network_protocol_ntp_blank_entries():
     servers = ["", None, " pool.ntp.org ", "pool.ntp.org"]
     np = parse_network_protocol({"NTP": {"ProtocolEnabled": True, "NTPServers": servers}})
     assert np["ntp"]["servers"] == ["pool.ntp.org"]
-    assert np["snmp"] == {"enabled": None, "port": None, "v1": None, "v2c": None, "v3": None,
-                          "communities": []}
+    assert np["snmp"] == {
+        "enabled": None,
+        "port": None,
+        "v1": None,
+        "v2c": None,
+        "v3": None,
+        "communities": [],
+    }

tests/parsers/test_service_root.py

@@ -26,13 +26,15 @@ def test_service_root_links(rackmount):
 
 
 def test_service_root_vendor_and_oem():
-    root = parse_service_root({
-        "RedfishVersion": "1.0.1",
-        "Vendor": "Supermicro",
-        "Product": "X10DRL-i",
-        "Oem": {"Supermicro": {"@odata.type": "#SmcServiceRoot"}},
-        "Systems": {"@odata.id": "/redfish/v1/Systems"},
-    })
+    root = parse_service_root(
+        {
+            "RedfishVersion": "1.0.1",
+            "Vendor": "Supermicro",
+            "Product": "X10DRL-i",
+            "Oem": {"Supermicro": {"@odata.type": "#SmcServiceRoot"}},
+            "Systems": {"@odata.id": "/redfish/v1/Systems"},
+        }
+    )
     assert root["vendor"] == "Supermicro"
     assert root["product"] == "X10DRL-i"
     assert root["oem_vendors"] == ["Supermicro"]

tests/parsers/test_system.py

@@ -47,16 +47,18 @@ def test_system_links(rackmount):
 
 def test_system_placeholders_and_blanks_are_none():
     # Supermicro boards report the DMI defaults the board vendor never filled in.
-    s = parse_system({
-        "Id": "1",
-        "Manufacturer": "Supermicro",
-        "Model": "X10DRL-i",
-        "SerialNumber": "0123456789",
-        "SKU": "To Be Filled By O.E.M.",
-        "HostName": "",
-        "UUID": "  ",
-        "PartNumber": "Default string",
-    })
+    s = parse_system(
+        {
+            "Id": "1",
+            "Manufacturer": "Supermicro",
+            "Model": "X10DRL-i",
+            "SerialNumber": "0123456789",
+            "SKU": "To Be Filled By O.E.M.",
+            "HostName": "",
+            "UUID": "  ",
+            "PartNumber": "Default string",
+        }
+    )
     assert s["manufacturer"] == "Supermicro"
     assert s["model"] == "X10DRL-i"
     assert s["serial_number"] is None
@@ -88,8 +90,15 @@ def test_reset_action_inline_allowable_values(rackmount):
     reset = parse_reset_action(rackmount(SYSTEM))
     assert reset["target"] == SYSTEM + "/Actions/ComputerSystem.Reset"
     assert reset["allowed"] == [
-        "On", "ForceOff", "GracefulShutdown", "GracefulRestart", "ForceRestart",
-        "Nmi", "ForceOn", "PushPowerButton", "FullPowerCycle",
+        "On",
+        "ForceOff",
+        "GracefulShutdown",
+        "GracefulRestart",
+        "ForceRestart",
+        "Nmi",
+        "ForceOn",
+        "PushPowerButton",
+        "FullPowerCycle",
     ]
     assert reset["action_info"] is None
 
@@ -104,13 +113,25 @@ def test_reset_action_with_action_info(telemetry):
 
 def test_action_info(telemetry):
     allowed = parse_action_info(telemetry("/redfish/v1/Systems/1/ResetActionInfo"))
-    assert allowed == ["On", "ForceOff", "GracefulShutdown", "GracefulRestart",
-                       "ForceRestart", "Nmi", "ForceOn", "PushPowerButton"]
+    assert allowed == [
+        "On",
+        "ForceOff",
+        "GracefulShutdown",
+        "GracefulRestart",
+        "ForceRestart",
+        "Nmi",
+        "ForceOn",
+        "PushPowerButton",
+    ]
 
 
 def test_action_info_other_parameter():
-    info = {"Parameters": [{"Name": "ResetType", "AllowableValues": ["On"]},
-                           {"Name": "Delay", "AllowableValues": ["0", "5"]}]}
+    info = {
+        "Parameters": [
+            {"Name": "ResetType", "AllowableValues": ["On"]},
+            {"Name": "Delay", "AllowableValues": ["0", "5"]},
+        ]
+    }
     assert parse_action_info(info, "Delay") == ["0", "5"]
     assert parse_action_info(info, "Missing") is None
     assert parse_action_info({}) is None

tests/quality_baseline.json

@@ -0,0 +1,25 @@
+{
+ "coverage_floor": 94.0,
+ "function_complexity": {
+  "napalm_redfish/driver.py::RedfishDriver._accepted_actions": 14,
+  "napalm_redfish/driver.py::RedfishDriver.get_environment": 14,
+  "napalm_redfish/driver.py::RedfishDriver.get_facts": 15,
+  "napalm_redfish/parsers/environment.py::build_environment": 13,
+  "napalm_redfish/parsers/manager.py::parse_network_protocol": 14,
+  "napalm_redfish/transport.py::RedfishTransport._request": 14,
+  "napalm_redfish/transport.py::RedfishTransport.members": 15
+ },
+ "function_lines": {},
+ "mypy_relaxed": [
+  "napalm_redfish.bmc",
+  "napalm_redfish.driver",
+  "napalm_redfish.ipmi",
+  "napalm_redfish.parsers.common",
+  "napalm_redfish.parsers.environment",
+  "napalm_redfish.parsers.manager",
+  "napalm_redfish.parsers.system",
+  "napalm_redfish.transport"
+ ],
+ "python_file_lines": {},
+ "untyped_public": []
+}

tests/test_bmc_driver.py

@@ -35,8 +35,9 @@ def fake_pyghmi(monkeypatch):
 def x10(http):
     server = MockupServer(http, MOCKUP)
     for path, status in json.loads((FIXTURES / MOCKUP / "STATUS.json").read_text()).items():
-        body = json.loads((FIXTURES / MOCKUP / path[len("/redfish/v1/"):] / "index.json")
-                          .read_text())
+        body = json.loads(
+            (FIXTURES / MOCKUP / path[len("/redfish/v1/") :] / "index.json").read_text()
+        )
         server.override("GET", path, status, body)
     return server
 
@@ -49,8 +50,9 @@ def _open(**optional):
 
 @pytest.fixture
 def unlicensed(x10):
-    x10.override("GET", "/redfish/v1/", 403,
-                 json.loads((X10 / "service_root_403.json").read_text()))
+    x10.override(
+        "GET", "/redfish/v1/", 403, json.loads((X10 / "service_root_403.json").read_text())
+    )
     d = _open()
     yield d
     d.close()
@@ -84,7 +86,11 @@ def test_redfish_is_used_and_ipmi_is_not_even_opened(x10):
         assert d.get_facts()["model"] == "X10DRL-i"
         assert len(d.get_hardware_inventory()["cpus"]) == 2
         assert d.get_power_state()["actions"] == [
-            "GracefulShutdown", "ForceOff", "GracefulRestart", "ForceRestart"]
+            "GracefulShutdown",
+            "ForceOff",
+            "GracefulRestart",
+            "ForceRestart",
+        ]
     finally:
         d.close()
     assert FakeCommand.instances == []
@@ -117,10 +123,12 @@ def test_the_ipmi_port_is_its_own(x10):
 def test_the_partial_license_warning_passes_through(x10):
     d = _open()
     try:
-        assert d.get_device_warnings() == [{
-            "code": "bmc_license_partial",
-            "meta": {"licenses": "DCMS", "parts": ["drives", "firmware inventory"]},
-        }]
+        assert d.get_device_warnings() == [
+            {
+                "code": "bmc_license_partial",
+                "meta": {"licenses": "DCMS", "parts": ["drives", "firmware inventory"]},
+            }
+        ]
     finally:
         d.close()
 
@@ -134,7 +142,9 @@ def test_an_unlicensed_bmc_is_read_over_ipmi(unlicensed):
     assert facts["model"] == "X10DRL-i"
     assert facts["os_version"] == "3.91"
     assert unlicensed.get_power_state() == {
-        "state": "On", "actions": ["GracefulShutdown", "ForceOff", "ForceRestart"]}
+        "state": "On",
+        "actions": ["GracefulShutdown", "ForceOff", "ForceRestart"],
+    }
 
 
 def test_an_unlicensed_bmc_switches_power_over_ipmi(unlicensed):

tests/test_discovery.py

@@ -53,7 +53,7 @@ def test_idrac_and_ilo_landing_pages_are_recognised():
 
 
 def test_a_switch_page_scores_nothing():
-    page = '<html><head><title>Loading...</title></head><body>Web Switch</body></html>'
+    page = "<html><head><title>Loading...</title></head><body>Web Switch</body></html>"
     assert _score(page) == 0.0
 
 

tests/test_driver.py

@@ -34,10 +34,22 @@ def ts(iso: str) -> float:
 @pytest.fixture
 def driver(server):
     # The mockup's system log advertises a nextLink page it does not contain.
-    server.override("GET", SYSTEM_LOG + "?$skiptoken=2", 200, {
-        "Members": [{"@odata.id": SYSTEM_LOG + "/3", "Id": "3", "Severity": "Warning",
-                     "Created": "2012-03-07T14:46:00Z", "Message": "Fan redundancy lost"}],
-    })
+    server.override(
+        "GET",
+        SYSTEM_LOG + "?$skiptoken=2",
+        200,
+        {
+            "Members": [
+                {
+                    "@odata.id": SYSTEM_LOG + "/3",
+                    "Id": "3",
+                    "Severity": "Warning",
+                    "Created": "2012-03-07T14:46:00Z",
+                    "Message": "Fan redundancy lost",
+                }
+            ],
+        },
+    )
     d = RedfishDriver("bmc.example.net", "netork", "s3cret", optional_args={"verify_ssl": False})
     d.open()
     yield d
@@ -59,9 +71,15 @@ def test_role():
     assert primary_role_of(RedfishDriver) == "management_controller"
     assert RedfishDriver.USES_SSH is False
     assert RedfishDriver.DRIVER_NAME == "redfish"
-    for method in ("get_power_state", "set_power_state", "get_hardware_inventory",
-                   "get_event_log", "get_firmware_inventory", "get_management_protocols",
-                   "reboot_host"):
+    for method in (
+        "get_power_state",
+        "set_power_state",
+        "get_hardware_inventory",
+        "get_event_log",
+        "get_firmware_inventory",
+        "get_management_protocols",
+        "reboot_host",
+    ):
         assert hasattr(RedfishDriver, method), method
 
 
@@ -92,8 +110,13 @@ def test_open_failure_is_connection_exception(server):
 
 
 def test_optional_args(server):
-    d = RedfishDriver("bmc.example.net", "u", "p", timeout=12,
-                      optional_args={"port": 8443, "verify_ssl": False, "auth": "basic"})
+    d = RedfishDriver(
+        "bmc.example.net",
+        "u",
+        "p",
+        timeout=12,
+        optional_args={"port": 8443, "verify_ssl": False, "auth": "basic"},
+    )
     assert d.transport.base_url == "https://bmc.example.net:8443"
     assert d.transport.verify is False
     assert d.transport.auth == "basic"
@@ -141,10 +164,14 @@ def test_get_interfaces(driver):
 
 def test_get_interfaces_ip(driver):
     assert driver.get_interfaces_ip() == {
-        "eth0": {"ipv4": {"192.168.0.10": {"prefix_length": 22}},
-                 "ipv6": {"fe80::1ec1:deff:fe6f:1e24": {"prefix_length": 64}}},
-        "ToHost": {"ipv4": {"192.168.20.55": {"prefix_length": 24}},
-                   "ipv6": {"fe80::1ec1:deff:fe6f:1e24": {"prefix_length": 64}}},
+        "eth0": {
+            "ipv4": {"192.168.0.10": {"prefix_length": 22}},
+            "ipv6": {"fe80::1ec1:deff:fe6f:1e24": {"prefix_length": 64}},
+        },
+        "ToHost": {
+            "ipv4": {"192.168.20.55": {"prefix_length": 24}},
+            "ipv6": {"fe80::1ec1:deff:fe6f:1e24": {"prefix_length": 64}},
+        },
     }
 
 
@@ -164,23 +191,33 @@ def test_get_environment_new_resources(driver, server):
     }
     assert len(env["temperature"]) == 8
     assert env["temperature"]["CPU #1 Temperature"] == {
-        "temperature": 44.0, "is_alert": True, "is_critical": False}
+        "temperature": 44.0,
+        "is_alert": True,
+        "is_critical": False,
+    }
     # The legacy resources were not read.
     assert not server.calls("GET", CHASSIS + "/Thermal")
     assert not server.calls("GET", CHASSIS + "/Power")
 
 
 def test_get_environment_legacy(driver, server):
-    server.override("GET", CHASSIS, 200,
-                    without(CHASSIS, "ThermalSubsystem", "PowerSubsystem", "Sensors"))
+    server.override(
+        "GET", CHASSIS, 200, without(CHASSIS, "ThermalSubsystem", "PowerSubsystem", "Sensors")
+    )
     env = driver.get_environment()
     assert_shape(env, napalm_models.EnvironmentDict)
-    assert env["fans"] == {"BaseBoard System Fan": {"status": True},
-                           "BaseBoard System Fan Backup": {"status": True}}
+    assert env["fans"] == {
+        "BaseBoard System Fan": {"status": True},
+        "BaseBoard System Fan Backup": {"status": True},
+    }
     assert env["temperature"]["CPU1 Temp"] == {
-        "temperature": 41.0, "is_alert": False, "is_critical": False}
-    assert env["power"] == {"Power Supply Bay": {"status": False, "capacity": 800.0,
-                                                 "output": 325.0}}
+        "temperature": 41.0,
+        "is_alert": False,
+        "is_critical": False,
+    }
+    assert env["power"] == {
+        "Power Supply Bay": {"status": False, "capacity": 800.0, "output": 325.0}
+    }
     assert not server.calls("GET", CHASSIS + "/Sensors")
 
 
@@ -202,10 +239,17 @@ def test_get_users(driver):
 
 def test_get_ntp_servers(driver, server):
     assert driver.get_ntp_servers() == {}
-    server.override("GET", MANAGER + "/NetworkProtocol", 200,
-                    load("dmtf-public-telemetry", "/redfish/v1/Managers/1/NetworkProtocol"))
+    server.override(
+        "GET",
+        MANAGER + "/NetworkProtocol",
+        200,
+        load("dmtf-public-telemetry", "/redfish/v1/Managers/1/NetworkProtocol"),
+    )
     assert driver.get_ntp_servers() == {
-        "0.pool.ntp.org": {}, "1.pool.ntp.org": {}, "time-a.nist.gov": {}}
+        "0.pool.ntp.org": {},
+        "1.pool.ntp.org": {},
+        "time-a.nist.gov": {},
+    }
 
 
 def test_get_snmp_information(driver):
@@ -234,9 +278,10 @@ def test_get_power_state(driver):
     state = driver.get_power_state()
     assert_shape(state, models.PowerStateDict)
     # On: "On" is not offered; Redfish-only values (Nmi, ForceOn, FullPowerCycle) are dropped.
-    assert state == {"state": "On",
-                     "actions": ["GracefulShutdown", "ForceOff", "GracefulRestart",
-                                 "ForceRestart"]}
+    assert state == {
+        "state": "On",
+        "actions": ["GracefulShutdown", "ForceOff", "GracefulRestart", "ForceRestart"],
+    }
 
 
 def test_get_power_state_off(driver, server):
@@ -252,8 +297,16 @@ def test_get_power_state_from_action_info(driver, server):
     del reset["ResetType@Redfish.AllowableValues"]
     reset["@Redfish.ActionInfo"] = SYSTEM + "/ResetActionInfo"
     server.override("GET", SYSTEM, 200, system)
-    server.override("GET", SYSTEM + "/ResetActionInfo", 200, {"Parameters": [
-        {"Name": "ResetType", "AllowableValues": ["On", "ForceOff", "PowerCycle"]}]})
+    server.override(
+        "GET",
+        SYSTEM + "/ResetActionInfo",
+        200,
+        {
+            "Parameters": [
+                {"Name": "ResetType", "AllowableValues": ["On", "ForceOff", "PowerCycle"]}
+            ]
+        },
+    )
     assert driver.get_power_state()["actions"] == ["ForceOff", "PowerCycle"]
 
 
@@ -263,7 +316,12 @@ def test_get_power_state_unlisted_values(driver, server):
     del system["Actions"]["#ComputerSystem.Reset"]["ResetType@Redfish.AllowableValues"]
     server.override("GET", SYSTEM, 200, system)
     assert driver.get_power_state()["actions"] == [
-        "GracefulShutdown", "ForceOff", "GracefulRestart", "ForceRestart", "PowerCycle"]
+        "GracefulShutdown",
+        "ForceOff",
+        "GracefulRestart",
+        "ForceRestart",
+        "PowerCycle",
+    ]
 
 
 def test_get_power_state_without_reset_action(driver, server):
@@ -351,42 +409,73 @@ def test_get_hardware_inventory(driver):
         assert_shape(cpu, models.BmcCpuDict)
     # The FPGA is a processor but not a CPU.
     assert inv["cpus"] == [
-        {"socket": "CPU 1", "model": "Multi-Core Intel(R) Xeon(R) processor 7xxx Series",
-         "cores": 8, "threads": 16, "status": "OK"},
+        {
+            "socket": "CPU 1",
+            "model": "Multi-Core Intel(R) Xeon(R) processor 7xxx Series",
+            "cores": 8,
+            "threads": 16,
+            "status": "OK",
+        },
         {"socket": "CPU 2", "model": "", "cores": None, "threads": None, "status": "Absent"},
     ]
 
     for dimm in inv["memory"]:
         assert_shape(dimm, models.BmcDimmDict)
     assert [d["slot"] for d in inv["memory"]] == ["DIMM 1", "DIMM 2", "DIMM 3", "DIMM 4"]
-    assert inv["memory"][0] == {"slot": "DIMM 1", "size_bytes": 32768 * 1024 * 1024,
-                                "type": "DDR4", "speed_mts": None, "part_number": "",
-                                "serial_number": "", "status": "OK"}
+    assert inv["memory"][0] == {
+        "slot": "DIMM 1",
+        "size_bytes": 32768 * 1024 * 1024,
+        "type": "DDR4",
+        "speed_mts": None,
+        "part_number": "",
+        "serial_number": "",
+        "status": "OK",
+    }
     assert inv["memory"][3]["status"] == "Absent"
 
     for drive in inv["drives"]:
         assert_shape(drive, models.BmcDriveDict)
     # This system has SimpleStorage only.
-    assert inv["drives"][0] == {"slot": "SATA Bay 1", "model": "3000GT8", "serial_number": "",
-                                "size_bytes": 8000000000000, "media": "", "protocol": "",
-                                "status": "OK"}
+    assert inv["drives"][0] == {
+        "slot": "SATA Bay 1",
+        "model": "3000GT8",
+        "serial_number": "",
+        "size_bytes": 8000000000000,
+        "media": "",
+        "protocol": "",
+        "status": "OK",
+    }
     assert [d["status"] for d in inv["drives"]] == ["OK", "Warning", "Absent", "Absent"]
 
     for psu in inv["power_supplies"]:
         assert_shape(psu, models.BmcPsuDict)
     assert inv["power_supplies"] == [
-        {"slot": "Power Supply Bay 1", "model": "RKS-440DC", "serial_number": "3488247",
-         "capacity_watts": 400, "status": "Warning"},
-        {"slot": "Power Supply Bay 2", "model": "", "serial_number": "",
-         "capacity_watts": None, "status": "Absent"},
+        {
+            "slot": "Power Supply Bay 1",
+            "model": "RKS-440DC",
+            "serial_number": "3488247",
+            "capacity_watts": 400,
+            "status": "Warning",
+        },
+        {
+            "slot": "Power Supply Bay 2",
+            "model": "",
+            "serial_number": "",
+            "capacity_watts": None,
+            "status": "Absent",
+        },
     ]
 
     for nic in inv["nics"]:
         assert_shape(nic, models.BmcNicDict)
     nics = {n["name"]: n for n in inv["nics"]}
     assert set(nics) == {"12446A3B0411", "12446A3B8890", "ToManager", "VLAN1"}
-    assert nics["12446A3B0411"] == {"name": "12446A3B0411", "mac_address": "12:44:6a:3b:04:11",
-                                    "speed_mbps": 1000, "link_up": True}
+    assert nics["12446A3B0411"] == {
+        "name": "12446A3B0411",
+        "mac_address": "12:44:6a:3b:04:11",
+        "speed_mbps": 1000,
+        "link_up": True,
+    }
     assert nics["ToManager"]["link_up"] is None
 
 
@@ -396,25 +485,38 @@ def test_get_hardware_inventory_storage_drives(driver, server):
     del system["SimpleStorage"]
     system["Storage"] = {"@odata.id": SYSTEM + "/Storage"}
     server.override("GET", SYSTEM, 200, system)
-    server.override("GET", SYSTEM + "/Storage", 200,
-                    load("dmtf-public-localstorage", SYSTEM + "/Storage"))
-    server.override("GET", SYSTEM + "/Storage/1", 200,
-                    load("dmtf-public-localstorage", SYSTEM + "/Storage/1"))
+    server.override(
+        "GET", SYSTEM + "/Storage", 200, load("dmtf-public-localstorage", SYSTEM + "/Storage")
+    )
+    server.override(
+        "GET", SYSTEM + "/Storage/1", 200, load("dmtf-public-localstorage", SYSTEM + "/Storage/1")
+    )
     for drive in ("35D38F11ACEF7BD3", "3F5A8C54207B7233", "32ADF365C6C1B7BD", "3D58ECBC375FD9F2"):
         path = f"{CHASSIS}/Drives/{drive}"
         server.override("GET", path, 200, load("dmtf-public-localstorage", path))
     drives = driver.get_hardware_inventory()["drives"]
     assert len(drives) == 4
-    assert drives[0] == {"slot": "Drive Sample", "model": "C123", "serial_number": "1234567",
-                         "size_bytes": 899527000064, "media": "HDD", "protocol": "SAS",
-                         "status": "OK"}
+    assert drives[0] == {
+        "slot": "Drive Sample",
+        "model": "C123",
+        "serial_number": "1234567",
+        "size_bytes": 899527000064,
+        "media": "HDD",
+        "protocol": "SAS",
+        "status": "OK",
+    }
 
 
 def test_get_hardware_inventory_legacy_power(driver, server):
     server.override("GET", CHASSIS, 200, without(CHASSIS, "PowerSubsystem"))
     assert driver.get_hardware_inventory()["power_supplies"] == [
-        {"slot": "Power Supply Bay", "model": "499253-B21", "serial_number": "1Z0000001",
-         "capacity_watts": 800, "status": "Warning"},
+        {
+            "slot": "Power Supply Bay",
+            "model": "499253-B21",
+            "serial_number": "1Z0000001",
+            "capacity_watts": 800,
+            "status": "Warning",
+        },
     ]
 
 
@@ -423,14 +525,30 @@ def test_get_event_log(driver):
     for entry in log:
         assert_shape(entry, models.EventLogEntryDict)
     assert log == [
-        {"id": SYSTEM_LOG + "/3", "timestamp": ts("2012-03-07T14:46:00Z"),
-         "severity": "Warning", "message": "Fan redundancy lost"},
-        {"id": SYSTEM_LOG + "/2", "timestamp": ts("2012-03-07T14:45:00Z"),
-         "severity": "Critical", "message": "Temperature threshold exceeded"},
-        {"id": SYSTEM_LOG + "/1", "timestamp": ts("2012-03-07T14:44:00Z"),
-         "severity": "Critical", "message": "Temperature threshold exceeded"},
-        {"id": MANAGER + "/LogServices/Log/Entries/1", "timestamp": ts("2012-03-07T14:44:00Z"),
-         "severity": "Critical", "message": "System May be Melting"},
+        {
+            "id": SYSTEM_LOG + "/3",
+            "timestamp": ts("2012-03-07T14:46:00Z"),
+            "severity": "Warning",
+            "message": "Fan redundancy lost",
+        },
+        {
+            "id": SYSTEM_LOG + "/2",
+            "timestamp": ts("2012-03-07T14:45:00Z"),
+            "severity": "Critical",
+            "message": "Temperature threshold exceeded",
+        },
+        {
+            "id": SYSTEM_LOG + "/1",
+            "timestamp": ts("2012-03-07T14:44:00Z"),
+            "severity": "Critical",
+            "message": "Temperature threshold exceeded",
+        },
+        {
+            "id": MANAGER + "/LogServices/Log/Entries/1",
+            "timestamp": ts("2012-03-07T14:44:00Z"),
+            "severity": "Critical",
+            "message": "System May be Melting",
+        },
     ]
 
 
@@ -441,11 +559,18 @@ def test_get_event_log_limit(driver):
 
 def test_get_event_log_timestamps(driver, server):
     entries = SYSTEM_LOG + "?$skiptoken=2"
-    server.override("GET", entries, 200, {"Members": [
-        {"Id": "3", "Created": "2026-10-13T08:00:00+02:00", "Message": "offset"},
-        {"Id": "4", "Message": "no time", "Severity": "Informational"},
-        {"Id": "5", "Created": "garbage", "Message": "bad time"},
-    ]})
+    server.override(
+        "GET",
+        entries,
+        200,
+        {
+            "Members": [
+                {"Id": "3", "Created": "2026-10-13T08:00:00+02:00", "Message": "offset"},
+                {"Id": "4", "Message": "no time", "Severity": "Informational"},
+                {"Id": "5", "Created": "garbage", "Message": "bad time"},
+            ]
+        },
+    )
     log = {e["message"]: e for e in driver.get_event_log()}
     assert log["offset"]["timestamp"] == ts("2026-10-13T06:00:00Z")
     assert log["no time"]["timestamp"] is None
@@ -482,8 +607,12 @@ def test_get_management_protocols(driver):
 
 
 def test_get_management_protocols_missing(driver, server):
-    server.override("GET", MANAGER + "/NetworkProtocol", 200,
-                    {"HTTPS": {"ProtocolEnabled": True, "Port": 443}, "IPMI": {"Port": 623}})
+    server.override(
+        "GET",
+        MANAGER + "/NetworkProtocol",
+        200,
+        {"HTTPS": {"ProtocolEnabled": True, "Port": 443}, "IPMI": {"Port": 623}},
+    )
     protocols = driver.get_management_protocols()
     assert protocols["https"] == {"enabled": True, "port": 443}
     assert protocols["ipmi"] == {"enabled": False, "port": 623}
@@ -493,8 +622,9 @@ def test_get_management_protocols_missing(driver, server):
 def _unlicensed(server):
     # A Supermicro X10 refuses Redfish entirely until a license is activated.
     x10 = Path(__file__).parent / "fixtures" / "supermicro-x10drl-i-3.91"
-    server.override("GET", "/redfish/v1/", 403,
-                    json.loads((x10 / "service_root_403.json").read_text()))
+    server.override(
+        "GET", "/redfish/v1/", 403, json.loads((x10 / "service_root_403.json").read_text())
+    )
     d = RedfishDriver("bmc.example.net", "netork", "s3cret", optional_args={"verify_ssl": False})
     d.open()
     return d
@@ -506,15 +636,24 @@ def test_unlicensed_bmc_still_opens(server):
     d = _unlicensed(server)
     assert d.is_alive() == {"is_alive": True}
     assert d.get_facts() == {
-        "uptime": -1.0, "vendor": "", "model": "", "hostname": "bmc.example.net",
-        "fqdn": "bmc.example.net", "os_version": "", "serial_number": "", "interface_list": [],
+        "uptime": -1.0,
+        "vendor": "",
+        "model": "",
+        "hostname": "bmc.example.net",
+        "fqdn": "bmc.example.net",
+        "os_version": "",
+        "serial_number": "",
+        "interface_list": [],
     }
 
 
 def test_unlicensed_bmc_names_every_license_it_asks_for_as_missing(server):
     licenses = _unlicensed(server).get_licenses()
     assert [(lic["name"], lic["status"]) for lic in licenses] == [
-        ("SUM", "missing"), ("DCMS", "missing"), ("OOB", "missing")]
+        ("SUM", "missing"),
+        ("DCMS", "missing"),
+        ("OOB", "missing"),
+    ]
     # Refused altogether: nothing is withheld in part.
     assert all("withholds" not in lic and "Not licensed" in lic["detail"] for lic in licenses)
 

tests/test_ipmi.py

@@ -105,8 +105,12 @@ def test_open_logs_in_on_the_ipmi_port_and_close_logs_out():
     d = IpmiDriver("192.0.2.150", "netork", "s3cret", optional_args={"port": 6230})
     d.open()
     (cmd,) = FakeCommand.instances
-    assert cmd.args == {"bmc": "192.0.2.150", "userid": "netork", "password": "s3cret",
-                        "port": 6230}
+    assert cmd.args == {
+        "bmc": "192.0.2.150",
+        "userid": "netork",
+        "password": "s3cret",
+        "port": 6230,
+    }
     assert d.is_alive() == {"is_alive": True}
     d.close()
     assert cmd.logged_out
@@ -167,8 +171,11 @@ def test_get_environment(driver):
     assert set(env) == {"fans", "temperature", "power", "cpu", "memory"}
     assert env["cpu"] == {} and env["memory"] == {}
     assert len(env["temperature"]) == 19
-    assert env["temperature"]["CPU1 Temp"] == {"temperature": 34.0, "is_alert": False,
-                                               "is_critical": False}
+    assert env["temperature"]["CPU1 Temp"] == {
+        "temperature": 34.0,
+        "is_alert": False,
+        "is_critical": False,
+    }
     assert env["fans"] == {name: {"status": True} for name in ("FAN2", "FAN3", "FAN4", "FANB")}
     # Only supplies that report themselves present.
     assert env["power"] == {"PS1 Status": {"status": True, "capacity": -1.0, "output": -1.0}}
@@ -176,17 +183,46 @@ def test_get_environment(driver):
 
 def test_sensor_health_flags(driver, monkeypatch):
     readings = [
-        {"name": "Hot", "type": "Temperature", "value": 91.0, "units": "°C", "health": 1,
-         "states": [], "unavailable": 0},
-        {"name": "Melting", "type": "Temperature", "value": 105.0, "units": "°C", "health": 2,
-         "states": [], "unavailable": 0},
-        {"name": "Gone", "type": "Temperature", "value": None, "units": "°C", "health": 0,
-         "states": [], "unavailable": 1},
-        {"name": "FAN1", "type": "Fan", "value": 0.0, "units": "RPM", "health": 2,
-         "states": [], "unavailable": 0},
+        {
+            "name": "Hot",
+            "type": "Temperature",
+            "value": 91.0,
+            "units": "°C",
+            "health": 1,
+            "states": [],
+            "unavailable": 0,
+        },
+        {
+            "name": "Melting",
+            "type": "Temperature",
+            "value": 105.0,
+            "units": "°C",
+            "health": 2,
+            "states": [],
+            "unavailable": 0,
+        },
+        {
+            "name": "Gone",
+            "type": "Temperature",
+            "value": None,
+            "units": "°C",
+            "health": 0,
+            "states": [],
+            "unavailable": 1,
+        },
+        {
+            "name": "FAN1",
+            "type": "Fan",
+            "value": 0.0,
+            "units": "RPM",
+            "health": 2,
+            "states": [],
+            "unavailable": 0,
+        },
     ]
-    monkeypatch.setattr(FakeCommand, "get_sensor_data",
-                        lambda self: iter(SimpleNamespace(**r) for r in readings))
+    monkeypatch.setattr(
+        FakeCommand, "get_sensor_data", lambda self: iter(SimpleNamespace(**r) for r in readings)
+    )
     env = driver.get_environment()
     assert env["temperature"] == {
         "Hot": {"temperature": 91.0, "is_alert": True, "is_critical": False},
@@ -201,8 +237,7 @@ def test_sensor_health_flags(driver, monkeypatch):
 def test_get_power_state(driver):
     state = driver.get_power_state()
     assert_shape(state, models.PowerStateDict)
-    assert state == {"state": "On",
-                     "actions": ["GracefulShutdown", "ForceOff", "ForceRestart"]}
+    assert state == {"state": "On", "actions": ["GracefulShutdown", "ForceOff", "ForceRestart"]}
 
 
 def test_a_server_that_is_off_can_only_be_switched_on(driver):
@@ -216,12 +251,15 @@ def test_every_offered_action_is_a_role_power_action(driver):
         assert set(driver.get_power_state()["actions"]) <= set(models.POWER_ACTIONS)
 
 
-@pytest.mark.parametrize("action, powerstate, expected", [
-    ("GracefulShutdown", "on", "shutdown"),
-    ("ForceOff", "on", "off"),
-    ("ForceRestart", "on", "reset"),
-    ("On", "off", "on"),
-])
+@pytest.mark.parametrize(
+    "action, powerstate, expected",
+    [
+        ("GracefulShutdown", "on", "shutdown"),
+        ("ForceOff", "on", "off"),
+        ("ForceRestart", "on", "reset"),
+        ("On", "off", "on"),
+    ],
+)
 def test_set_power_state_sends_the_chassis_control(driver, action, powerstate, expected):
     cmd = FakeCommand.instances[0]
     cmd.powerstate = powerstate
@@ -236,12 +274,15 @@ def test_set_power_state_rejects_an_unknown_action(driver):
     assert FakeCommand.instances[0].power_requests == []
 
 
-@pytest.mark.parametrize("action, powerstate", [
-    ("On", "on"),
-    ("ForceOff", "off"),
-    ("GracefulRestart", "on"),  # IPMI has no graceful restart
-    ("PowerCycle", "on"),
-])
+@pytest.mark.parametrize(
+    "action, powerstate",
+    [
+        ("On", "on"),
+        ("ForceOff", "off"),
+        ("GracefulRestart", "on"),  # IPMI has no graceful restart
+        ("PowerCycle", "on"),
+    ],
+)
 def test_set_power_state_refuses_what_is_not_offered_now(driver, action, powerstate):
     cmd = FakeCommand.instances[0]
     cmd.powerstate = powerstate
@@ -268,8 +309,13 @@ def test_get_hardware_inventory(driver):
     assert inv["system_uuid"] == "00000000-534d-c40c-0000-000000000001"
     assert inv["power_supplies"] == [
         {"slot": "PS1", "model": "", "serial_number": "", "capacity_watts": None, "status": "OK"},
-        {"slot": "PS2", "model": "", "serial_number": "", "capacity_watts": None,
-         "status": "Absent"},
+        {
+            "slot": "PS2",
+            "model": "",
+            "serial_number": "",
+            "capacity_watts": None,
+            "status": "Absent",
+        },
     ]
     assert inv["cpus"] == [] and inv["memory"] == [] and inv["drives"] == [] and inv["nics"] == []
 
@@ -277,14 +323,24 @@ def test_get_hardware_inventory(driver):
 def test_get_event_log(driver):
     (entry,) = driver.get_event_log()
     assert_shape(entry, models.EventLogEntryDict)
-    assert entry == {"id": "1", "timestamp": 1790788051.0, "severity": "OK",
-                     "message": "PS2 Status: Recovered"}
+    assert entry == {
+        "id": "1",
+        "timestamp": 1790788051.0,
+        "severity": "OK",
+        "message": "PS2 Status: Recovered",
+    }
 
 
 def test_event_log_newest_first_and_limited(driver, monkeypatch):
     sel = [
-        {"record_id": i, "timecode": 1000 + i, "component": "FAN1", "event": "Lower critical",
-         "severity": 2, "deassertion": False}
+        {
+            "record_id": i,
+            "timecode": 1000 + i,
+            "component": "FAN1",
+            "event": "Lower critical",
+            "severity": 2,
+            "deassertion": False,
+        }
         for i in range(1, 6)
     ]
     monkeypatch.setattr(FakeCommand, "get_event_log", lambda self: iter(sel))

tests/test_transport.py

@@ -189,8 +189,13 @@ def test_login_url_from_service_root(server):
     custom = dict(_root(), Links={"Sessions": {"@odata.id": "/redfish/v1/Custom/Sessions"}})
     server.override("GET", "/redfish/v1/", 200, custom)
     server.override("GET", "/redfish/v1", 200, custom)
-    server.override("POST", "/redfish/v1/Custom/Sessions", 201, {},
-                    {"X-Auth-Token": TOKEN, "Location": SESSION_PATH})
+    server.override(
+        "POST",
+        "/redfish/v1/Custom/Sessions",
+        201,
+        {},
+        {"X-Auth-Token": TOKEN, "Location": SESSION_PATH},
+    )
     t = make()
     t.open()
     assert server.calls("POST", "/redfish/v1/Custom/Sessions")
@@ -314,9 +319,18 @@ def test_get_optional_returns_none_on_404(transport):
 
 
 def test_get_error_carries_redfish_message(transport, server):
-    server.override("GET", "/redfish/v1/Managers", 500, {
-        "error": {"code": "Base.1.0.InternalError", "message": "General error",
-                  "@Message.ExtendedInfo": [{"Message": "The BMC is busy."}]}})
+    server.override(
+        "GET",
+        "/redfish/v1/Managers",
+        500,
+        {
+            "error": {
+                "code": "Base.1.0.InternalError",
+                "message": "General error",
+                "@Message.ExtendedInfo": [{"Message": "The BMC is busy."}],
+            }
+        },
+    )
     with pytest.raises(RedfishHTTPError) as exc:
         transport.get("/redfish/v1/Managers")
     assert exc.value.status == 500
@@ -366,20 +380,30 @@ def test_members_uses_expanded_members_without_extra_requests(transport, server)
 
 def test_members_follows_next_link(transport, server):
     path = "/redfish/v1/Systems/437XR1138R2/LogServices/Log1/Entries"
-    server.override("GET", path + "?$skiptoken=2", 200, {
-        "Members": [{"@odata.id": path + "/3", "Id": "3", "Message": "third"}],
-        "@odata.id": path,
-    })
+    server.override(
+        "GET",
+        path + "?$skiptoken=2",
+        200,
+        {
+            "Members": [{"@odata.id": path + "/3", "Id": "3", "Message": "third"}],
+            "@odata.id": path,
+        },
+    )
     entries = transport.members(path)
     assert [e["Id"] for e in entries] == ["1", "2", "3"]
 
 
 def test_members_next_link_loop_terminates(transport, server):
     path = "/redfish/v1/Managers"
-    server.override("GET", path, 200, {
-        "Members": [{"@odata.id": "/redfish/v1/Managers/BMC"}],
-        "Members@odata.nextLink": path,
-    })
+    server.override(
+        "GET",
+        path,
+        200,
+        {
+            "Members": [{"@odata.id": "/redfish/v1/Managers/BMC"}],
+            "Members@odata.nextLink": path,
+        },
+    )
     assert [m["Id"] for m in transport.members(path)] == ["BMC"]
 
 

tests/test_x10_redfish.py

@@ -23,16 +23,18 @@ REFUSED = json.loads((FIXTURES / MOCKUP / "STATUS.json").read_text())
 def x10(http):
     server = MockupServer(http, MOCKUP)
     for path, status in REFUSED.items():
-        body = json.loads((FIXTURES / MOCKUP / path[len("/redfish/v1/"):] / "index.json")
-                          .read_text())
+        body = json.loads(
+            (FIXTURES / MOCKUP / path[len("/redfish/v1/") :] / "index.json").read_text()
+        )
         server.override("GET", path, status, body)
     return server
 
 
 @pytest.fixture
 def driver(x10):
-    d = RedfishDriver(BASE_URL.removeprefix("https://"), "netork", "secret",
-                      optional_args={"verify_ssl": False})
+    d = RedfishDriver(
+        BASE_URL.removeprefix("https://"), "netork", "secret", optional_args={"verify_ssl": False}
+    )
     d.open()
     yield d
     d.close()
@@ -65,10 +67,12 @@ def test_the_firmware_comes_from_the_manager_and_the_system_without_updateservic
 
 
 def test_the_missing_license_is_named_with_what_it_withholds(driver):
-    assert driver.get_device_warnings() == [{
-        "code": "bmc_license_partial",
-        "meta": {"licenses": "DCMS", "parts": ["drives", "firmware inventory"]},
-    }]
+    assert driver.get_device_warnings() == [
+        {
+            "code": "bmc_license_partial",
+            "meta": {"licenses": "DCMS", "parts": ["drives", "firmware inventory"]},
+        }
+    ]
 
 
 def test_the_power_actions_are_the_ones_the_bmc_advertises(driver):
@@ -95,8 +99,10 @@ def test_the_sensors_and_the_event_log_read(driver):
 
 
 def test_the_missing_license_is_a_fact_with_what_it_withholds(driver):
-    assert driver.get_licenses() == [{
-        "name": "DCMS",
-        "status": "missing",
-        "withholds": ["drives", "firmware inventory"],
-    }]
+    assert driver.get_licenses() == [
+        {
+            "name": "DCMS",
+            "status": "missing",
+            "withholds": ["drives", "firmware inventory"],
+        }
+    ]