napalm-vmware/tests/test_vcsim.py

"""End-to-end against govmomi's vcsim: real sessions, real tasks, no mocks.

Skipped unless the simulators are running::

    docker run -d --rm -p 127.0.0.1:8989:8989 vmware/vcsim -l 0.0.0.0:8989
    docker run -d --rm -p 127.0.0.1:8990:8989 vmware/vcsim -esx -l 0.0.0.0:8989
    VCSIM_VCENTER_PORT=8989 VCSIM_ESXI_PORT=8990 pytest -m vcsim

vcsim accepts any credentials. It is a simulator: this proves the drivers speak
the vSphere API correctly, not that they understand a particular real host.
"""

from __future__ import annotations

import os

import pytest
from napalm.base.exceptions import ConnectionException

from napalm_vmware import VmwareEsxiDriver, VmwareVcenterDriver

pytestmark = pytest.mark.vcsim


def _driver(cls, env):
    port = os.environ.get(env)
    if not port:
        pytest.skip(f"{env} not set; vcsim not running")
    driver = cls("127.0.0.1", "user", "pass", optional_args={"port": port, "verify_ssl": False})
    driver.open()
    yield driver
    driver.close()


@pytest.fixture
def esxi():
    yield from _driver(VmwareEsxiDriver, "VCSIM_ESXI_PORT")


@pytest.fixture
def vcenter():
    yield from _driver(VmwareVcenterDriver, "VCSIM_VCENTER_PORT")


def test_esxi_reads(esxi):
    assert esxi.is_alive() == {"is_alive": True}
    assert esxi.get_facts()["os_version"].startswith("VMware ESXi")
    assert "vmnic0" in esxi.get_interfaces()
    assert esxi.get_vms()
    assert esxi.get_vm_storage_pools()
    assert esxi.get_virtual_networks()
    assert "cpu" in esxi.get_environment()
    assert isinstance(esxi.get_hypervisor_hosts(), list)
    assert isinstance(esxi.get_licenses(), list)
    assert isinstance(esxi.get_lldp_neighbors(), dict)


def test_vcenter_reads(vcenter):
    assert vcenter.get_facts()["model"] == "VMware vCenter Server"
    vms = vcenter.get_vms()
    assert len({vm["node"] for vm in vms}) > 1
    assert vcenter.get_vm_config(vms[0]["vmid"])["vmid"] == vms[0]["vmid"]


def test_power_cycle(esxi):
    vm = esxi.get_vms()[0]
    esxi.stop_vm(vm["vmid"], force=True)
    assert _status(esxi, vm) == "stopped"
    with pytest.raises(RuntimeError, match="not running"):
        esxi.suspend_vm(vm["vmid"])
    esxi.start_vm(vm["vmid"])
    assert _status(esxi, vm) == "running"
    esxi.suspend_vm(vm["vmid"])
    assert _status(esxi, vm) == "suspended"
    esxi.start_vm(vm["vmid"])
    esxi.reboot_vm(vm["vmid"], force=True)
    assert _status(esxi, vm) == "running"


def test_snapshot_lifecycle(vcenter):
    vm = vcenter.get_vms()[0]
    vcenter.create_vm_snapshot(vm["vmid"], "netork-test", description="from pytest")
    try:
        (snap,) = [s for s in vcenter.get_vm_snapshots(vm["vmid"]) if s["name"] == "netork-test"]
        assert snap["description"] == "from pytest"
        with pytest.raises(ValueError, match="already"):
            vcenter.create_vm_snapshot(vm["vmid"], "netork-test")
        vcenter.rollback_vm_snapshot(vm["vmid"], "netork-test")
    finally:
        vcenter.delete_vm_snapshot(vm["vmid"], "netork-test")
    assert not [s for s in vcenter.get_vm_snapshots(vm["vmid"]) if s["name"] == "netork-test"]


def test_wrong_driver_is_refused():
    port = os.environ.get("VCSIM_VCENTER_PORT")
    if not port:
        pytest.skip("VCSIM_VCENTER_PORT not set")
    driver = VmwareEsxiDriver(
        "127.0.0.1", "u", "p", optional_args={"port": port, "verify_ssl": False}
    )
    with pytest.raises(ConnectionException, match="vmware_vcenter"):
        driver.open()


def _status(driver, vm):
    return next(v["status"] for v in driver.get_vms() if v["vmid"] == vm["vmid"])


@pytest.fixture
def image_url(tmp_path):
    """A tiny qcow2 served over HTTP, like a catalog image."""
    import functools
    import http.server
    import shutil
    import subprocess
    import threading

    if not shutil.which("qemu-img"):
        pytest.skip("qemu-img not installed")
    subprocess.run(
        ["qemu-img", "create", "-q", "-f", "qcow2", str(tmp_path / "tiny.qcow2"), "64M"], check=True
    )
    handler = functools.partial(http.server.SimpleHTTPRequestHandler, directory=str(tmp_path))
    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{server.server_address[1]}/tiny.qcow2"
    server.shutdown()


@pytest.mark.parametrize("which", ["esxi", "vcenter"])
def test_provision_and_destroy(which, image_url, tmp_path, request):
    driver = request.getfixturevalue(which)
    driver._image_cache_dir = tmp_path / "cache"
    result = driver.create_vm_from_cloud_init(
        "netork-e2e",
        image_url=image_url,
        cpu=2,
        memory=1024,
        nics=[{"bridge": "VM Network", "dhcp": True, "mac": "00:50:56:01:02:03"}],
        cloud_init_config={"hostname": "netork-e2e"},
        ssh_public_keys=["ssh-ed25519 AAAA test"],
        disk_resize_gb=1,
    )
    try:
        assert driver.get_vm_status(result["vmid"])["status"] == "running"
        config = driver.get_vm_config(result["vmid"])
        assert (config["vcpus"], config["memory"]) == (2, 1024)
        assert config["disks"][0]["size"] == 1
        assert config["nics"][0]["mac"] == "00:50:56:01:02:03"
    finally:
        driver.destroy_vm(result["vmid"])
    assert not any(vm["vmid"] == result["vmid"] for vm in driver.get_vms())