napalm-vmware/tests/test_parse_host_state.py

"""Datastores, sensors/utilisation, CDP/LLDP neighbours and raw device warnings."""

from __future__ import annotations

import pytest

from napalm_vmware.parse.environment import environment
from napalm_vmware.parse.lldp import lldp_neighbors
from napalm_vmware.parse.storage import storage_pools
from napalm_vmware.parse.warnings import host_warnings


class TestStoragePools:
    def test_from_vcsim(self, esxi_data):
        pools = storage_pools(esxi_data["objects"]["Datastore"])
        (name,) = pools
        pool = pools[name]
        assert pool["name"] == name
        assert pool["type"] == "other"
        assert pool["used"] == pool["total"] - pool["available"]
        assert pool["enabled"] is True
        assert pool["shared"] is False

    def test_shared_vmfs(self):
        ds = {
            "_moref": "datastore-1",
            "name": "san01",
            "summary": {
                "type": "VMFS",
                "capacity": 1000,
                "freeSpace": 250,
                "accessible": False,
                "multipleHostAccess": True,
            },
        }
        assert storage_pools([ds])["san01"] == {
            "name": "san01",
            "type": "vmfs",
            "total": 1000,
            "used": 750,
            "available": 250,
            "enabled": False,
            "shared": True,
        }


def _sensor(name, kind, reading, health="green", modifier=0):
    return {
        "name": name,
        "sensorType": kind,
        "currentReading": reading,
        "unitModifier": modifier,
        "healthState": {"key": health},
    }


class TestEnvironment:
    def test_utilisation_of_one_host(self, esxi_host):
        esxi_host["summary.hardware"].update(cpuMhz=2000, numCpuCores=4, memorySize=8 * 1024**3)
        esxi_host["summary.quickStats"].update(overallCpuUsage=2000, overallMemoryUsage=2048)
        env = environment([esxi_host])
        assert env["cpu"] == {"0": {"%usage": 25.0}}
        assert env["memory"] == {"available_ram": 8 * 1024**3, "used_ram": 2048 * 1024**2}

    def test_sensors(self, esxi_host):
        esxi_host["runtime.healthSystemRuntime.systemHealthInfo.numericSensorInfo"] = [
            _sensor("CPU1 Temp", "temperature", 4500, modifier=-2),
            _sensor("Inlet Temp", "temperature", 41, health="yellow"),
            _sensor("PCH Temp", "temperature", 90, health="red"),
            _sensor("FAN1", "fan", 5400),
            _sensor("FAN2", "fan", 0, health="red"),
            _sensor("PSU1 Output", "power", 180),
            _sensor("12V", "voltage", 12),
        ]
        env = environment([esxi_host])
        assert env["temperature"] == {
            "CPU1 Temp": {"temperature": 45.0, "is_alert": False, "is_critical": False},
            "Inlet Temp": {"temperature": 41.0, "is_alert": True, "is_critical": False},
            "PCH Temp": {"temperature": 90.0, "is_alert": True, "is_critical": True},
        }
        assert env["fans"] == {"FAN1": {"status": True}, "FAN2": {"status": False}}
        assert env["power"] == {"PSU1 Output": {"status": True, "capacity": -1.0, "output": 180.0}}

    def test_several_hosts_are_keyed_by_host(self, vcenter_data):
        hosts = vcenter_data["objects"]["HostSystem"]
        hosts[0]["runtime.healthSystemRuntime.systemHealthInfo.numericSensorInfo"] = [
            _sensor("Inlet Temp", "temperature", 30)
        ]
        env = environment(hosts)
        assert set(env["cpu"]) == {h["name"] for h in hosts}
        assert f"{hosts[0]['name']}/Inlet Temp" in env["temperature"]
        assert env["memory"]["available_ram"] == sum(
            h["summary.hardware"]["memorySize"] for h in hosts
        )

    def test_host_without_stats(self):
        env = environment([{"_moref": "host-1", "name": "h"}])
        assert env["cpu"] == {"0": {"%usage": 0.0}}
        assert env["memory"] == {"available_ram": 0, "used_ram": 0}


class TestLldpNeighbors:
    def test_lldp_preferred_over_cdp(self):
        hints = [
            {
                "device": "vmnic0",
                "connectedSwitchPort": {"devId": "cdp-sw", "portId": "Gi1/0/1"},
                "lldpInfo": {
                    "chassisId": "00:11:22:33:44:55",
                    "portId": "1/1/7",
                    "parameter": [{"key": "System Name", "value": "core-sw"}],
                },
            },
            {"device": "vmnic1", "connectedSwitchPort": {"devId": "cdp-sw", "portId": "Gi1/0/2"}},
            {"device": "vmnic2", "lldpInfo": {"chassisId": "aa:bb", "portId": "5"}},
            {"device": "vmnic3"},
        ]
        assert lldp_neighbors(hints) == {
            "vmnic0": [{"hostname": "core-sw", "port": "1/1/7"}],
            "vmnic1": [{"hostname": "cdp-sw", "port": "Gi1/0/2"}],
            "vmnic2": [{"hostname": "aa:bb", "port": "5"}],
        }

    def test_nothing_heard(self):
        assert lldp_neighbors([]) == {}


class TestHostWarnings:
    def test_healthy_vcsim_host(self, esxi_host):
        assert host_warnings([esxi_host], [], standalone=True) == []

    def test_maintenance_mode(self, esxi_host):
        esxi_host["runtime.inMaintenanceMode"] = True
        assert host_warnings([esxi_host], [], standalone=True) == [
            {"code": "vmware_host_maintenance_mode", "meta": {"host": "localhost.localdomain"}}
        ]

    @pytest.mark.parametrize("state", ["disconnected", "notResponding"])
    def test_disconnected_host_under_vcenter(self, esxi_host, state):
        esxi_host["runtime.connectionState"] = state
        assert host_warnings([esxi_host], [], standalone=False) == [
            {
                "code": "vmware_host_disconnected",
                "meta": {"host": "localhost.localdomain", "state": state},
            }
        ]

    def test_config_issues(self, esxi_host):
        esxi_host["configIssue"] = [
            {"fullFormattedMessage": "SSH for the host has been enabled"},
            {"fullFormattedMessage": ""},
        ]
        assert host_warnings([esxi_host], [], standalone=True) == [
            {
                "code": "vmware_host_config_issue",
                "meta": {
                    "host": "localhost.localdomain",
                    "issues": ["SSH for the host has been enabled"],
                },
            }
        ]

    def test_standalone_driver_on_a_vcenter_managed_host(self, esxi_host):
        esxi_host["summary.managementServerIp"] = "192.0.2.5"
        assert host_warnings([esxi_host], [], standalone=True) == [
            {"code": "vmware_host_managed_by_vcenter", "meta": {"vcenter": "192.0.2.5"}}
        ]

    def test_vcenter_driver_does_not_report_its_own_management(self, esxi_host):
        esxi_host["summary.managementServerIp"] = "192.0.2.5"
        assert host_warnings([esxi_host], [], standalone=False) == []

    def test_free_license_makes_the_api_read_only(self, esxi_host):
        licenses = [{"editionKey": "esxBasic", "name": "VMware vSphere 8 Hypervisor"}]
        assert host_warnings([esxi_host], licenses, standalone=True) == [
            {
                "code": "vmware_api_read_only",
                "meta": {"edition": "VMware vSphere 8 Hypervisor"},
            }
        ]

    def test_evaluation_license_is_not_read_only(self, esxi_host):
        licenses = [{"editionKey": "eval", "name": "Evaluation Mode"}]
        assert host_warnings([esxi_host], licenses, standalone=True) == []