Add parsers for the Redfish resources in redfish-map.md

Pure functions for ServiceRoot, ComputerSystem (facts, power state, reset action and ActionInfo), Processors, Memory, Storage/Drives/SimpleStorage, Chassis Thermal and Power as well as ThermalSubsystem/PowerSubsystem/Sensors (with environment_sources to pick whichever exists), Manager, EthernetInterfaces, ManagerNetworkProtocol, accounts, firmware inventory and log services/entries.
This commit is contained in:
Christian Manivong committed 2026-10-12 23:42:14 +00:00
1 parent 8998e1342e
commit cab1c84cba
18 files changed
+1926

No files matched your search

+52
View File
@@ -0,0 +1,52 @@
from napalm_redfish.parsers import parse_account, to_napalm_users
def test_account(rackmount):
acc = parse_account(rackmount("/redfish/v1/AccountService/Accounts/1"))
assert acc == {
"id": "1",
"username": "Administrator",
"role_id": "Administrator",
"enabled": True,
"locked": False,
"account_types": ["Redfish"],
}
def test_account_snmp_user(rackmount):
acc = parse_account(rackmount("/redfish/v1/AccountService/Accounts/2"))
assert acc["username"] == "contoso_employee457"
assert acc["account_types"] == ["Redfish", "SNMP"]
def test_empty_account_slot_is_none():
# iDRAC and Supermicro list every slot, unused ones with an empty UserName.
assert parse_account({"Id": "3", "UserName": "", "Enabled": False, "RoleId": "None"}) is None
assert parse_account({"Id": "4"}) is None
def test_account_defaults():
acc = parse_account({"Id": "2", "UserName": "ADMIN"})
assert acc["enabled"] is None
assert acc["locked"] is None
assert acc["role_id"] is None
assert acc["account_types"] == []
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.append(None)
assert to_napalm_users(accounts) == {
"Administrator": {"level": 15, "password": "", "sshkeys": []},
"contoso_employee457": {"level": 15, "password": "", "sshkeys": []},
"netork": {"level": 1, "password": "", "sshkeys": []},
"ops": {"level": 5, "password": "", "sshkeys": []},
"custom": {"level": 0, "password": "", "sshkeys": []},
}
+283
View File
@@ -0,0 +1,283 @@
import pytest
from napalm_redfish.parsers import (
build_environment,
environment_sources,
parse_fan,
parse_power,
parse_power_subsystem,
parse_power_supply,
parse_sensor,
parse_thermal,
parse_thermal_metrics,
parse_thermal_subsystem,
)
CHASSIS = "/redfish/v1/Chassis/1U"
# --- choosing legacy or new resources ------------------------------------------------------
def test_sources_prefer_new_resources(rackmount):
# This chassis links both; the new ones are current, Thermal/Power are deprecated.
src = environment_sources(rackmount(CHASSIS))
assert src == {
"thermal": {"schema": "ThermalSubsystem", "path": CHASSIS + "/ThermalSubsystem"},
"power": {"schema": "PowerSubsystem", "path": CHASSIS + "/PowerSubsystem"},
"sensors": CHASSIS + "/Sensors",
}
def test_sources_legacy_only(localstorage):
src = environment_sources(localstorage(CHASSIS))
assert src == {
"thermal": {"schema": "Thermal", "path": CHASSIS + "/Thermal"},
"power": {"schema": "Power", "path": CHASSIS + "/Power"},
"sensors": None,
}
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"},
})
assert src["thermal"]["schema"] == "ThermalSubsystem"
assert src["power"]["schema"] == "Power"
assert environment_sources({}) == {"thermal": None, "power": None, "sensors": None}
# --- legacy Thermal and Power --------------------------------------------------------------
def test_thermal(rackmount):
env = parse_thermal(rackmount(CHASSIS + "/Thermal"))
assert env["temperature"] == {
"CPU1 Temp": {"temperature": 41.0, "is_alert": False, "is_critical": False},
# "CPU2 Temp" is Disabled and has no reading: left out.
"Chassis Intake Temp": {"temperature": 25.0, "is_alert": False, "is_critical": False},
}
assert env["fans"] == {
"BaseBoard System Fan": {"status": True},
"BaseBoard System Fan Backup": {"status": True},
}
@pytest.mark.parametrize(
"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
(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,
}]})
assert env["temperature"]["T"] == {
"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},
],
})
assert env["fans"] == {
"FAN1": {"status": True},
"FAN3": {"status": False},
"FAN1 (3)": {"status": True},
}
assert list(env["temperature"]) == ["System Temp", "System Temp (1)"]
def test_power(rackmount):
env = parse_power(rackmount(CHASSIS + "/Power"))
assert env["power"] == {
# Health Warning: the supply works but reports a problem.
"Power Supply Bay": {"status": False, "capacity": 800.0, "output": 325.0},
}
assert env["consumed_watts"] == 344.0
assert env["voltages"] == {
"VRM1 Voltage": {"voltage": 12.0, "is_alert": False, "is_critical": False},
"VRM2 Voltage": {"voltage": 5.0, "is_alert": False, "is_critical": False},
}
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"}},
]})
assert env["power"] == {"PS1": {"status": True, "capacity": -1.0, "output": -1.0}}
assert env["consumed_watts"] is None
assert env["voltages"] == {}
# --- ThermalSubsystem, PowerSubsystem, Sensors ---------------------------------------------
def test_thermal_subsystem(rackmount):
ts = parse_thermal_subsystem(rackmount(CHASSIS + "/ThermalSubsystem"))
assert ts == {
"fans_path": CHASSIS + "/ThermalSubsystem/Fans",
"thermal_metrics_path": CHASSIS + "/ThermalSubsystem/ThermalMetrics",
"state": "Enabled",
"health": "OK",
}
def test_power_subsystem(rackmount):
ps = parse_power_subsystem(rackmount(CHASSIS + "/PowerSubsystem"))
assert ps == {
"power_supplies_path": CHASSIS + "/PowerSubsystem/PowerSupplies",
"capacity_watts": 2000.0,
"state": "Enabled",
"health": "OK",
}
def test_fan(rackmount):
fan = parse_fan(rackmount(CHASSIS + "/ThermalSubsystem/Fans/CPU1"))
assert fan == {
"id": "CPU1",
"name": "Fan for CPU 1",
"present": True,
"status": True,
"speed_rpm": 1490,
"speed_percent": 45.0,
"state": "Enabled",
"health": "OK",
}
def test_power_supply_with_metrics(rackmount):
base = CHASSIS + "/PowerSubsystem/PowerSupplies/Bay1"
psu = parse_power_supply(rackmount(base), rackmount(base + "/Metrics"))
assert psu == {
"id": "Bay1",
"name": "Power Supply Bay 1",
"present": True,
"status": False, # predicted failure, Health Warning
"capacity": 400.0,
"output": 937.4,
"manufacturer": "Contoso Power",
"model": "RKS-440DC",
"serial_number": "3488247",
"part_number": "23456-133",
"firmware_version": "1.00",
"state": "Enabled",
"health": "Warning",
}
assert psu["output"] == 937.4
def test_power_supply_without_metrics(rackmount):
base = CHASSIS + "/PowerSubsystem/PowerSupplies/Bay1"
assert parse_power_supply(rackmount(base))["output"] == -1.0
absent = parse_power_supply(rackmount(CHASSIS + "/PowerSubsystem/PowerSupplies/Bay2"))
assert absent["present"] is False
assert absent["capacity"] == -1.0
def test_sensor_temperature_by_units(rackmount):
# No ReadingType on this sensor: "Cel" makes it a temperature.
s = parse_sensor(rackmount(CHASSIS + "/Sensors/CPU1Temp"))
assert s["id"] == "CPU1Temp"
assert s["name"] == "CPU #1 Temperature"
assert s["reading_type"] == "Temperature"
assert s["reading"] == 44.0
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
def test_sensor_power(rackmount):
s = parse_sensor(rackmount(CHASSIS + "/Sensors/TotalPower"))
assert s["reading_type"] == "Power"
assert s["reading"] == 374.0
assert s["electrical_context"] == "Total"
assert s["is_alert"] is False
def test_sensor_voltage_and_fan(rackmount):
v = parse_sensor(rackmount(CHASSIS + "/Sensors/PS1InputVoltage"))
assert v["reading_type"] == "Voltage"
assert v["reading"] == 119.27
assert v["is_alert"] is False
fan = parse_sensor(rackmount(CHASSIS + "/Sensors/CPUFan1"))
assert fan["reading_type"] == "Percent"
assert fan["reading"] == 45.0
def test_sensor_without_reading():
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},
}}
def test_build_environment_from_new_resources(rackmount):
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")]
env = build_environment(fans=fans, power_supplies=psus, sensors=sensors)
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 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},
}
assert env["consumed_watts"] == 374.0
assert env["voltages"] == {
"Power Supply #1 Input Voltage": {
"voltage": 119.27, "is_alert": False, "is_critical": False},
}
def test_build_environment_empty():
assert build_environment() == {
"fans": {}, "temperature": {}, "power": {}, "voltages": {}, "consumed_watts": None}
+31
View File
@@ -0,0 +1,31 @@
from napalm_redfish.parsers import parse_firmware
INVENTORY = "/redfish/v1/UpdateService/FirmwareInventory"
def test_firmware_bmc(rackmount):
fw = parse_firmware(rackmount(INVENTORY + "/BMC"))
assert fw == {
"id": "BMC",
"name": "Contoso BMC Firmware",
"version": "1.45.455b66-rev4",
"updateable": True,
"manufacturer": "Contoso",
"release_date": "2017-08-22T12:00:00Z",
"state": "Enabled",
"health": "OK",
"related": ["/redfish/v1/Managers/BMC"],
}
def test_firmware_bios(rackmount):
fw = parse_firmware(rackmount(INVENTORY + "/BIOS"))
assert fw["version"] == "P79 v1.45"
assert fw["related"] == ["/redfish/v1/Systems/437XR1138R2/Bios"]
def test_firmware_minimal():
fw = parse_firmware({"Id": "BMC", "Name": "BMC", "Version": "03.91"})
assert fw["version"] == "03.91"
assert fw["updateable"] is None
assert fw["related"] == []
+175
View File
@@ -0,0 +1,175 @@
from napalm_redfish.parsers import (
parse_drive,
parse_memory,
parse_processor,
parse_simple_storage,
parse_storage,
)
SYSTEM = "/redfish/v1/Systems/437XR1138R2"
# --- Processors ----------------------------------------------------------------------------
def test_processor(rackmount):
cpu = parse_processor(rackmount(SYSTEM + "/Processors/CPU1"))
assert cpu == {
"id": "CPU1",
"name": "Processor",
"socket": "CPU 1",
"processor_type": "CPU",
"manufacturer": "Intel(R) Corporation",
"model": "Multi-Core Intel(R) Xeon(R) processor 7xxx Series",
"architecture": "x86",
"instruction_set": "x86-64",
"max_speed_mhz": 3700,
"cores": 8,
"threads": 16,
"serial_number": None,
"present": True,
"state": "Enabled",
# The CPU itself is OK; its HealthRollup (temperature warning) is not its health.
"health": "OK",
}
def test_processor_absent(rackmount):
cpu = parse_processor(rackmount(SYSTEM + "/Processors/CPU2"))
assert cpu["present"] is False
assert cpu["state"] == "Absent"
assert cpu["health"] is None
assert cpu["socket"] == "CPU 2"
assert cpu["model"] is None
assert cpu["cores"] is None
def test_processor_fpga(rackmount):
fpga = parse_processor(rackmount(SYSTEM + "/Processors/FPGA1"))
assert fpga["processor_type"] == "FPGA"
assert fpga["model"] == "Stratix 10"
# No Socket property: the location label stands in.
assert fpga["socket"] == "U168"
# --- Memory --------------------------------------------------------------------------------
def test_memory(rackmount):
dimm = parse_memory(rackmount(SYSTEM + "/Memory/DIMM1"))
assert dimm == {
"id": "DIMM1",
"name": "DIMM Slot 1",
"location": "DIMM 1",
"capacity_mib": 32768,
"memory_device_type": "DDR4",
"base_module_type": "RDIMM",
"speed_mhz": None,
"manufacturer": None,
"part_number": None,
"serial_number": None,
"present": True,
"state": "Enabled",
"health": "OK",
}
def test_memory_absent(rackmount):
dimm = parse_memory(rackmount(SYSTEM + "/Memory/DIMM4"))
assert dimm["present"] is False
assert dimm["capacity_mib"] is None
assert dimm["location"] == "DIMM 4"
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"},
})
assert dimm["location"] == "P1-DIMMA1"
assert dimm["speed_mhz"] == 2133
assert dimm["manufacturer"] == "Samsung"
assert dimm["serial_number"] == "12345678"
def test_memory_zero_capacity_is_empty_slot():
# Some BMCs list empty slots as enabled with a capacity of 0.
dimm = parse_memory({"Id": "DIMM9", "CapacityMiB": 0, "Status": {"State": "Enabled"}})
assert dimm["present"] is False
# --- Storage -------------------------------------------------------------------------------
def test_storage(localstorage):
st = parse_storage(localstorage(SYSTEM + "/Storage/1"))
assert st["id"] == "1"
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["drive_paths"] == [
"/redfish/v1/Chassis/1U/Drives/35D38F11ACEF7BD3",
"/redfish/v1/Chassis/1U/Drives/3F5A8C54207B7233",
"/redfish/v1/Chassis/1U/Drives/32ADF365C6C1B7BD",
"/redfish/v1/Chassis/1U/Drives/3D58ECBC375FD9F2",
]
assert st["controllers_path"] is None
def test_storage_with_controller_collection():
# Storage 1.15 and later link a StorageController collection instead of an array.
st = parse_storage({"Id": "RAID", "Controllers": {"@odata.id": "/redfish/v1/x/Controllers"}})
assert st["controllers"] == []
assert st["controllers_path"] == "/redfish/v1/x/Controllers"
assert st["drive_paths"] == []
def test_drive(localstorage):
drive = parse_drive(localstorage("/redfish/v1/Chassis/1U/Drives/35D38F11ACEF7BD3"))
assert drive == {
"id": "35D38F11ACEF7BD3",
"name": "Drive Sample",
"manufacturer": "Contoso",
"model": "C123",
"serial_number": "1234567",
"part_number": "C123-1111",
"revision": "100A",
"capacity_bytes": 899527000064,
"media_type": "HDD",
"protocol": "SAS",
"failure_predicted": False,
"present": True,
"state": "Enabled",
"health": "OK",
}
def test_simple_storage(rackmount):
st = parse_simple_storage(rackmount(SYSTEM + "/SimpleStorage/1"))
assert st["id"] == "1"
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"]
bay1, bay2, bay3, _ = st["drives"]
assert bay1["model"] == "3000GT8"
assert bay1["capacity_bytes"] == 8000000000000
assert bay1["present"] is True
assert bay2["health"] == "Warning"
assert bay3["present"] is False
assert bay3["model"] is None
# Same keys as parse_drive, so callers can treat both alike.
assert set(bay1) == set(parse_drive({}))
+64
View File
@@ -0,0 +1,64 @@
from napalm_redfish.parsers import parse_log_entry, parse_log_service
SYSTEM_LOG = "/redfish/v1/Systems/437XR1138R2/LogServices/Log1"
MANAGER_LOG = "/redfish/v1/Managers/BMC/LogServices/Log"
def test_log_service(rackmount):
ls = parse_log_service(rackmount(MANAGER_LOG))
assert ls == {
"id": "Log",
"name": "System Log Service",
"enabled": True,
"entries_path": MANAGER_LOG + "/Entries",
"max_records": 1000,
"overwrite_policy": "WrapsWhenFull",
"entry_type": None,
"state": "Enabled",
"health": "OK",
}
def test_log_service_sel():
entries = {"@odata.id": "/redfish/v1/Systems/1/LogServices/SEL/Entries"}
ls = parse_log_service({"Id": "SEL", "LogEntryType": "SEL", "Entries": entries})
assert ls["entry_type"] == "SEL"
assert ls["entries_path"] == "/redfish/v1/Systems/1/LogServices/SEL/Entries"
assert ls["enabled"] is None
def test_sel_entry(rackmount):
(first, second) = rackmount(SYSTEM_LOG + "/Entries")["Members"]
entry = parse_log_entry(first)
assert entry == {
"id": "1",
"created": "2012-03-07T14:44:00Z",
"severity": "Critical",
"message": "Temperature threshold exceeded",
"message_id": "0x592A28",
"message_args": [],
"entry_type": "SEL",
"entry_code": "Upper Critical - going high",
"sensor_type": "Temperature",
"sensor_number": 1,
"origin": "/redfish/v1/Chassis/1U/Thermal",
}
assert parse_log_entry(second)["sensor_number"] == 2
def test_event_entry(rackmount):
(entry,) = rackmount(MANAGER_LOG + "/Entries")["Members"]
parsed = parse_log_entry(entry)
assert parsed["entry_type"] == "Event"
assert parsed["message"] == "System May be Melting"
assert parsed["message_id"] == "Event.1.0.TempWayTooHot"
assert parsed["message_args"] == ["88"]
assert parsed["entry_code"] is None
def test_entry_minimal():
entry = parse_log_entry({"Id": "7", "Message": " Power supply AC lost \n"})
assert entry["message"] == "Power supply AC lost"
assert entry["created"] is None
assert entry["severity"] is None
assert entry["origin"] is None
+159
View File
@@ -0,0 +1,159 @@
import pytest
from napalm_redfish.parsers import (
parse_ethernet_interface,
parse_manager,
parse_network_protocol,
)
MANAGER = "/redfish/v1/Managers/BMC"
def test_manager(rackmount):
m = parse_manager(rackmount(MANAGER))
assert m["id"] == "BMC"
assert m["name"] == "Manager"
assert m["manager_type"] == "BMC"
assert m["model"] == "Joo Janta 200"
assert m["firmware_version"] == "1.45.455b66-rev4"
assert m["uuid"] == "58893887-8974-2487-2389-841168418919"
assert m["power_state"] == "On"
assert m["date_time"] == "2015-03-13T04:14:33+06:00"
assert m["state"] == "Enabled"
assert m["health"] == "OK"
assert m["links"]["EthernetInterfaces"] == MANAGER + "/EthernetInterfaces"
assert m["links"]["NetworkProtocol"] == MANAGER + "/NetworkProtocol"
assert m["links"]["LogServices"] == MANAGER + "/LogServices"
assert m["manager_for_servers"] == ["/redfish/v1/Systems/437XR1138R2"]
assert m["manager_for_chassis"] == ["/redfish/v1/Chassis/1U"]
def test_manager_minimal():
m = parse_manager({"Id": "1", "FirmwareVersion": " 3.91 "})
assert m["firmware_version"] == "3.91"
assert m["model"] is None
assert m["links"] == {}
assert m["manager_for_servers"] == []
# --- EthernetInterfaces --------------------------------------------------------------------
def test_ethernet_interface(rackmount):
eth = parse_ethernet_interface(rackmount(MANAGER + "/EthernetInterfaces/eth0"))
assert eth["name"] == "eth0"
assert eth["interface"] == {
"is_up": True,
"is_enabled": True,
"description": "Management network interface on a dedicated port",
"last_flapped": -1.0,
"speed": 100.0,
"mtu": 1500,
"mac_address": "23:11:8A:33:CF:EA",
}
assert eth["ipv4"] == {"192.168.0.10": {"prefix_length": 22}}
assert eth["ipv6"] == {"fe80::1ec1:deff:fe6f:1e24": {"prefix_length": 64}}
assert eth["hostname"] == "web483-bmc"
assert eth["fqdn"] == "web483-bmc.dmtf.org"
assert eth["vlan"] == 101
assert eth["permanent_mac_address"] == "23:11:8A:33:CF:EA"
def test_ethernet_interface_without_link_status(rackmount):
# No LinkStatus (as on older schemas): enabled and Status Enabled counts as up.
eth = parse_ethernet_interface(rackmount(MANAGER + "/EthernetInterfaces/ToHost"))
assert eth["interface"]["is_up"] is True
assert eth["ipv4"] == {"192.168.20.55": {"prefix_length": 24}}
assert eth["vlan"] is None
@pytest.mark.parametrize(
"props, is_up, is_enabled",
[
({"LinkStatus": "LinkDown", "InterfaceEnabled": True}, False, True),
({"LinkStatus": "NoLink", "InterfaceEnabled": True}, False, True),
({"LinkStatus": "LinkUp", "InterfaceEnabled": False}, False, False),
({"Status": {"State": "Disabled"}}, False, False),
({"Status": {"State": "Enabled"}}, True, True),
({}, False, False),
],
)
def test_ethernet_link_state(props, is_up, is_enabled):
eth = parse_ethernet_interface({"Id": "1", **props})
assert eth["interface"]["is_up"] is is_up
assert eth["interface"]["is_enabled"] is 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"},
],
})
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}}
# --- NetworkProtocol -----------------------------------------------------------------------
def test_network_protocol(rackmount):
np = parse_network_protocol(rackmount(MANAGER + "/NetworkProtocol"))
assert np["hostname"] == "web483-bmc"
assert np["fqdn"] == "web483-bmc.dmtf.org"
assert np["protocols"] == {
"HTTP": {"enabled": True, "port": 80},
"HTTPS": {"enabled": True, "port": 443},
"IPMI": {"enabled": True, "port": 623},
"SSH": {"enabled": True, "port": 22},
"SNMP": {"enabled": True, "port": 161},
"VirtualMedia": {"enabled": True, "port": 17988},
"SSDP": {"enabled": True, "port": 1900},
"Telnet": {"enabled": True, "port": 23},
"KVMIP": {"enabled": True, "port": 5288},
}
assert np["ntp"] == {"enabled": None, "servers": []}
assert np["snmp"] == {
"enabled": True, "port": 161, "v1": True, "v2c": False, "v3": True,
"communities": [
{"name": "Read-only community", "access": "Limited"},
{"name": "Full access community", "access": "Full"},
],
}
def test_network_protocol_ntp(telemetry):
np = parse_network_protocol(telemetry("/redfish/v1/Managers/1/NetworkProtocol"))
assert np["ntp"] == {
"enabled": True,
"servers": ["0.pool.ntp.org", "1.pool.ntp.org", "time-a.nist.gov"],
}
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": []}
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": []}
+46
View File
@@ -0,0 +1,46 @@
from napalm_redfish.parsers import parse_service_root
def test_service_root(rackmount):
root = parse_service_root(rackmount("/redfish/v1/"))
assert root["redfish_version"] == "1.15.0"
assert root["uuid"] == "92384634-2938-2342-8820-489239905423"
assert root["name"] == "Root Service"
# Vendor and Product only exist from ServiceRoot 1.5 on, and this mockup has neither.
assert root["vendor"] is None
assert root["product"] is None
assert root["oem_vendors"] == []
def test_service_root_links(rackmount):
links = parse_service_root(rackmount("/redfish/v1/"))["links"]
assert links["Systems"] == "/redfish/v1/Systems"
assert links["Chassis"] == "/redfish/v1/Chassis"
assert links["Managers"] == "/redfish/v1/Managers"
assert links["AccountService"] == "/redfish/v1/AccountService"
assert links["UpdateService"] == "/redfish/v1/UpdateService"
assert links["SessionService"] == "/redfish/v1/SessionService"
# Links.Sessions is where the session login goes.
assert links["Sessions"] == "/redfish/v1/SessionService/Sessions"
assert "@odata.id" not in links
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"},
})
assert root["vendor"] == "Supermicro"
assert root["product"] == "X10DRL-i"
assert root["oem_vendors"] == ["Supermicro"]
assert root["uuid"] is None
assert root["links"] == {"Systems": "/redfish/v1/Systems"}
def test_service_root_empty():
root = parse_service_root({})
assert root["redfish_version"] is None
assert root["links"] == {}
+129
View File
@@ -0,0 +1,129 @@
import pytest
from napalm_redfish.parsers import (
parse_action_info,
parse_power_state,
parse_reset_action,
parse_system,
)
SYSTEM = "/redfish/v1/Systems/437XR1138R2"
def test_system_facts(rackmount):
s = parse_system(rackmount(SYSTEM))
assert s["id"] == "437XR1138R2"
assert s["name"] == "WebFrontEnd483"
assert s["manufacturer"] == "Contoso"
assert s["model"] == "3500"
assert s["serial_number"] == "437XR1138R2"
assert s["uuid"] == "38947555-7742-3448-3784-823347823834"
assert s["hostname"] == "web483"
assert s["sku"] == "8675309"
assert s["part_number"] == "224071-J23"
assert s["asset_tag"] == "Chicago-45Z-2381"
assert s["bios_version"] == "P79 v1.45 (12/06/2017)"
assert s["system_type"] == "Physical"
assert s["power_state"] == "On"
assert s["state"] == "Enabled"
assert s["health"] == "OK"
assert s["processor_count"] == 2
assert s["processor_model"] == "Multi-Core Intel(R) Xeon(R) processor 7xxx Series"
assert s["memory_gib"] == 96
def test_system_links(rackmount):
links = parse_system(rackmount(SYSTEM))["links"]
assert links["Processors"] == SYSTEM + "/Processors"
assert links["Memory"] == SYSTEM + "/Memory"
assert links["SimpleStorage"] == SYSTEM + "/SimpleStorage"
assert links["EthernetInterfaces"] == SYSTEM + "/EthernetInterfaces"
assert links["LogServices"] == SYSTEM + "/LogServices"
assert "Storage" not in links
s = parse_system(rackmount(SYSTEM))
assert s["chassis"] == ["/redfish/v1/Chassis/1U"]
assert s["managed_by"] == ["/redfish/v1/Managers/BMC"]
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",
})
assert s["manufacturer"] == "Supermicro"
assert s["model"] == "X10DRL-i"
assert s["serial_number"] is None
assert s["sku"] is None
assert s["hostname"] is None
assert s["uuid"] is None
assert s["part_number"] is None
assert s["processor_count"] is None
assert s["memory_gib"] is None
assert s["links"] == {}
assert s["chassis"] == []
def test_power_state(rackmount):
assert parse_power_state(rackmount(SYSTEM)) == "On"
@pytest.mark.parametrize("value", ["On", "Off", "PoweringOn", "PoweringOff", "Paused"])
def test_power_state_values(value):
assert parse_power_state({"PowerState": value}) == value
def test_power_state_missing():
assert parse_power_state({}) is None
assert parse_power_state({"PowerState": None}) is None
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",
]
assert reset["action_info"] is None
def test_reset_action_with_action_info(telemetry):
reset = parse_reset_action(telemetry("/redfish/v1/Systems/1"))
assert reset["target"] == "/redfish/v1/Systems/1/Actions/ComputerSystem.Reset"
# Values live in the ActionInfo resource, not inline: unknown until that is read.
assert reset["allowed"] is None
assert reset["action_info"] == "/redfish/v1/Systems/1/ResetActionInfo"
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"]
def test_action_info_other_parameter():
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
def test_reset_action_manager(rackmount):
reset = parse_reset_action(rackmount("/redfish/v1/Managers/BMC"), "#Manager.Reset")
assert reset["target"] == "/redfish/v1/Managers/BMC/Actions/Manager.Reset"
assert reset["allowed"] == ["ForceRestart", "GracefulRestart"]
def test_reset_action_missing():
assert parse_reset_action({}) == {"target": None, "allowed": None, "action_info": None}
# Reset advertised without allowable values: allowed stays unknown, not empty.
reset = parse_reset_action({"Actions": {"#ComputerSystem.Reset": {"target": "/x"}}})
assert reset == {"target": "/x", "allowed": None, "action_info": None}