Files
napalm-redfish/tests/test_driver.py
T
christianmanivong 93bbc63f3d fix: stop scoring every HTTPS port as a BMC; name the missing license
netOrk's discovery took six switches for BMCs: the PortSpec for 443
added 5 points to every device with a web UI, and nothing else scored
them. The port now carries no weight.

The real BMC, a Supermicro X10DRL-i, was not recognised at all. Its web
UI (ATEN firmware) gets a rule, as do iDRAC and iLO; discovery fetches
those landing pages today. A rule for a Redfish error body covers the
service root once discovery reads it.

That board refuses Redfish without a Supermicro license, even the
service root (403 OemLicenseNotPassed, "SUM DCMS OOB"). The library
reports only the status, so open() reads the error body once more and
raises RedfishLicenseError naming the licenses.

Release 0.1.1.
2026-10-13 09:58:40 +02:00

504 lines
20 KiB
Python

"""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 test_unlicensed_bmc_says_so(server):
# A Supermicro X10 refuses Redfish entirely until a license is activated.
x10 = Path(__file__).parent / "fixtures" / "supermicro-x10drl-i-3.91"
body = json.loads((x10 / "service_root_403.json").read_text())
server.override("GET", "/redfish/v1/", 403, body)
d = RedfishDriver("bmc.example.net", "netork", "s3cret", optional_args={"verify_ssl": False})
with pytest.raises(ConnectionException) as exc:
d.open()
message = str(exc.value)
assert "license" in message.lower()
assert "SUM DCMS OOB" in message