napalm-vmware/tests/test_parse_host_state.py

"""Datastores, sensors/utilisation, CDP/LLDP neighbours, host and license facts."""

from __future__ import annotations

import pytest

from napalm_vmware.parse.environment import environment
from napalm_vmware.parse.hosts import hypervisor_hosts
from napalm_vmware.parse.licenses import license_list
from napalm_vmware.parse.lldp import lldp_neighbors
from napalm_vmware.parse.storage import storage_pools


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 TestHypervisorHosts:
    """Facts only (netOrk #1111): which of them is a warning is the caller's call."""

    def test_healthy_vcsim_host(self, esxi_host):
        assert hypervisor_hosts([esxi_host], standalone=True) == [
            {
                "name": "localhost.localdomain",
                "connected": None,
                "maintenance": False,
                "config_issues": [],
                "managed_by": None,
            }
        ]

    def test_maintenance_mode(self, esxi_host):
        esxi_host["runtime.inMaintenanceMode"] = True
        assert hypervisor_hosts([esxi_host], standalone=True)[0]["maintenance"] is True

    @pytest.mark.parametrize("state", ["disconnected", "notResponding"])
    def test_disconnected_host_under_vcenter(self, esxi_host, state):
        esxi_host["runtime.connectionState"] = state
        assert hypervisor_hosts([esxi_host], standalone=False)[0]["connected"] is False

    def test_connected_host_under_vcenter(self, esxi_host):
        assert hypervisor_hosts([esxi_host], standalone=False)[0]["connected"] is True

    def test_a_standalone_host_cannot_tell(self, esxi_host):
        """A disconnected host could not have answered the ESXi driver."""
        esxi_host["runtime.connectionState"] = "disconnected"
        assert hypervisor_hosts([esxi_host], standalone=True)[0]["connected"] is None

    def test_config_issues(self, esxi_host):
        esxi_host["configIssue"] = [
            {"fullFormattedMessage": "SSH for the host has been enabled"},
            {"fullFormattedMessage": ""},
        ]
        assert hypervisor_hosts([esxi_host], standalone=True)[0]["config_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 hypervisor_hosts([esxi_host], standalone=True)[0]["managed_by"] == "192.0.2.5"

    @pytest.mark.parametrize("ip", ["", "0.0.0.0", None])
    def test_unmanaged(self, esxi_host, ip):
        esxi_host["summary.managementServerIp"] = ip
        assert hypervisor_hosts([esxi_host], standalone=True)[0]["managed_by"] is None

    def test_vcenter_driver_does_not_report_its_own_management(self, esxi_host):
        esxi_host["summary.managementServerIp"] = "192.0.2.5"
        assert hypervisor_hosts([esxi_host], standalone=False)[0]["managed_by"] is None


class TestLicenseList:
    def test_the_edition_key_is_the_edition(self):
        licenses = [
            {
                "editionKey": "esxBasic",
                "name": "VMware vSphere 8 Hypervisor",
                "licenseKey": "AAAAA-BBBBB-CCCCC-DDDDD-EEEEE",
            }
        ]
        assert license_list(licenses) == [
            {"name": "VMware vSphere 8 Hypervisor", "status": "active", "edition": "esxBasic"}
        ]

    def test_the_key_is_never_reported(self):
        licenses = [{"editionKey": "eval", "name": "Evaluation Mode", "licenseKey": "00000"}]
        assert "00000" not in str(license_list(licenses))

    def test_none(self):
        assert license_list([]) == []