chore: adopt the NAPALM driver standard #8

Merged
christianmanivong merged 4 commits from chore/driver-standard into main 2026-10-23 10:43:57 +00:00
6 changed files with 27 additions and 10 deletions
Showing only changes of commit 06b446934f - Show all commits

No files matched your search

+2 -1
View File
@@ -11,7 +11,8 @@ from typing import Any
from napalm_device_types.models import HypervisorHostDict
_CONNECTED = "connected"
_UNMANAGED = {None, "", "0.0.0.0"}
# Addresses vSphere reports for "no managing vCenter"; data, not a bind.
_UNMANAGED = {None, "", "0.0.0.0"} # nosec B104
def _host(host: dict[str, Any], standalone: bool) -> HypervisorHostDict:
+2 -1
View File
@@ -12,7 +12,8 @@ import ipaddress
from typing import Any
_DEFAULT_MTU = 1500
_UNASSIGNED = {"", "0.0.0.0"}
# Addresses vSphere reports for an interface without one; data, not a bind.
_UNASSIGNED = {"", "0.0.0.0"} # nosec B104
def _uplink_mtus(host: dict[str, Any]) -> dict[str, int]:
+7 -3
View File
@@ -12,7 +12,9 @@ import hashlib
import json
import os
import shutil
import subprocess
# qemu-img is run with a fixed argv and no shell (see below).
import subprocess # nosec B404
import tempfile
from collections.abc import Callable
from pathlib import Path
@@ -46,7 +48,8 @@ def fetch(url: str, dest: Path, timeout: float) -> None: # pragma: no cover - n
def convert_to_vmdk(src: Path, dst: Path) -> None:
subprocess.run(
# Fixed argv, no shell: qemu-img from PATH and the cache's own file paths.
subprocess.run( # nosec B603
[
_qemu_img(),
"convert",
@@ -64,7 +67,8 @@ def convert_to_vmdk(src: Path, dst: Path) -> None:
def virtual_size(path: Path) -> int:
"""The disk size the image describes, in bytes (not the file size)."""
out = subprocess.run(
# Fixed argv, no shell: qemu-img from PATH and the cache's own file path.
out = subprocess.run( # nosec B603
[_qemu_img(), "info", "--output", "json", str(path)],
check=True,
capture_output=True,
+2 -1
View File
@@ -11,7 +11,8 @@ VMXNET3. Both drivers are in the mainline kernel; whether every distribution's
from __future__ import annotations
from xml.sax.saxutils import escape, quoteattr
# Escaping for the descriptor this module writes; no XML is parsed here.
from xml.sax.saxutils import escape, quoteattr # nosec B406
_HW_VERSION = "vmx-13" # ESXi 6.5 and later
_GUEST_OS = "otherLinux64Guest"
+7 -4
View File
@@ -252,7 +252,7 @@ class VmwareProvisioningMixin:
try:
lease.HttpNfcLeaseAbort()
except Exception: # noqa: BLE001 - the original error is the one to report
pass
logger.debug("Aborting the import lease failed: ignored", exc_info=True)
raise
return info["info"]["entity"]
@@ -279,13 +279,16 @@ class VmwareProvisioningMixin:
changes = _mac_changes(adapters, nics) + _disk_growth(devices, disk_gb, image_size)
macs = [
(nic.get("mac") or adapter.macAddress or "").lower()
for adapter, nic in zip(adapters, nics)
for adapter, nic in zip(adapters, nics, strict=True)
]
seed = nocloud_iso(
user,
meta_data(name, f"iid-{uuid.uuid4()}"),
network_config(
[(mac, nic.get("dhcp", i == 0)) for i, (mac, nic) in enumerate(zip(macs, nics))]
[
(mac, nic.get("dhcp", i == 0))
for i, (mac, nic) in enumerate(zip(macs, nics, strict=True))
]
),
)
vm_dir = posixpath.dirname(str(props["config.files.vmPathName"]).split("] ", 1)[1])
@@ -365,7 +368,7 @@ class VmwareProvisioningMixin:
def _mac_changes(adapters: list[Any], nics: list[NICConfigDict]) -> list[Any]:
"""Device edits pinning the MACs the caller asked for."""
changes = []
for adapter, nic in zip(adapters, nics):
for adapter, nic in zip(adapters, nics, strict=True):
if not nic.get("mac"):
continue
adapter.addressType = "manual"
+7
View File
@@ -258,6 +258,13 @@ class TestFinishVm:
_, 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):