diff --git a/README.md b/README.md
index 4fa4bd8..0a55dd4 100644
--- a/README.md
+++ b/README.md
@@ -87,10 +87,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()
```
diff --git a/docs/role.md b/docs/role.md
index 6b00122..b914ce4 100644
--- a/docs/role.md
+++ b/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:
diff --git a/napalm_redfish/bmc.py b/napalm_redfish/bmc.py
index bdb512a..6fd5d4f 100644
--- a/napalm_redfish/bmc.py
+++ b/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)
diff --git a/napalm_redfish/driver.py b/napalm_redfish/driver.py
index 22cb8b4..f0e3e19 100644
--- a/napalm_redfish/driver.py
+++ b/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."""
diff --git a/napalm_redfish/ipmi.py b/napalm_redfish/ipmi.py
index e965948..a7ae06e 100644
--- a/napalm_redfish/ipmi.py
+++ b/napalm_redfish/ipmi.py
@@ -85,8 +85,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
@@ -219,7 +218,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 +235,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 +260,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]:
diff --git a/napalm_redfish/parsers/accounts.py b/napalm_redfish/parsers/accounts.py
index 14508d5..68dd84b 100644
--- a/napalm_redfish/parsers/accounts.py
+++ b/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
diff --git a/napalm_redfish/parsers/common.py b/napalm_redfish/parsers/common.py
index 7e6a567..bb3f0a6 100644
--- a/napalm_redfish/parsers/common.py
+++ b/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:
diff --git a/napalm_redfish/parsers/environment.py b/napalm_redfish/parsers/environment.py
index 508c299..d68769f 100644
--- a/napalm_redfish/parsers/environment.py
+++ b/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
diff --git a/napalm_redfish/parsers/firmware.py b/napalm_redfish/parsers/firmware.py
index 1c008e1..87641bd 100644
--- a/napalm_redfish/parsers/firmware.py
+++ b/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
diff --git a/napalm_redfish/parsers/inventory.py b/napalm_redfish/parsers/inventory.py
index 6b7f4bb..adfff2a 100644
--- a/napalm_redfish/parsers/inventory.py
+++ b/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")),
diff --git a/napalm_redfish/parsers/logs.py b/napalm_redfish/parsers/logs.py
index dfe3ff9..e33fc54 100644
--- a/napalm_redfish/parsers/logs.py
+++ b/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
diff --git a/napalm_redfish/parsers/manager.py b/napalm_redfish/parsers/manager.py
index 9cdbca6..034ff10 100644
--- a/napalm_redfish/parsers/manager.py
+++ b/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")),
diff --git a/napalm_redfish/parsers/service_root.py b/napalm_redfish/parsers/service_root.py
index 2bf8f66..e87ba65 100644
--- a/napalm_redfish/parsers/service_root.py
+++ b/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
diff --git a/napalm_redfish/parsers/system.py b/napalm_redfish/parsers/system.py
index 753dbee..78739c4 100644
--- a/napalm_redfish/parsers/system.py
+++ b/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
diff --git a/napalm_redfish/transport.py b/napalm_redfish/transport.py
index bddae5f..119e464 100644
--- a/napalm_redfish/transport.py
+++ b/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:
diff --git a/pyproject.toml b/pyproject.toml
index 75f2ea4..b0509f3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -39,3 +39,14 @@ include = ["napalm_redfish*"]
[tool.pytest.ini_options]
testpaths = ["tests"]
+
+[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"]
diff --git a/tests/conftest.py b/tests/conftest.py
index 1ebd55c..35ac84b 100644
--- a/tests/conftest.py
+++ b/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
diff --git a/tests/parsers/test_accounts.py b/tests/parsers/test_accounts.py
index 8b9349c..932e5e8 100644
--- a/tests/parsers/test_accounts.py
+++ b/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": []},
diff --git a/tests/parsers/test_environment.py b/tests/parsers/test_environment.py
index 7aecaa5..a1b51a1 100644
--- a/tests/parsers/test_environment.py
+++ b/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):
diff --git a/tests/parsers/test_inventory.py b/tests/parsers/test_inventory.py
index 5ed9e72..b0ccea1 100644
--- a/tests/parsers/test_inventory.py
+++ b/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
diff --git a/tests/parsers/test_manager.py b/tests/parsers/test_manager.py
index d5e2e31..0c8a0cb 100644
--- a/tests/parsers/test_manager.py
+++ b/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": [],
+ }
diff --git a/tests/parsers/test_service_root.py b/tests/parsers/test_service_root.py
index 2be1313..95edf1e 100644
--- a/tests/parsers/test_service_root.py
+++ b/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"]
diff --git a/tests/parsers/test_system.py b/tests/parsers/test_system.py
index 4080f2c..3a520bc 100644
--- a/tests/parsers/test_system.py
+++ b/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
diff --git a/tests/test_bmc_driver.py b/tests/test_bmc_driver.py
index 1ecb44b..c06adb5 100644
--- a/tests/test_bmc_driver.py
+++ b/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):
diff --git a/tests/test_discovery.py b/tests/test_discovery.py
index 36386d7..bdbe179 100644
--- a/tests/test_discovery.py
+++ b/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 = 'Loading...Web Switch'
+ page = "Loading...Web Switch"
assert _score(page) == 0.0
diff --git a/tests/test_driver.py b/tests/test_driver.py
index a5517c3..13c8f1d 100644
--- a/tests/test_driver.py
+++ b/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)
diff --git a/tests/test_ipmi.py b/tests/test_ipmi.py
index ddd1266..fdd6d39 100644
--- a/tests/test_ipmi.py
+++ b/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))
diff --git a/tests/test_transport.py b/tests/test_transport.py
index ed8732e..3780b9f 100644
--- a/tests/test_transport.py
+++ b/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"]
diff --git a/tests/test_x10_redfish.py b/tests/test_x10_redfish.py
index 558c281..31c7c5c 100644
--- a/tests/test_x10_redfish.py
+++ b/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"],
+ }
+ ]