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.
400 lines
13 KiB
Python
400 lines
13 KiB
Python
"""IpmiDriver against recorded IPMI answers of a Supermicro X10DRL-i (BMC 3.91).
|
|
|
|
The board refuses Redfish without a license; IPMI answers without one. The driver
|
|
reads, and switches the server's power; it does not reset the BMC.
|
|
"""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from napalm.base.exceptions import ConnectionException
|
|
from napalm_device_types import ManagementControllerDriver, models
|
|
from napalm_device_types.roles import primary_role_of
|
|
|
|
import napalm_redfish.ipmi as ipmi
|
|
from napalm_redfish.ipmi import IpmiDriver
|
|
|
|
X10 = Path(__file__).parent / "fixtures" / "supermicro-x10drl-i-3.91" / "ipmi"
|
|
|
|
|
|
def fixture(name):
|
|
return json.loads((X10 / f"{name}.json").read_text())
|
|
|
|
|
|
class FakeCommand:
|
|
"""pyghmi's Command, answering from the recorded files."""
|
|
|
|
instances: list = []
|
|
|
|
def __init__(self, bmc=None, userid=None, password=None, port=623, **kw):
|
|
self.args = {"bmc": bmc, "userid": userid, "password": password, "port": port}
|
|
self.logged_out = False
|
|
self.powerstate = fixture("power")["powerstate"]
|
|
self.power_requests: list = []
|
|
self.ipmi_session = SimpleNamespace(logout=self._logout)
|
|
FakeCommand.instances.append(self)
|
|
|
|
def _logout(self):
|
|
self.logged_out = True
|
|
|
|
def get_power(self):
|
|
return {"powerstate": self.powerstate}
|
|
|
|
def set_power(self, powerstate, wait=False):
|
|
self.power_requests.append((powerstate, wait))
|
|
return {"pendingpowerstate": powerstate}
|
|
|
|
def get_net_configuration(self):
|
|
return fixture("net_configuration")
|
|
|
|
def get_firmware(self):
|
|
return iter(tuple(item) for item in fixture("firmware"))
|
|
|
|
def get_inventory(self):
|
|
return iter(tuple(item) for item in fixture("inventory"))
|
|
|
|
def get_sensor_data(self):
|
|
return iter(SimpleNamespace(**s) for s in fixture("sensors"))
|
|
|
|
def get_event_log(self):
|
|
return iter(fixture("event_log"))
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def fake_pyghmi(monkeypatch):
|
|
FakeCommand.instances = []
|
|
monkeypatch.setattr(ipmi, "Command", FakeCommand)
|
|
|
|
|
|
@pytest.fixture
|
|
def driver():
|
|
d = IpmiDriver("192.0.2.150", "netork", "s3cret")
|
|
d.open()
|
|
yield d
|
|
d.close()
|
|
|
|
|
|
def assert_shape(value, typed_dict):
|
|
assert isinstance(value, dict)
|
|
assert set(value) == set(typed_dict.__annotations__), typed_dict.__name__
|
|
|
|
|
|
# --- class and connection ------------------------------------------------------------------
|
|
|
|
|
|
def test_role_power_control_and_no_bmc_reset():
|
|
assert issubclass(IpmiDriver, ManagementControllerDriver)
|
|
assert primary_role_of(IpmiDriver) == "management_controller"
|
|
assert IpmiDriver.USES_SSH is False
|
|
assert IpmiDriver.DRIVER_NAME == "ipmi"
|
|
# The server's power can be switched; the BMC is not reset over IPMI.
|
|
assert callable(getattr(IpmiDriver, "set_power_state", None))
|
|
assert not hasattr(IpmiDriver, "reboot_host")
|
|
|
|
|
|
def test_entry_point():
|
|
from importlib.metadata import entry_points
|
|
|
|
(ep,) = [e for e in entry_points(group="napalm.drivers") if e.name == "ipmi"]
|
|
assert ep.load() is IpmiDriver
|
|
|
|
|
|
def test_open_logs_in_on_the_ipmi_port_and_close_logs_out():
|
|
d = IpmiDriver("192.0.2.150", "netork", "s3cret", optional_args={"port": 6230})
|
|
d.open()
|
|
(cmd,) = FakeCommand.instances
|
|
assert cmd.args == {
|
|
"bmc": "192.0.2.150",
|
|
"userid": "netork",
|
|
"password": "s3cret",
|
|
"port": 6230,
|
|
}
|
|
assert d.is_alive() == {"is_alive": True}
|
|
d.close()
|
|
assert cmd.logged_out
|
|
assert d.is_alive() == {"is_alive": False}
|
|
|
|
|
|
def test_default_port_is_623():
|
|
IpmiDriver("192.0.2.150", "u", "p").open()
|
|
assert FakeCommand.instances[0].args["port"] == 623
|
|
|
|
|
|
def test_open_failure_is_connection_exception(monkeypatch):
|
|
def refuse(**kwargs):
|
|
raise RuntimeError("Incorrect password provided")
|
|
|
|
monkeypatch.setattr(ipmi, "Command", refuse)
|
|
with pytest.raises(ConnectionException, match="Incorrect password"):
|
|
IpmiDriver("192.0.2.150", "netork", "wrong").open()
|
|
|
|
|
|
# --- NAPALM getters ------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_facts(driver):
|
|
facts = driver.get_facts()
|
|
assert facts == {
|
|
"uptime": -1.0,
|
|
"vendor": "Supermicro",
|
|
"model": "X10DRL-i",
|
|
"hostname": "192.0.2.150",
|
|
"fqdn": "192.0.2.150",
|
|
"os_version": "3.91",
|
|
"serial_number": "ZM000S000001",
|
|
"interface_list": ["bmc"],
|
|
}
|
|
|
|
|
|
def test_get_interfaces(driver):
|
|
assert driver.get_interfaces() == {
|
|
"bmc": {
|
|
"is_up": True,
|
|
"is_enabled": True,
|
|
"description": "BMC LAN (DHCP)",
|
|
"last_flapped": -1.0,
|
|
"speed": -1.0,
|
|
"mtu": -1,
|
|
"mac_address": "0c:c4:7a:00:00:01",
|
|
}
|
|
}
|
|
|
|
|
|
def test_get_interfaces_ip(driver):
|
|
assert driver.get_interfaces_ip() == {"bmc": {"ipv4": {"192.0.2.150": {"prefix_length": 24}}}}
|
|
|
|
|
|
def test_get_environment(driver):
|
|
env = driver.get_environment()
|
|
assert set(env) == {"fans", "temperature", "power", "cpu", "memory"}
|
|
assert env["cpu"] == {} and env["memory"] == {}
|
|
assert len(env["temperature"]) == 19
|
|
assert env["temperature"]["CPU1 Temp"] == {
|
|
"temperature": 34.0,
|
|
"is_alert": False,
|
|
"is_critical": False,
|
|
}
|
|
assert env["fans"] == {name: {"status": True} for name in ("FAN2", "FAN3", "FAN4", "FANB")}
|
|
# Only supplies that report themselves present.
|
|
assert env["power"] == {"PS1 Status": {"status": True, "capacity": -1.0, "output": -1.0}}
|
|
|
|
|
|
def test_sensor_health_flags(driver, monkeypatch):
|
|
readings = [
|
|
{
|
|
"name": "Hot",
|
|
"type": "Temperature",
|
|
"value": 91.0,
|
|
"units": "°C",
|
|
"health": 1,
|
|
"states": [],
|
|
"unavailable": 0,
|
|
},
|
|
{
|
|
"name": "Melting",
|
|
"type": "Temperature",
|
|
"value": 105.0,
|
|
"units": "°C",
|
|
"health": 2,
|
|
"states": [],
|
|
"unavailable": 0,
|
|
},
|
|
{
|
|
"name": "Gone",
|
|
"type": "Temperature",
|
|
"value": None,
|
|
"units": "°C",
|
|
"health": 0,
|
|
"states": [],
|
|
"unavailable": 1,
|
|
},
|
|
{
|
|
"name": "FAN1",
|
|
"type": "Fan",
|
|
"value": 0.0,
|
|
"units": "RPM",
|
|
"health": 2,
|
|
"states": [],
|
|
"unavailable": 0,
|
|
},
|
|
]
|
|
monkeypatch.setattr(
|
|
FakeCommand, "get_sensor_data", lambda self: iter(SimpleNamespace(**r) for r in readings)
|
|
)
|
|
env = driver.get_environment()
|
|
assert env["temperature"] == {
|
|
"Hot": {"temperature": 91.0, "is_alert": True, "is_critical": False},
|
|
"Melting": {"temperature": 105.0, "is_alert": True, "is_critical": True},
|
|
}
|
|
assert env["fans"] == {"FAN1": {"status": False}}
|
|
|
|
|
|
# --- role methods --------------------------------------------------------------------------
|
|
|
|
|
|
def test_get_power_state(driver):
|
|
state = driver.get_power_state()
|
|
assert_shape(state, models.PowerStateDict)
|
|
assert state == {"state": "On", "actions": ["GracefulShutdown", "ForceOff", "ForceRestart"]}
|
|
|
|
|
|
def test_a_server_that_is_off_can_only_be_switched_on(driver):
|
|
FakeCommand.instances[0].powerstate = "off"
|
|
assert driver.get_power_state() == {"state": "Off", "actions": ["On"]}
|
|
|
|
|
|
def test_every_offered_action_is_a_role_power_action(driver):
|
|
for powerstate in ("on", "off"):
|
|
FakeCommand.instances[0].powerstate = powerstate
|
|
assert set(driver.get_power_state()["actions"]) <= set(models.POWER_ACTIONS)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"action, powerstate, expected",
|
|
[
|
|
("GracefulShutdown", "on", "shutdown"),
|
|
("ForceOff", "on", "off"),
|
|
("ForceRestart", "on", "reset"),
|
|
("On", "off", "on"),
|
|
],
|
|
)
|
|
def test_set_power_state_sends_the_chassis_control(driver, action, powerstate, expected):
|
|
cmd = FakeCommand.instances[0]
|
|
cmd.powerstate = powerstate
|
|
driver.set_power_state(action)
|
|
# Returns once the BMC has accepted the request, not when the server is done.
|
|
assert cmd.power_requests == [(expected, False)]
|
|
|
|
|
|
def test_set_power_state_rejects_an_unknown_action(driver):
|
|
with pytest.raises(ValueError, match="Reboot"):
|
|
driver.set_power_state("Reboot")
|
|
assert FakeCommand.instances[0].power_requests == []
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"action, powerstate",
|
|
[
|
|
("On", "on"),
|
|
("ForceOff", "off"),
|
|
("GracefulRestart", "on"), # IPMI has no graceful restart
|
|
("PowerCycle", "on"),
|
|
],
|
|
)
|
|
def test_set_power_state_refuses_what_is_not_offered_now(driver, action, powerstate):
|
|
cmd = FakeCommand.instances[0]
|
|
cmd.powerstate = powerstate
|
|
with pytest.raises(RuntimeError, match=f"{action} not accepted"):
|
|
driver.set_power_state(action)
|
|
assert cmd.power_requests == []
|
|
|
|
|
|
def test_a_refused_power_request_is_a_runtime_error(driver):
|
|
def refuse(powerstate, wait=False):
|
|
raise ipmi.IpmiException("Insufficient privilege level")
|
|
|
|
FakeCommand.instances[0].set_power = refuse
|
|
with pytest.raises(RuntimeError, match="ForceOff refused: Insufficient privilege"):
|
|
driver.set_power_state("ForceOff")
|
|
|
|
|
|
def test_get_hardware_inventory(driver):
|
|
inv = driver.get_hardware_inventory()
|
|
assert_shape(inv, models.HardwareInventoryDict)
|
|
assert inv["manufacturer"] == "Supermicro"
|
|
assert inv["model"] == "X10DRL-i"
|
|
assert inv["serial_number"] == "ZM000S000001"
|
|
assert inv["system_uuid"] == "00000000-534d-c40c-0000-000000000001"
|
|
assert inv["power_supplies"] == [
|
|
{"slot": "PS1", "model": "", "serial_number": "", "capacity_watts": None, "status": "OK"},
|
|
{
|
|
"slot": "PS2",
|
|
"model": "",
|
|
"serial_number": "",
|
|
"capacity_watts": None,
|
|
"status": "Absent",
|
|
},
|
|
]
|
|
assert inv["cpus"] == [] and inv["memory"] == [] and inv["drives"] == [] and inv["nics"] == []
|
|
|
|
|
|
def test_get_event_log(driver):
|
|
(entry,) = driver.get_event_log()
|
|
assert_shape(entry, models.EventLogEntryDict)
|
|
assert entry == {
|
|
"id": "1",
|
|
"timestamp": 1790788051.0,
|
|
"severity": "OK",
|
|
"message": "PS2 Status: Recovered",
|
|
}
|
|
|
|
|
|
def test_event_log_newest_first_and_limited(driver, monkeypatch):
|
|
sel = [
|
|
{
|
|
"record_id": i,
|
|
"timecode": 1000 + i,
|
|
"component": "FAN1",
|
|
"event": "Lower critical",
|
|
"severity": 2,
|
|
"deassertion": False,
|
|
}
|
|
for i in range(1, 6)
|
|
]
|
|
monkeypatch.setattr(FakeCommand, "get_event_log", lambda self: iter(sel))
|
|
entries = driver.get_event_log(limit=2)
|
|
assert [e["id"] for e in entries] == ["5", "4"]
|
|
assert entries[0]["severity"] == "Critical"
|
|
assert entries[0]["message"] == "FAN1: Lower critical"
|
|
|
|
|
|
def test_get_firmware_inventory(driver):
|
|
(fw,) = driver.get_firmware_inventory()
|
|
assert_shape(fw, models.FirmwareDict)
|
|
assert fw == {"name": "BMC", "version": "3.91", "updateable": None}
|
|
|
|
|
|
def test_napalm_still_finds_redfish_by_name():
|
|
# NAPALM takes the first driver class of the package napalm_<name>.
|
|
import napalm
|
|
|
|
from napalm_redfish import RedfishDriver
|
|
|
|
assert napalm.get_network_driver("redfish") is RedfishDriver
|
|
|
|
|
|
def test_each_open_and_close_forgets_pyghmis_cached_session(monkeypatch):
|
|
# pyghmi files a session in Session.initting_sessions under the credentials as str
|
|
# and deletes it only under bytes, so it stays; the next Command for the BMC in the
|
|
# same process got the old, logged-out one back and failed with "'<' not supported
|
|
# between float and NoneType". Reproduced live against the X10, fixed by this.
|
|
cache = {}
|
|
monkeypatch.setattr(ipmi, "Session", SimpleNamespace(initting_sessions=cache))
|
|
key = ("192.0.2.150", "netork", "s3cret", 623, None)
|
|
|
|
cache[key] = "stale session of an earlier poll"
|
|
d = IpmiDriver("192.0.2.150", "netork", "s3cret")
|
|
d.open()
|
|
assert key not in cache
|
|
|
|
cache[key] = "this poll's session"
|
|
d.close()
|
|
assert key not in cache
|
|
assert FakeCommand.instances[0].logged_out
|
|
|
|
|
|
def test_both_drivers_collect_the_bmcs_own_snmp_health():
|
|
# Many BMCs run net-snmp with the UCD-MIB (the X10's answers load, memory, CPU idle and
|
|
# five interfaces), so netOrk's SNMP health poll covers the BMC itself. The managed
|
|
# server's sensors stay with get_environment.
|
|
from napalm_device_types import HealthMetricsMixin
|
|
from napalm_device_types.roles import role_keys_of
|
|
|
|
from napalm_redfish import RedfishDriver
|
|
|
|
for cls in (IpmiDriver, RedfishDriver):
|
|
assert issubclass(cls, HealthMetricsMixin), cls.__name__
|
|
assert role_keys_of(cls) == ["management_controller"]
|