"""RedfishDriver end to end: the DMTF public-rackmount1 mockup over mocked HTTP.""" import json from datetime import datetime from importlib.metadata import entry_points from pathlib import Path 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} 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())) d = RedfishDriver("bmc.example.net", "netork", "s3cret", optional_args={"verify_ssl": False}) d.open() return d def test_unlicensed_bmc_still_opens(server): # Reachable, just refusing: the poll goes through and says why instead of # reporting the BMC offline. 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": [], } 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")] # Refused altogether: nothing is withheld in part. assert all("withholds" not in lic and "Not licensed" in lic["detail"] for lic in licenses) def test_a_licensed_bmc_reports_no_license_facts(server): d = RedfishDriver("bmc.example.net", "netork", "s3cret", optional_args={"verify_ssl": False}) d.open() assert d.get_licenses() == [] def test_unlicensed_bmc_reports_the_missing_license(server): (warning,) = _unlicensed(server).get_device_warnings() assert warning["code"] == "bmc_license_required" assert warning["meta"]["licenses"] == "SUM DCMS OOB" assert "Not licensed" in warning["meta"]["message"] # Raw signal only: presentation is netOrk's. assert set(warning) == {"code", "meta"} def test_unlicensed_bmc_refuses_everything_else(server): d = _unlicensed(server) with pytest.raises(ConnectionException, match="license"): d.get_interfaces() with pytest.raises(ConnectionException, match="license"): d.get_hardware_inventory() def test_licensed_bmc_has_no_warnings(driver): assert driver.get_device_warnings() == []