Christian Manivong
09029ca0a4
Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100, Python 3.10). Mapping and Iterable come from collections.abc, imports are sorted. What ruff cannot fix comes in the next commit.
686 lines
23 KiB
Python
686 lines
23 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 _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() == []
|