napalm-redfish/tests/test_ipmi.py

"""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
is read-only: it neither switches the server's power nor resets 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, keepalive=True, **kw):
        self.args = {"bmc": bmc, "userid": userid, "password": password, "port": port}
        self.logged_out = False
        self.ipmi_session = SimpleNamespace(logout=self._logout)
        FakeCommand.instances.append(self)

    def _logout(self):
        self.logged_out = True

    def get_power(self):
        return fixture("power")

    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_and_read_only():
    assert issubclass(IpmiDriver, ManagementControllerDriver)
    assert primary_role_of(IpmiDriver) == "management_controller"
    assert IpmiDriver.USES_SSH is False
    assert IpmiDriver.DRIVER_NAME == "ipmi"
    # Read-only: no power control, no BMC reset over IPMI.
    assert not hasattr(IpmiDriver, "set_power_state")
    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": []}


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