napalm-vmware/tests/test_actions.py

"""Power and snapshot actions: which vSphere call each one makes, and refusals."""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest

from napalm_vmware import VmwareEsxiDriver
from tests.conftest import FakeInventory


@pytest.fixture
def vm0(esxi_data):
    return esxi_data["objects"]["VirtualMachine"][0]


@pytest.fixture
def driver(esxi_data):
    d = VmwareEsxiDriver("esx01", "root", "secret")
    d._si = MagicMock()
    d._inventory = FakeInventory(esxi_data)
    d.mo = MagicMock(name="vm")
    d._mo = MagicMock(return_value=d.mo)
    return d


@pytest.fixture
def waited():
    with patch("napalm_vmware.actions.wait_for_task") as wait:
        yield wait


class TestPower:
    def test_start(self, driver, vm0, waited):
        vm0["runtime.powerState"] = "poweredOff"
        driver.start_vm(vm0["name"])
        driver.mo.PowerOnVM_Task.assert_called_once_with()
        assert waited.call_args.args[1] is driver.mo.PowerOnVM_Task.return_value

    def test_forced_stop_powers_off(self, driver, vm0, waited):
        driver.stop_vm(vm0["name"], force=True)
        driver.mo.PowerOffVM_Task.assert_called_once_with()

    def test_graceful_stop_asks_tools_and_waits_for_power_off(self, driver, vm0):
        vm0["guest.toolsRunningStatus"] = "guestToolsRunning"
        with patch("napalm_vmware.actions.wait_until") as until:
            driver.stop_vm(vm0["name"])
        driver.mo.ShutdownGuest.assert_called_once_with()
        driver._inventory.data["objects"]["VirtualMachine"][0]["runtime.powerState"] = "poweredOff"
        driver._inventory.properties = lambda obj, paths: {"runtime.powerState": "poweredOff"}
        check = until.call_args.args[0]
        assert check() is True

    def test_graceful_stop_without_tools_is_refused(self, driver, vm0):
        with pytest.raises(RuntimeError, match="VMware Tools"):
            driver.stop_vm(vm0["name"])
        driver.mo.ShutdownGuest.assert_not_called()

    def test_stopping_a_stopped_vm_is_refused(self, driver, vm0):
        vm0["runtime.powerState"] = "poweredOff"
        with pytest.raises(RuntimeError, match="not running"):
            driver.stop_vm(vm0["name"], force=True)

    def test_graceful_reboot(self, driver, vm0):
        vm0["guest.toolsRunningStatus"] = "guestToolsRunning"
        driver.reboot_vm(vm0["name"])
        driver.mo.RebootGuest.assert_called_once_with()

    def test_forced_reboot_resets(self, driver, vm0, waited):
        driver.reboot_vm(vm0["name"], force=True)
        driver.mo.ResetVM_Task.assert_called_once_with()

    def test_suspend(self, driver, vm0, waited):
        driver.suspend_vm(vm0["name"])
        driver.mo.SuspendVM_Task.assert_called_once_with()

    def test_suspend_requires_a_running_vm(self, driver, vm0):
        vm0["runtime.powerState"] = "suspended"
        with pytest.raises(RuntimeError, match="not running"):
            driver.suspend_vm(vm0["name"])

    def test_free_license_refusal_is_explained(self, driver, vm0):
        from pyVmomi import vim

        vm0["runtime.powerState"] = "poweredOff"
        driver.mo.PowerOnVM_Task.side_effect = vim.fault.RestrictedVersion(msg="prohibited")
        with pytest.raises(RuntimeError, match="free vSphere Hypervisor"):
            driver.start_vm(vm0["name"])

    def test_unknown_vm(self, driver):
        with pytest.raises(ValueError):
            driver.start_vm("nope")


def _with_snapshot(vm, name="before-upgrade", moref="snapshot-1"):
    vm["snapshot"] = {
        "rootSnapshotList": [
            {"name": name, "snapshot": moref, "createTime": 1.0, "state": "poweredOff"}
        ]
    }


class TestSnapshots:
    def test_create(self, driver, vm0, waited):
        driver.create_vm_snapshot(vm0["name"], "pre", description="d", include_memory=True)
        driver.mo.CreateSnapshot_Task.assert_called_once_with(
            name="pre", description="d", memory=True, quiesce=False
        )

    def test_memory_is_only_captured_from_a_running_vm(self, driver, vm0, waited):
        vm0["runtime.powerState"] = "poweredOff"
        driver.create_vm_snapshot(vm0["name"], "pre", include_memory=True)
        assert driver.mo.CreateSnapshot_Task.call_args.kwargs["memory"] is False

    def test_create_refuses_a_duplicate_name(self, driver, vm0):
        _with_snapshot(vm0, "pre")
        with pytest.raises(ValueError, match="already"):
            driver.create_vm_snapshot(vm0["name"], "pre")

    def test_delete_keeps_the_children(self, driver, vm0, waited):
        _with_snapshot(vm0)
        driver.delete_vm_snapshot(vm0["name"], "before-upgrade")
        assert driver._mo.call_args.args[1] == "snapshot-1"
        driver.mo.RemoveSnapshot_Task.assert_called_once_with(removeChildren=False)

    def test_rollback(self, driver, vm0, waited):
        _with_snapshot(vm0)
        driver.rollback_vm_snapshot(vm0["name"], "before-upgrade")
        driver.mo.RevertToSnapshot_Task.assert_called_once_with()

    def test_unknown_snapshot(self, driver, vm0):
        with pytest.raises(ValueError, match="no snapshot"):
            driver.rollback_vm_snapshot(vm0["name"], "nope")


class TestRebootHost:
    def test_esxi_in_maintenance_mode_is_rebooted(self, driver, esxi_host):
        esxi_host["runtime.inMaintenanceMode"] = True
        driver.reboot_host()
        driver.mo.RebootHost_Task.assert_called_once_with(force=False)

    def test_refused_outside_maintenance_mode(self, driver):
        """force=True would power off running VMs without asking them."""
        with pytest.raises(RuntimeError, match="maintenance mode"):
            driver.reboot_host()
        driver.mo.RebootHost_Task.assert_not_called()

    def test_a_vcenter_cannot_reboot_itself(self):
        from napalm_vmware import VmwareVcenterDriver

        assert not hasattr(VmwareVcenterDriver, "reboot_host")