diff --git a/README.md b/README.md index 6100000..87e9205 100644 --- a/README.md +++ b/README.md @@ -5,12 +5,11 @@ NAPALM driver for server **baseboard management controllers (BMCs)** over boards that predate Redfish. Built on [napalm-device-types](https://git.netork.io/NAPALM/napalm-device-types). Written for netOrk. -**Status: early development.** The HTTPS transport (`napalm_redfish/transport.py`) and pure -parsers for the Redfish resources in [docs/redfish-map.md](docs/redfish-map.md) -(`napalm_redfish/parsers/`) exist and are tested against DMTF mockups. The driver class and -its `napalm.drivers` entry point wait for the `ManagementControllerDriver` role base in -napalm-device-types. The first test device is a Supermicro X10DRL-i (BMC firmware 3.91, -Redfish). +**Status: early development.** The `redfish` driver (`napalm_redfish.RedfishDriver`, a +`ManagementControllerDriver` from napalm-device-types) implements the NAPALM getters and +the role methods in [docs/redfish-map.md](docs/redfish-map.md), tested end to end against +DMTF mockups. It has not yet run against real hardware; the first test device is a +Supermicro X10DRL-i (BMC firmware 3.91, Redfish). The `ipmi` fallback does not exist yet. ## Why Redfish, and why not one driver per vendor @@ -64,10 +63,31 @@ These follow the other drivers in the `NAPALM` organisation: - `napalm-device-types` is not on PyPI: install it from `git+https://git.netork.io/NAPALM/napalm-device-types.git` before this package. +## Usage + +```python +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": [...]} + bmc.get_hardware_inventory() +``` + +Optional args: `port` (443), `verify_ssl` (`True`, `False` or a CA file path; default +`True`), `auth` (`auto`, `session` or `basic`; `auto` uses a Redfish session and falls back +to basic auth only if the service cannot create sessions), `retries` (1). + +`reboot_host` restarts the BMC, never the server; the server is switched through +`set_power_state`. `get_config` is not implemented: a BMC has no configuration file. + ## Development ```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 ``` diff --git a/napalm_redfish/__init__.py b/napalm_redfish/__init__.py index 265035b..7ff3894 100644 --- a/napalm_redfish/__init__.py +++ b/napalm_redfish/__init__.py @@ -1,3 +1,6 @@ """NAPALM driver for server BMCs over DMTF Redfish.""" +from napalm_redfish.driver import RedfishDriver + __version__ = "0.1.0.dev0" +__all__ = ["RedfishDriver"] diff --git a/napalm_redfish/driver.py b/napalm_redfish/driver.py new file mode 100644 index 0000000..d1a4ece --- /dev/null +++ b/napalm_redfish/driver.py @@ -0,0 +1,542 @@ +"""NAPALM driver for server BMCs over DMTF Redfish. + +The device is the BMC. NAPALM's getters describe the BMC (its ports, users, services); +the ``ManagementControllerDriver`` methods describe and control the server it manages. +``reboot_host`` restarts the BMC (``Manager.Reset``), never the server. + +Optional args +------------- +port : int + HTTPS port (default 443). +verify_ssl : bool or str + ``True`` (default) verifies the certificate against the CA bundle of ``requests``, + a path verifies against that CA file, ``False`` turns verification off (self-signed + BMC certificates are common). +auth : str + ``auto`` (default: session, basic auth if the service cannot create sessions), + ``session`` or ``basic``. +retries : int + Extra attempts after a connection error (default 1). + +The managed server is the first member of ``Systems``; its BMC is the first manager in +its ``Links.ManagedBy`` and its chassis the first in its ``Links.Chassis`` (the first +member of ``Managers``/``Chassis`` when the links are missing). +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone +from typing import Any + +from napalm.base.exceptions import ConnectionException +from napalm_device_types import ManagementControllerDriver +from napalm_device_types.base import FingerprintRule, PortSpec +from napalm_device_types.models import POWER_ACTIONS + +from napalm_redfish.parsers import ( + build_environment, + environment_sources, + parse_account, + parse_action_info, + parse_drive, + parse_ethernet_interface, + parse_fan, + parse_firmware, + parse_log_entry, + parse_log_service, + parse_manager, + parse_memory, + parse_network_protocol, + parse_power, + parse_power_state, + parse_power_subsystem, + parse_power_supply, + parse_processor, + parse_reset_action, + parse_sensor, + parse_service_root, + parse_simple_storage, + parse_storage, + parse_system, + parse_thermal, + parse_thermal_metrics, + parse_thermal_subsystem, + to_napalm_users, +) +from napalm_redfish.parsers.common import link, link_list, odata_id +from napalm_redfish.transport import RedfishError, RedfishNotFound, RedfishTransport + +LOG = logging.getLogger(__name__) + +_JsonDict = dict[str, Any] + +#: ``get_management_protocols`` key -> ManagerNetworkProtocol property. +_PROTOCOLS = { + "ipmi": "IPMI", + "ssh": "SSH", + "http": "HTTP", + "https": "HTTPS", + "snmp": "SNMP", + "kvm": "KVMIP", + "virtual_media": "VirtualMedia", +} +_SEVERITIES = ("OK", "Warning", "Critical") + + +def _timestamp(created: str | None) -> float | None: + """Unix time of a Redfish ``Created`` value; a time without offset counts as UTC.""" + if not created: + return None + try: + when = datetime.fromisoformat(created.replace("Z", "+00:00")) + except ValueError: + return None + if when.tzinfo is None: + when = when.replace(tzinfo=timezone.utc) + return when.timestamp() + + +def _status(item: _JsonDict) -> str: + """Role ``status``: ``Absent`` for an empty slot, else the Redfish health.""" + if not item.get("present", True): + return "Absent" + return item.get("health") or "" + + +class RedfishDriver(ManagementControllerDriver): + """NAPALM driver for Redfish BMCs (Supermicro, iDRAC, iLO, XCC and others).""" + + platform = "redfish" + DRIVER_NAME = "redfish" + #: Every Redfish service answers ``GET /redfish/v1/`` without credentials. + PORT_SPECS = [PortSpec("https", 443, ("/redfish/v1/",))] + HTTP_FINGERPRINT = [FingerprintRule("redfishversion", 10.0), + FingerprintRule("#serviceroot.", 5.0)] + #: A BMC restart takes a few minutes before Redfish answers again. + REBOOT_SETTLE_SECONDS = 180 + + def __init__( + self, + hostname: str, + username: str, + password: str, + timeout: int = 60, + optional_args: _JsonDict | None = None, + ) -> None: + self.hostname = hostname + self.username = username + self.password = password + self.timeout = timeout + self.optional_args: _JsonDict = optional_args or {} + self.transport = RedfishTransport( + hostname, + username, + password, + port=self.optional_args.get("port"), + verify=self.optional_args.get("verify_ssl", True), + auth=self.optional_args.get("auth", "auto"), + timeout=timeout, + retries=self.optional_args.get("retries", 1), + ) + self._cache: _JsonDict = {} + + # --- connection ---------------------------------------------------------------------- + + def open(self) -> None: + try: + self.transport.open() + except RedfishError as exc: + raise ConnectionException(f"Cannot connect to Redfish at {self.hostname}: {exc}") \ + from exc + + def close(self) -> None: + self.transport.close() + self._cache = {} + + def is_alive(self) -> _JsonDict: + try: + self.transport.get("/redfish/v1/") + except Exception: + return {"is_alive": False} + return {"is_alive": True} + + # --- discovery (cached per connection) ------------------------------------------------- + + def _cached(self, key: str, load): + if key not in self._cache: + self._cache[key] = load() + return self._cache[key] + + def _root(self) -> _JsonDict: + return self._cached("root", lambda: parse_service_root(self.transport.root)) + + def _first(self, collection: str) -> _JsonDict: + path = self._root()["links"].get(collection) + if path is None: + raise RedfishError(f"the service root links no {collection}") + members = self.transport.members(path, limit=1) + if not members: + raise RedfishError(f"{path} has no members") + return members[0] + + def _system_raw(self) -> _JsonDict: + return self._cached("system", lambda: self._first("Systems")) + + def _system_path(self) -> str: + return odata_id(self._system_raw()) or "" + + def _linked_or_first(self, key: str, links_key: str, collection: str) -> _JsonDict: + def load() -> _JsonDict: + linked = link_list(self._system_raw(), "Links", links_key) + if linked: + return self.transport.get(linked[0]) + return self._first(collection) + + return self._cached(key, load) + + def _manager_raw(self) -> _JsonDict: + return self._linked_or_first("manager", "ManagedBy", "Managers") + + def _chassis_raw(self) -> _JsonDict: + return self._linked_or_first("chassis", "Chassis", "Chassis") + + def _members_of(self, resource: _JsonDict, key: str) -> list[_JsonDict]: + """Members of the collection ``resource[key]`` links; [] if not linked.""" + path = link(resource, key) + return self.transport.members(path) if path else [] + + def _service(self, name: str) -> _JsonDict | None: + path = self._root()["links"].get(name) + return self.transport.get_optional(path) if path else None + + def _network_protocol(self) -> _JsonDict: + raw = self.transport.follow(self._manager_raw(), "NetworkProtocol") + 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")] + + # --- NAPALM getters ---------------------------------------------------------------- + + def get_facts(self) -> _JsonDict: + 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) + return { + "uptime": -1.0, # Redfish has no uptime + "vendor": system["manufacturer"] or "", + "model": system["model"] or "", + "hostname": hostname, + "fqdn": fqdn, + "os_version": manager["firmware_version"] or "", + "serial_number": system["serial_number"] or "", + "interface_list": sorted(i["name"] for i in interfaces if i["name"]), + } + + def get_interfaces(self) -> _JsonDict: + return {i["name"]: i["interface"] for i in self._manager_interfaces() if i["name"]} + + def get_interfaces_ip(self) -> _JsonDict: + result = {} + for iface in self._manager_interfaces(): + families = {f: iface[f] for f in ("ipv4", "ipv6") if iface[f]} + if iface["name"] and families: + result[iface["name"]] = families + return result + + def get_environment(self) -> _JsonDict: + """Fans, temperatures and power supplies of the chassis; ``cpu`` and ``memory`` are + empty, since a BMC does not know the host OS's load.""" + sources = environment_sources(self._chassis_raw()) + fans: _JsonDict = {} + temperature: _JsonDict = {} + power: _JsonDict = {} + + thermal = sources["thermal"] + if thermal and thermal["schema"] == "Thermal": + legacy = parse_thermal(self.transport.get(thermal["path"])) + 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 [] + if 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"]) + temperature = parse_thermal_metrics(metrics or {})["temperature"] + fans = build_environment(fans=parsed_fans)["fans"] + + supply = sources["power"] + 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"])] + power = build_environment(power_supplies=psus)["power"] + + 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)) + path = subsystem["power_supplies_path"] + return self.transport.members(path) if path else [] + + def get_users(self) -> _JsonDict: + service = self._service("AccountService") + if service is None: + return {} + 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"]} + + def get_snmp_information(self) -> _JsonDict: + """SNMP state from ManagerNetworkProtocol. Redfish hides community strings, so the + ``community`` keys are the communities' names; contact and location are not in + Redfish.""" + snmp = self._network_protocol()["snmp"] + communities = {} + if snmp["enabled"]: + for community in snmp["communities"]: + if community["name"]: + mode = "rw" if community["access"] == "Full" else "ro" + communities[community["name"]] = {"acl": "", "mode": mode} + return {"chassis_id": "", "community": communities, "contact": "", "location": ""} + + # --- role: power ------------------------------------------------------------------- + + def _accepted_actions(self, system: _JsonDict) -> list[str]: + """The PowerAction values the BMC accepts for ``system`` in its current state.""" + reset = parse_reset_action(system) + if reset["target"] is None: + return [] + allowed = reset["allowed"] + if allowed is None and reset["action_info"]: + info = self.transport.get_optional(reset["action_info"]) + allowed = parse_action_info(info) if info else None + if allowed is None: + # Not advertised: assume the standard values; the BMC rejects what it lacks. + allowed = list(POWER_ACTIONS) + actions = [a for a in POWER_ACTIONS if a in allowed] + state = parse_power_state(system) + if state == "Off": + return [a for a in actions if a == "On"] + if state == "On": + return [a for a in actions if a != "On"] + return actions + + 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)} + + def set_power_state(self, action: str) -> None: + if action not in POWER_ACTIONS: + raise ValueError(f"{action!r} is not one of {', '.join(POWER_ACTIONS)}") + system = self.transport.get(self._system_path()) + target = parse_reset_action(system)["target"] + if target is None: + raise RuntimeError(f"{self.hostname}: the system offers no reset action") + accepted = self._accepted_actions(system) + if action not in accepted: + 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'})") + try: + self.transport.post(target, {"ResetType": action}) + except RedfishError as exc: + raise RuntimeError(f"{self.hostname}: {action} refused: {exc}") from exc + + def reboot_host(self) -> None: + """Restart the BMC (``Manager.Reset``, graceful when offered). The server keeps + running; the Redfish session ends with the restart.""" + manager = self.transport.get(odata_id(self._manager_raw()) or "") + reset = parse_reset_action(manager, "#Manager.Reset") + if reset["target"] is None: + 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) + if reset_type is None: + raise RuntimeError(f"{self.hostname}: BMC reset types " + f"{reset['allowed']} include no restart") + body["ResetType"] = reset_type + try: + self.transport.post(reset["target"], body) + except RedfishError as exc: + raise RuntimeError(f"{self.hostname}: BMC reset refused: {exc}") from exc + + # --- role: inventory ----------------------------------------------------------------- + + def get_hardware_inventory(self) -> _JsonDict: + raw = self._system_raw() + system = parse_system(raw) + return { + "manufacturer": system["manufacturer"] or "", + "model": system["model"] or "", + "serial_number": system["serial_number"] or "", + "system_uuid": (system["uuid"] or "").lower(), + "cpus": self._cpus(raw), + "memory": self._dimms(raw), + "drives": self._drives(raw), + "power_supplies": self._psus(), + "nics": self._nics(raw), + } + + def _cpus(self, system: _JsonDict) -> list[_JsonDict]: + cpus = [] + for raw in self._members_of(system, "Processors"): + 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)}) + return cpus + + def _dimms(self, system: _JsonDict) -> list[_JsonDict]: + dimms = [] + 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), + }) + return dimms + + def _drives(self, system: _JsonDict) -> list[_JsonDict]: + parsed: list[_JsonDict] = [] + if link(system, "Storage"): + seen: set[str] = set() + for storage in self._members_of(system, "Storage"): + for path in parse_storage(storage)["drive_paths"]: + if path not in seen: + seen.add(path) + drive = self.transport.get_optional(path) + if drive is not None: + parsed.append(parse_drive(drive)) + else: + 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] + + def _psus(self) -> list[_JsonDict]: + source = environment_sources(self._chassis_raw())["power"] + if source is None: + return [] + if source["schema"] == "Power": + items = self.transport.get(source["path"]).get("PowerSupplies") or [] + else: + items = self._power_supplies_raw(source["path"]) + psus = [] + for raw in items: + 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), + }) + return psus + + def _nics(self, system: _JsonDict) -> list[_JsonDict]: + nics = [] + for raw in self._members_of(system, "EthernetInterfaces"): + 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, + }) + return nics + + # --- role: event log, firmware, protocols -------------------------------------------- + + def get_event_log(self, limit: int = 500) -> list[_JsonDict]: + """Entries of every log service of the system and of the BMC, newest first. + + Where the SEL lives varies by vendor, so all of them are read; entries without a + time sort last. + """ + entries: list[_JsonDict] = [] + for owner in (self._system_raw(), self._manager_raw()): + try: + services = self._members_of(owner, "LogServices") + except RedfishNotFound: + continue + for raw_service in services: + service = parse_log_service(raw_service) + if not service["entries_path"]: + continue + try: + raw_entries = self.transport.members(service["entries_path"]) + except RedfishNotFound: + 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) + return entries[:limit] + + def get_firmware_inventory(self) -> list[_JsonDict]: + service = self._service("UpdateService") + if service is None: + return [] + 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] + + def get_management_protocols(self) -> _JsonDict: + """What the BMC listens on. A protocol the BMC does not report counts as off.""" + protocols = self._network_protocol()["protocols"] + result = {} + for key, name in _PROTOCOLS.items(): + entry = protocols.get(name) or {} + result[key] = {"enabled": bool(entry.get("enabled")), "port": entry.get("port")} + return result diff --git a/pyproject.toml b/pyproject.toml index 2c82a1e..8e4a94b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,9 @@ readme = "README.md" license = { file = "LICENSE" } requires-python = ">=3.10" dependencies = [ + "napalm>=5.0", + # Not on PyPI: install from git before this package (see README). + "napalm-device-types>=4.6", "redfish>=3.2", ] classifiers = [ @@ -25,6 +28,9 @@ test = [ "responses>=0.23", ] +[project.entry-points."napalm.drivers"] +redfish = "napalm_redfish:RedfishDriver" + [tool.setuptools.packages.find] include = ["napalm_redfish*"] diff --git a/tests/test_driver.py b/tests/test_driver.py new file mode 100644 index 0000000..05a068c --- /dev/null +++ b/tests/test_driver.py @@ -0,0 +1,489 @@ +"""RedfishDriver end to end: the DMTF public-rackmount1 mockup over mocked HTTP.""" + +import json +from datetime import datetime +from importlib.metadata import entry_points + +import napalm +import pytest +from conftest import SESSION_PATH, SESSIONS, load +from napalm.base import models as napalm_models +from napalm.base.exceptions import ConnectionException +from napalm_device_types import ManagementControllerDriver, models +from napalm_device_types.roles import primary_role_of + +from napalm_redfish import RedfishDriver + +SYSTEM = "/redfish/v1/Systems/437XR1138R2" +CHASSIS = "/redfish/v1/Chassis/1U" +MANAGER = "/redfish/v1/Managers/BMC" +SYSTEM_LOG = SYSTEM + "/LogServices/Log1/Entries" + + +def assert_shape(value, typed_dict): + """``value`` has exactly the keys of ``typed_dict`` (all of them required).""" + assert isinstance(value, dict) + assert set(value) == set(typed_dict.__annotations__), typed_dict.__name__ + + +def ts(iso: str) -> float: + return datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp() + + +@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"}], + }) + d = RedfishDriver("bmc.example.net", "netork", "s3cret", optional_args={"verify_ssl": False}) + d.open() + yield d + d.close() + + +def without(path: str, *keys: str) -> dict: + data = load("dmtf-public-rackmount1", path) + for key in keys: + data.pop(key) + return data + + +# --- class, registration, connection ------------------------------------------------------- + + +def test_role(): + assert issubclass(RedfishDriver, ManagementControllerDriver) + 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"): + assert hasattr(RedfishDriver, method), method + + +def test_entry_point(): + (ep,) = [e for e in entry_points(group="napalm.drivers") if e.name == "redfish"] + assert ep.load() is RedfishDriver + + +def test_napalm_finds_the_driver(): + assert napalm.get_network_driver("redfish") is RedfishDriver + + +def test_open_and_close(server): + d = RedfishDriver("bmc.example.net", "netork", "s3cret", optional_args={"verify_ssl": False}) + d.open() + assert d.is_alive() == {"is_alive": True} + d.close() + assert d.is_alive() == {"is_alive": False} + assert len(server.calls("POST", SESSIONS)) == 1 + assert len(server.calls("DELETE", SESSION_PATH)) == 1 + + +def test_open_failure_is_connection_exception(server): + server.override("POST", SESSIONS, 401, {}) + d = RedfishDriver("bmc.example.net", "netork", "wrong", optional_args={"verify_ssl": False}) + with pytest.raises(ConnectionException): + d.open() + + +def test_optional_args(server): + 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" + assert d.transport.timeout == 12 + assert RedfishDriver("h", "u", "p").transport.verify is True + + +def test_discovery_is_cached(driver, server): + driver.get_facts() + driver.get_hardware_inventory() + assert len(server.calls("GET", SYSTEM)) == 1 + # The power state is read fresh each time; the collection is still not re-read. + driver.get_power_state() + driver.get_power_state() + assert len(server.calls("GET", SYSTEM)) == 3 + assert len(server.calls("GET", "/redfish/v1/Systems")) == 1 + + +# --- NAPALM getters ------------------------------------------------------------------------ + + +def test_get_facts(driver): + facts = driver.get_facts() + assert_shape(facts, napalm_models.FactsDict) + assert facts == { + "uptime": -1.0, + "vendor": "Contoso", + "model": "3500", + "hostname": "web483-bmc", + "fqdn": "web483-bmc.dmtf.org", + "os_version": "1.45.455b66-rev4", + "serial_number": "437XR1138R2", + "interface_list": ["ToHost", "eth0"], + } + + +def test_get_interfaces(driver): + ifaces = driver.get_interfaces() + assert set(ifaces) == {"eth0", "ToHost"} + assert_shape(ifaces["eth0"], napalm_models.InterfaceDict) + assert ifaces["eth0"]["mac_address"] == "23:11:8A:33:CF:EA" + assert ifaces["eth0"]["is_up"] is True + assert ifaces["eth0"]["speed"] == 100.0 + + +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}}}, + } + + +def test_get_environment_new_resources(driver, server): + env = driver.get_environment() + assert_shape(env, napalm_models.EnvironmentDict) + assert env["cpu"] == {} + assert env["memory"] == {} + assert env["fans"] == { + "Fan Bay 1": {"status": True}, + "Fan Bay 2": {"status": True}, + "Fan for CPU 1": {"status": True}, + "Fan for CPU 2": {"status": True}, + } + assert env["power"] == { + "Power Supply Bay 1": {"status": False, "capacity": 400.0, "output": 937.4}, + } + assert len(env["temperature"]) == 8 + assert env["temperature"]["CPU #1 Temperature"] == { + "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")) + 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["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}} + assert not server.calls("GET", CHASSIS + "/Sensors") + + +def test_get_environment_thermal_metrics_without_sensors(driver, server): + server.override("GET", CHASSIS, 200, without(CHASSIS, "Sensors")) + env = driver.get_environment() + assert env["temperature"] == { + "Intake": {"temperature": 24.8, "is_alert": False, "is_critical": False}, + "Exhaust": {"temperature": 40.5, "is_alert": False, "is_critical": False}, + } + assert len(env["fans"]) == 4 + + +def test_get_users(driver): + users = driver.get_users() + # The collection lists account 1 only; account 2 exists but is not a member. + assert users == {"Administrator": {"level": 15, "password": "", "sshkeys": []}} + + +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")) + assert driver.get_ntp_servers() == { + "0.pool.ntp.org": {}, "1.pool.ntp.org": {}, "time-a.nist.gov": {}} + + +def test_get_snmp_information(driver): + snmp = driver.get_snmp_information() + assert_shape(snmp, napalm_models.SNMPDict) + assert snmp == { + "chassis_id": "", + "contact": "", + "location": "", + "community": { + "Read-only community": {"acl": "", "mode": "ro"}, + "Full access community": {"acl": "", "mode": "rw"}, + }, + } + + +def test_get_config_not_implemented(driver): + with pytest.raises(NotImplementedError): + driver.get_config() + + +# --- role: power --------------------------------------------------------------------------- + + +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"]} + + +def test_get_power_state_off(driver, server): + system = load("dmtf-public-rackmount1", SYSTEM) + system["PowerState"] = "Off" + server.override("GET", SYSTEM, 200, system) + assert driver.get_power_state() == {"state": "Off", "actions": ["On"]} + + +def test_get_power_state_from_action_info(driver, server): + system = load("dmtf-public-rackmount1", SYSTEM) + reset = system["Actions"]["#ComputerSystem.Reset"] + 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"]}]}) + assert driver.get_power_state()["actions"] == ["ForceOff", "PowerCycle"] + + +def test_get_power_state_unlisted_values(driver, server): + # A reset action without allowable values: the standard actions are assumed. + system = load("dmtf-public-rackmount1", SYSTEM) + 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"] + + +def test_get_power_state_without_reset_action(driver, server): + server.override("GET", SYSTEM, 200, without(SYSTEM, "Actions")) + assert driver.get_power_state() == {"state": "On", "actions": []} + + +def test_set_power_state(driver, server): + target = SYSTEM + "/Actions/ComputerSystem.Reset" + server.override("POST", target, 204) + driver.set_power_state("GracefulRestart") + (call,) = server.calls("POST", target) + assert json.loads(call.request.body) == {"ResetType": "GracefulRestart"} + + +def test_set_power_state_invalid_action(driver, server): + with pytest.raises(ValueError): + driver.set_power_state("Nmi") + assert not server.calls("POST", SYSTEM + "/Actions/ComputerSystem.Reset") + + +@pytest.mark.parametrize("action", ["On", "PowerCycle"]) +def test_set_power_state_not_accepted_now(driver, server, action): + # "On" for a server that is on; "PowerCycle" which this BMC does not list. + with pytest.raises(RuntimeError): + driver.set_power_state(action) + assert not server.calls("POST", SYSTEM + "/Actions/ComputerSystem.Reset") + + +def test_set_power_state_refused_by_bmc(driver, server): + target = SYSTEM + "/Actions/ComputerSystem.Reset" + server.override("POST", target, 409, {"error": {"message": "Operation in progress"}}) + with pytest.raises(RuntimeError, match="Operation in progress"): + driver.set_power_state("ForceOff") + + +def test_set_power_state_rereads_state(driver, server): + # The state that gates the action is read fresh, not from the discovery cache. + driver.get_power_state() + system = load("dmtf-public-rackmount1", SYSTEM) + system["PowerState"] = "Off" + server.override("GET", SYSTEM, 200, system) + server.override("POST", SYSTEM + "/Actions/ComputerSystem.Reset", 204) + driver.set_power_state("On") + + +def test_reboot_host_resets_the_bmc(driver, server): + target = MANAGER + "/Actions/Manager.Reset" + server.override("POST", target, 204) + driver.reboot_host() + (call,) = server.calls("POST", target) + assert json.loads(call.request.body) == {"ResetType": "GracefulRestart"} + # Never the managed server. + assert not server.calls("POST", SYSTEM + "/Actions/ComputerSystem.Reset") + + +def test_reboot_host_force_only(driver, server): + manager = load("dmtf-public-rackmount1", MANAGER) + manager["Actions"]["#Manager.Reset"]["ResetType@Redfish.AllowableValues"] = ["ForceRestart"] + server.override("GET", MANAGER, 200, manager) + server.override("POST", MANAGER + "/Actions/Manager.Reset", 204) + driver.reboot_host() + (call,) = server.calls("POST", MANAGER + "/Actions/Manager.Reset") + assert json.loads(call.request.body) == {"ResetType": "ForceRestart"} + + +def test_reboot_host_without_action(driver, server): + server.override("GET", MANAGER, 200, without(MANAGER, "Actions")) + with pytest.raises(RuntimeError): + driver.reboot_host() + + +# --- role: inventory, logs, firmware, protocols -------------------------------------------- + + +def test_get_hardware_inventory(driver): + inv = driver.get_hardware_inventory() + assert_shape(inv, models.HardwareInventoryDict) + assert inv["manufacturer"] == "Contoso" + assert inv["model"] == "3500" + assert inv["serial_number"] == "437XR1138R2" + assert inv["system_uuid"] == "38947555-7742-3448-3784-823347823834" + + for cpu in inv["cpus"]: + 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 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"][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 [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"}, + ] + + 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["ToManager"]["link_up"] is None + + +def test_get_hardware_inventory_storage_drives(driver, server): + # A system with Storage reads the drives each storage subsystem links. + system = load("dmtf-public-rackmount1", SYSTEM) + 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")) + 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"} + + +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"}, + ] + + +def test_get_event_log(driver): + log = driver.get_event_log() + 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"}, + ] + + +def test_get_event_log_limit(driver): + log = driver.get_event_log(limit=2) + assert [e["id"] for e in log] == [SYSTEM_LOG + "/3", SYSTEM_LOG + "/2"] + + +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"}, + ]}) + 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 + assert log["no time"]["severity"] == "OK" + assert log["bad time"]["timestamp"] is None + # Entries without a time sort last. + assert list(log)[0] == "offset" + assert set(list(log)[-2:]) == {"no time", "bad time"} + + +def test_get_firmware_inventory(driver): + fw = driver.get_firmware_inventory() + for item in fw: + assert_shape(item, models.FirmwareDict) + assert fw == [ + {"name": "Contoso BMC Firmware", "version": "1.45.455b66-rev4", "updateable": True}, + {"name": "Contoso Simple Storage Firmware", "version": "2.50", "updateable": True}, + {"name": "Contoso BIOS Firmware", "version": "P79 v1.45", "updateable": True}, + ] + + +def test_get_management_protocols(driver): + protocols = driver.get_management_protocols() + assert_shape(protocols, models.ManagementProtocolsDict) + assert protocols == { + "ipmi": {"enabled": True, "port": 623}, + "ssh": {"enabled": True, "port": 22}, + "http": {"enabled": True, "port": 80}, + "https": {"enabled": True, "port": 443}, + "snmp": {"enabled": True, "port": 161}, + "kvm": {"enabled": True, "port": 5288}, + "virtual_media": {"enabled": True, "port": 17988}, + } + + +def test_get_management_protocols_missing(driver, server): + 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} + assert protocols["ssh"] == {"enabled": False, "port": None}