feat(provisioning): accept cpu_type, and refuse one that cannot be honoured

napalm_device_types' create_vm_from_cloud_init gained cpu_type, the CPU model
a new VM gets (netork#494). VMware sets CPU compatibility per cluster (EVC),
not per VM, and this driver has no get_vm_cpu_types() to offer a model, so
None is the only value it can honour. Anything else raises ValueError before
the image is fetched rather than being dropped silently.
This commit is contained in:
Christian Manivong
2026-10-04 12:02:48 +02:00
parent 25bad821ab
commit 4b4410741f
2 changed files with 16 additions and 0 deletions
+5
View File
@@ -147,9 +147,14 @@ class VmwareProvisioningMixin:
ssh_public_keys: list[str] | None = None, ssh_public_keys: list[str] | None = None,
disk_resize_gb: int | None = None, disk_resize_gb: int | None = None,
storage: str | None = None, storage: str | None = None,
cpu_type: str | None = None,
download_timeout: int = 300, download_timeout: int = 300,
timeout: int = 180, timeout: int = 180,
) -> VMProvisionResultDict: ) -> VMProvisionResultDict:
if cpu_type is not None:
# CPU compatibility is a cluster setting here (EVC), not a per-VM
# one, and there is no get_vm_cpu_types() that could offer a model.
raise ValueError(f"VMware cannot give a single VM a CPU type ({cpu_type!r})")
self._check_nics(nics) self._check_nics(nics)
try: try:
vmdk, image_size = ImageCache(self._image_cache_dir).vmdk( vmdk, image_size = ImageCache(self._image_cache_dir).vmdk(
+11
View File
@@ -141,6 +141,17 @@ class TestCreate:
_create(driver) _create(driver)
staged.discard.assert_called_once_with("vm-99") 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()
def test_placement_refusal_is_a_runtime_error(self, driver, staged): def test_placement_refusal_is_a_runtime_error(self, driver, staged):
with patch.object(prov, "choose_placement", side_effect=ValueError("no host")): with patch.object(prov, "choose_placement", side_effect=ValueError("no host")):
with pytest.raises(RuntimeError, match="no host"): with pytest.raises(RuntimeError, match="no host"):