Christian Manivong
06b446934f
- zip() in provisioning is strict (B905): the OVF asks for one adapter per requested NIC, so the two lists match. A VM that came back with fewer adapters used to get a seed that left a NIC unconfigured; it now fails, and create_vm_from_cloud_init discards it as on any other error after the import. A test covers it. - A failed abort of the import lease is logged at debug level instead of passed over (B110); the original error is still the one raised. - nosec for false positives, each with its reason: the 0.0.0.0 entries in the parsers are addresses vSphere reports, not binds (B104); qemu-img runs with a fixed argv and no shell (B404, B603); xml.sax.saxutils only escapes the OVF descriptor the driver writes, nothing is parsed (B406), so defusedxml would not apply.
308 lines
12 KiB
Python
308 lines
12 KiB
Python
"""The HypervisorDriver provisioning methods, around the live import (#307)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pycdlib
|
|
import pytest
|
|
import yaml
|
|
|
|
from napalm_vmware import VmwareEsxiDriver
|
|
from napalm_vmware import provisioning as prov
|
|
from napalm_vmware.provision.placement import Placement
|
|
from tests.conftest import FakeInventory
|
|
|
|
PLACE = Placement(
|
|
host="ha-host",
|
|
host_name="esx01",
|
|
datastore="datastore-1",
|
|
datastore_name="LocalDS_0",
|
|
resource_pool="ha-root-pool",
|
|
folder="ha-folder-vm",
|
|
datacenter_name="ha-datacenter",
|
|
networks=["HaNetwork-VM Network"],
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def driver(esxi_data):
|
|
d = VmwareEsxiDriver("esx01", "root", "secret", optional_args={"image_cache_dir": "/tmp/x"})
|
|
d._si = MagicMock()
|
|
d._inventory = FakeInventory(esxi_data)
|
|
d.mo = MagicMock(name="mo")
|
|
d._mo = MagicMock(return_value=d.mo)
|
|
return d
|
|
|
|
|
|
class TestTargets:
|
|
def test_image_storages(self, driver):
|
|
(ds,) = driver.get_image_storages()
|
|
assert ds["name"] == "LocalDS_0"
|
|
assert ds["available_gb"] <= ds["total_gb"]
|
|
|
|
def test_inaccessible_datastore_is_not_offered(self, driver, esxi_data):
|
|
esxi_data["objects"]["Datastore"][0]["summary"]["accessible"] = False
|
|
assert driver.get_image_storages() == []
|
|
|
|
def test_network_targets_are_port_groups_with_a_fixed_vlan(self, driver, esxi_host):
|
|
esxi_host["config.network.portgroup"][0]["spec"]["vlanId"] = 30
|
|
targets = {t["name"]: t for t in driver.get_network_targets()}
|
|
assert targets["VM Network"] == {
|
|
"name": "VM Network",
|
|
"kind": "portgroup",
|
|
"vlan_aware": False,
|
|
"fixed_vlan_tag": 30,
|
|
}
|
|
assert targets["Management Network"]["fixed_vlan_tag"] is None
|
|
|
|
|
|
class TestCheckNics:
|
|
@pytest.mark.parametrize(
|
|
("nics", "message"),
|
|
[
|
|
([], "at least one"),
|
|
([{"bridge": "Nope"}], "no port group"),
|
|
([{"bridge": "VM Network", "trunk_vlan_tags": [10, 20]}], "trunk"),
|
|
([{"bridge": "VM Network", "vlan_tag": 40}], "carries VLAN"),
|
|
],
|
|
)
|
|
def test_refusals(self, driver, nics, message):
|
|
with pytest.raises(RuntimeError, match=message):
|
|
driver._check_nics(nics)
|
|
|
|
def test_matching_vlan_is_fine(self, driver, esxi_host):
|
|
esxi_host["config.network.portgroup"][0]["spec"]["vlanId"] = 40
|
|
driver._check_nics([{"bridge": "VM Network", "vlan_tag": 40}])
|
|
|
|
|
|
def _create(driver, **overrides):
|
|
kwargs = dict(
|
|
image_url="https://img/debian.qcow2",
|
|
cpu=2,
|
|
memory=2048,
|
|
nics=[{"bridge": "VM Network", "dhcp": True}],
|
|
cloud_init_config={"hostname": "web01"},
|
|
)
|
|
kwargs.update(overrides)
|
|
return driver.create_vm_from_cloud_init("web01", **kwargs)
|
|
|
|
|
|
@pytest.fixture
|
|
def staged(driver):
|
|
"""Everything around the live import replaced; returns the mocks."""
|
|
cache = MagicMock()
|
|
cache.return_value.vmdk.return_value = ("/tmp/x/a.vmdk", 2 * 1024**3)
|
|
with (
|
|
patch.object(prov, "ImageCache", cache),
|
|
patch.object(prov, "choose_placement", return_value=PLACE),
|
|
patch.object(driver, "_placement_inventory"),
|
|
patch.object(driver, "_import_ovf", return_value="vm-99") as import_ovf,
|
|
patch.object(driver, "_finish_vm") as finish,
|
|
patch.object(driver, "_run_task") as run_task,
|
|
patch.object(driver, "_discard") as discard,
|
|
):
|
|
driver._inventory.properties = MagicMock(return_value={"config.instanceUuid": "uuid-99"})
|
|
yield MagicMock(
|
|
cache=cache, import_ovf=import_ovf, finish=finish, run_task=run_task, discard=discard
|
|
)
|
|
|
|
|
|
class TestCreate:
|
|
def test_result_names_vm_and_host(self, driver, staged):
|
|
assert _create(driver) == {"vmid": "uuid-99", "name": "web01", "node": "esx01"}
|
|
|
|
def test_image_cache_is_used_with_the_download_timeout(self, driver, staged):
|
|
_create(driver, image_checksum="sha256:ab", download_timeout=42)
|
|
staged.cache.return_value.vmdk.assert_called_once_with(
|
|
"https://img/debian.qcow2", "sha256:ab", 42
|
|
)
|
|
|
|
def test_descriptor_describes_the_request(self, driver, staged):
|
|
_create(driver, nics=[{"bridge": "VM Network"}, {"bridge": "Management Network"}])
|
|
descriptor = staged.import_ovf.call_args.args[1]
|
|
assert "<rasd:VirtualQuantity>2048</rasd:VirtualQuantity>" in descriptor
|
|
assert descriptor.count("<rasd:ResourceType>10</rasd:ResourceType>") == 2
|
|
|
|
def test_ssh_keys_reach_the_user_data(self, driver, staged):
|
|
_create(driver, ssh_public_keys=["ssh-ed25519 AAA k"])
|
|
user = staged.finish.call_args.args[4]
|
|
assert yaml.safe_load(user)["ssh_authorized_keys"] == ["ssh-ed25519 AAA k"]
|
|
|
|
def test_powered_on_last(self, driver, staged):
|
|
_create(driver)
|
|
driver.mo.PowerOnVM_Task.assert_not_called() # handed to _run_task, not called
|
|
assert staged.run_task.call_args.args[0] is driver.mo.PowerOnVM_Task
|
|
|
|
def test_failure_after_import_removes_the_vm(self, driver, staged):
|
|
staged.finish.side_effect = OSError("upload refused")
|
|
with pytest.raises(RuntimeError, match="upload refused"):
|
|
_create(driver)
|
|
staged.discard.assert_called_once_with("vm-99")
|
|
|
|
def test_no_cpu_type_is_the_normal_case(self, driver, staged):
|
|
assert _create(driver, cpu_type=None)["vmid"] == "uuid-99"
|
|
|
|
def test_a_cpu_type_is_refused_before_anything_happens(self, driver, staged):
|
|
"""VMware sets CPU compatibility per cluster (EVC), not per VM, so a
|
|
requested model cannot be honoured -- and is not silently dropped."""
|
|
with pytest.raises(ValueError, match="CPU type"):
|
|
_create(driver, cpu_type="host")
|
|
staged.cache.return_value.vmdk.assert_not_called()
|
|
staged.import_ovf.assert_not_called()
|
|
|
|
@pytest.mark.parametrize("guest_os", ["freebsd", "openbsd"])
|
|
def test_a_guest_other_than_linux_is_refused_before_anything_happens(
|
|
self, driver, staged, guest_os
|
|
):
|
|
with pytest.raises(ValueError, match=guest_os):
|
|
_create(driver, guest_os=guest_os)
|
|
staged.cache.return_value.vmdk.assert_not_called()
|
|
staged.import_ovf.assert_not_called()
|
|
|
|
def test_linux_is_the_default_guest(self, driver, staged):
|
|
assert _create(driver, guest_os="linux")["vmid"] == "uuid-99"
|
|
|
|
def test_placement_refusal_is_a_runtime_error(self, driver, staged):
|
|
with patch.object(prov, "choose_placement", side_effect=ValueError("no host")):
|
|
with pytest.raises(RuntimeError, match="no host"):
|
|
_create(driver)
|
|
staged.import_ovf.assert_not_called()
|
|
|
|
|
|
MAC = "00:0c:29:39:31:62"
|
|
|
|
|
|
def _live_devices():
|
|
"""What vSphere hands back: full device objects, backings included."""
|
|
from pyVmomi import vim
|
|
|
|
disk = vim.vm.device.VirtualDisk(
|
|
key=2000,
|
|
controllerKey=1000,
|
|
unitNumber=0,
|
|
capacityInKB=2 * 1024 * 1024,
|
|
backing=vim.vm.device.VirtualDisk.FlatVer2BackingInfo(
|
|
fileName="[LocalDS_0] web01/disk.vmdk", diskMode="persistent"
|
|
),
|
|
deviceInfo=vim.Description(label="Hard disk 1", summary=""),
|
|
)
|
|
nic = vim.vm.device.VirtualVmxnet3(
|
|
key=4000,
|
|
macAddress=MAC,
|
|
addressType="generated",
|
|
backing=vim.vm.device.VirtualEthernetCard.NetworkBackingInfo(deviceName="VM Network"),
|
|
deviceInfo=vim.Description(label="Network adapter 1", summary=""),
|
|
)
|
|
return [disk, nic]
|
|
|
|
|
|
class TestFinishVm:
|
|
def _finish(self, driver, esxi_data, nics, disk_gb=None):
|
|
driver._inventory.properties = MagicMock(
|
|
return_value={
|
|
"config.hardware.device": _live_devices(),
|
|
"config.files.vmPathName": "[LocalDS_0] web01/web01.vmx",
|
|
}
|
|
)
|
|
with (
|
|
patch.object(driver, "_upload_seed") as upload,
|
|
patch.object(driver, "_run_task") as run,
|
|
):
|
|
driver._finish_vm(
|
|
"vm-99",
|
|
"web01",
|
|
PLACE,
|
|
nics,
|
|
"#cloud-config\n{}\n",
|
|
disk_gb=disk_gb,
|
|
image_size=2 * 1024**3,
|
|
)
|
|
return upload, run.call_args.kwargs["spec"]
|
|
|
|
def test_seed_goes_next_to_the_vm_and_is_attached(self, driver, esxi_data):
|
|
upload, spec = self._finish(driver, esxi_data, [{"bridge": "VM Network"}])
|
|
assert upload.call_args.args[1] == "web01/cidata.iso"
|
|
cdrom = next(
|
|
c.device for c in spec.deviceChange if type(c.device).__name__.endswith("VirtualCdrom")
|
|
)
|
|
assert cdrom.backing.fileName == "[LocalDS_0] web01/cidata.iso"
|
|
|
|
def test_seed_network_config_uses_the_real_mac(self, driver, esxi_data):
|
|
upload, _ = self._finish(driver, esxi_data, [{"bridge": "VM Network"}])
|
|
mac = MAC
|
|
reader = pycdlib.PyCdlib()
|
|
reader.open_fp(io.BytesIO(upload.call_args.args[2]))
|
|
out = io.BytesIO()
|
|
reader.get_file_from_iso_fp(out, rr_path="/network-config")
|
|
reader.close()
|
|
cfg = yaml.safe_load(out.getvalue())
|
|
assert cfg["ethernets"]["nic0"] == {"match": {"macaddress": mac.lower()}, "dhcp4": True}
|
|
|
|
def test_requested_mac_is_pinned_on_the_whole_device(self, driver, esxi_data):
|
|
_, spec = self._finish(
|
|
driver, esxi_data, [{"bridge": "VM Network", "mac": "00:50:56:01:02:03"}]
|
|
)
|
|
nic = next(c.device for c in spec.deviceChange if c.operation == "edit")
|
|
assert (nic.addressType, nic.macAddress) == ("manual", "00:50:56:01:02:03")
|
|
assert nic.backing.deviceName == "VM Network" # an edit replaces the device as sent
|
|
|
|
def test_disk_grows_only_when_asked_for_more(self, driver, esxi_data):
|
|
_, spec = self._finish(driver, esxi_data, [{"bridge": "VM Network"}], disk_gb=20)
|
|
disk = next(
|
|
c.device for c in spec.deviceChange if type(c.device).__name__.endswith("VirtualDisk")
|
|
)
|
|
assert disk.capacityInKB == 20 * 1024 * 1024
|
|
assert disk.backing.fileName == "[LocalDS_0] web01/disk.vmdk" # vcsim panicked without it
|
|
_, spec = self._finish(driver, esxi_data, [{"bridge": "VM Network"}], disk_gb=1)
|
|
assert not any(type(c.device).__name__.endswith("VirtualDisk") for c in spec.deviceChange)
|
|
|
|
def test_a_vm_without_an_adapter_per_requested_nic_is_refused(self, driver, esxi_data):
|
|
# The OVF asks for one adapter per NIC; a VM that came back with fewer would
|
|
# get a seed that leaves a NIC unconfigured. create_vm_from_cloud_init
|
|
# discards it on any error from here.
|
|
with pytest.raises(ValueError):
|
|
self._finish(driver, esxi_data, [{"bridge": "VM Network"}, {"bridge": "VM Network"}])
|
|
|
|
|
|
class TestDestroyAndStatus:
|
|
def test_destroy_powers_off_first(self, driver, esxi_data):
|
|
vm = esxi_data["objects"]["VirtualMachine"][0]
|
|
with patch.object(driver, "_run_task") as run:
|
|
driver.destroy_vm(vm["config.instanceUuid"])
|
|
assert [c.args[0] for c in run.call_args_list] == [
|
|
driver.mo.PowerOffVM_Task,
|
|
driver.mo.Destroy_Task,
|
|
]
|
|
|
|
def test_keep_disks_unregisters(self, driver, esxi_data):
|
|
vm = esxi_data["objects"]["VirtualMachine"][0]
|
|
vm["runtime.powerState"] = "poweredOff"
|
|
driver.destroy_vm(vm["config.instanceUuid"], remove_disk=False)
|
|
driver.mo.UnregisterVM.assert_called_once_with()
|
|
|
|
def test_destroy_unknown(self, driver):
|
|
with pytest.raises(RuntimeError):
|
|
driver.destroy_vm("nope")
|
|
|
|
def test_status(self, driver, esxi_data):
|
|
vm = esxi_data["objects"]["VirtualMachine"][0]
|
|
vm["guest.net"][0]["ipAddress"] = ["192.0.2.7"]
|
|
vm["guest.hostName"] = "web01"
|
|
status = driver.get_vm_status(vm["config.instanceUuid"])
|
|
assert status["status"] == "running"
|
|
assert status["ip_address"] == "192.0.2.7"
|
|
assert status["hostname"] == "web01"
|
|
assert status["mac_address"]
|
|
|
|
def test_waiting_for_an_ip_times_out(self, driver, esxi_data):
|
|
vm = esxi_data["objects"]["VirtualMachine"][0]
|
|
clock = iter([0.0, 1.0, 400.0])
|
|
with (
|
|
patch.object(prov.time, "monotonic", lambda: next(clock)),
|
|
patch.object(prov.time, "sleep"),
|
|
):
|
|
with pytest.raises(RuntimeError, match="no IP"):
|
|
driver.get_vm_status(vm["config.instanceUuid"], wait_for_ip=True, timeout=300)
|