napalm-vmware/tests/test_inventory.py

"""Inventory: the single seam between the drivers and the vSphere API."""

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest
from pyVmomi import vim

from napalm_vmware._inventory import Inventory


def _content(obj, props, missing=()):
    return SimpleNamespace(
        obj=obj,
        propSet=[SimpleNamespace(name=k, val=v) for k, v in props.items()],
        missingSet=[SimpleNamespace(path=p) for p in missing],
    )


@pytest.fixture
def content():
    """pyVmomi type-checks spec fields, so the objects put into them are real."""
    c = MagicMock()
    c.about = vim.AboutInfo(apiType="HostAgent", fullName="VMware ESXi 8.0.2")
    view = vim.view.ContainerView("session[1]view-1", None)
    view.Destroy = MagicMock()
    c.viewManager.CreateContainerView.return_value = view
    c.licenseManager = vim.LicenseManager("LicenseManager", None)
    return c


class TestCollect:
    def test_returns_plain_dicts_keyed_by_path_with_moref(self, content):
        pc = content.propertyCollector
        pc.RetrievePropertiesEx.return_value = SimpleNamespace(
            objects=[
                _content(
                    vim.VirtualMachine("vm-1", None),
                    {"name": "web01", "runtime.powerState": "poweredOn"},
                )
            ],
            token=None,
        )
        rows = Inventory(content).collect("VirtualMachine", ["name", "runtime.powerState"])
        assert rows == [{"_moref": "vm-1", "name": "web01", "runtime.powerState": "poweredOn"}]

    def test_follows_continuation_tokens(self, content):
        pc = content.propertyCollector
        pc.RetrievePropertiesEx.return_value = SimpleNamespace(
            objects=[_content(vim.VirtualMachine("vm-1", None), {"name": "a"})], token="t1"
        )
        pc.ContinueRetrievePropertiesEx.return_value = SimpleNamespace(
            objects=[_content(vim.VirtualMachine("vm-2", None), {"name": "b"})], token=None
        )
        rows = Inventory(content).collect("VirtualMachine", ["name"])
        assert [r["_moref"] for r in rows] == ["vm-1", "vm-2"]
        pc.ContinueRetrievePropertiesEx.assert_called_once_with(token="t1")

    def test_empty_result(self, content):
        content.propertyCollector.RetrievePropertiesEx.return_value = None
        assert Inventory(content).collect("Datastore", ["name"]) == []

    def test_missing_properties_are_simply_absent(self, content):
        """An inaccessible VM has no config; the row must still come back."""
        content.propertyCollector.RetrievePropertiesEx.return_value = SimpleNamespace(
            objects=[
                _content(vim.VirtualMachine("vm-9", None), {"name": "ghost"}, ["config.template"])
            ],
            token=None,
        )
        rows = Inventory(content).collect("VirtualMachine", ["name", "config.template"])
        assert rows == [{"_moref": "vm-9", "name": "ghost"}]

    def test_the_container_view_is_destroyed(self, content):
        content.propertyCollector.RetrievePropertiesEx.return_value = None
        Inventory(content).collect("HostSystem", ["name"])
        content.viewManager.CreateContainerView.return_value.Destroy.assert_called_once()

    def test_unknown_type_name_is_rejected(self, content):
        with pytest.raises(ValueError, match="NoSuchThing"):
            Inventory(content).collect("NoSuchThing", ["name"])


def test_about_is_plain(content):
    assert Inventory(content).about()["apiType"] == "HostAgent"


def test_licenses_are_read_from_the_license_manager(content):
    content.propertyCollector.RetrievePropertiesEx.return_value = SimpleNamespace(
        objects=[
            _content(
                vim.LicenseManager("LicenseManager", None),
                {"licenses": [vim.LicenseManager.LicenseInfo(editionKey="esxBasic", name="Free")]},
            )
        ],
        token=None,
    )
    assert Inventory(content).licenses()[0]["editionKey"] == "esxBasic"


def test_licenses_empty_without_a_license_manager(content):
    content.licenseManager = None
    assert Inventory(content).licenses() == []


def test_properties_of_a_single_object(content):
    content.propertyCollector.RetrievePropertiesEx.return_value = SimpleNamespace(
        objects=[_content(vim.Task("task-7", None), {"info.state": "success"})], token=None
    )
    task = vim.Task("task-7", None)
    assert Inventory(content).properties(task, ["info.state"]) == {
        "_moref": "task-7",
        "info.state": "success",
    }


def test_properties_of_a_vanished_object(content):
    content.propertyCollector.RetrievePropertiesEx.return_value = None
    assert Inventory(content).properties(vim.Task("task-7", None), ["info.state"]) == {}


def test_raw_properties_keep_the_pyvmomi_objects(content):
    disk = vim.vm.device.VirtualDisk(key=2000, capacityInKB=1024)
    content.propertyCollector.RetrievePropertiesEx.return_value = SimpleNamespace(
        objects=[_content(vim.VirtualMachine("vm-1", None), {"config.hardware.device": [disk]})],
        token=None,
    )
    props = Inventory(content).properties(
        vim.VirtualMachine("vm-1", None), ["config.hardware.device"], raw=True
    )
    assert props["config.hardware.device"][0] is disk