From 2b84ceae3a3f651c2d8b61a139d245e2d5e417c1 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Thu, 24 Sep 2026 10:00:33 +0200 Subject: [PATCH] feat: VM provisioning, and reboot_host on ESXi create_vm_from_cloud_init takes the same qcow2/raw cloud images netOrk offers for Proxmox. ESXi can neither boot nor download them, so the driver does both: download with checksum check and one retry, convert with qemu-img to a streamOptimized VMDK (cached by URL), import through a minimal OVF descriptor over NFC, pin requested MACs, grow the disk, and attach a NoCloud seed ISO placed next to the VM's files. NoCloud rather than guestinfo because it needs nothing in the guest; the user-data installs open-vm-tools, which the driver declares as its guest agent. A failure after the import removes the VM again. Placement is a pure decision over inventory rows: a connected host outside maintenance mode that sees the datastore and every port group, with the resource pool and VM folder found by walking up to the datacenter -- one path for a standalone host and for a vCenter. Two faults vcsim surfaced and the tests now pin: a chunked upload body next to a Content-Length is refused with 500, so the disk goes up as a sized file object that also reports lease progress; and a device edit replaces the device as sent, so disk and NIC edits start from the live objects, backing included (vcsim panicked on a disk without one). destroy_vm, get_vm_status, get_network_targets (port groups with their fixed VLAN) and get_image_storages complete the contract. reboot_host on ESXi uses RebootHost_Task and refuses outside maintenance mode: force=True would cut power to running VMs. Tested against vcsim in ESXi and vCenter mode, end to end. --- CHANGELOG.md | 8 + README.md | 38 ++- napalm_vmware/_inventory.py | 20 +- napalm_vmware/base.py | 4 + napalm_vmware/esxi.py | 19 +- napalm_vmware/paths.py | 16 ++ napalm_vmware/provision/__init__.py | 0 napalm_vmware/provision/image.py | 135 +++++++++ napalm_vmware/provision/ovf.py | 120 ++++++++ napalm_vmware/provision/placement.py | 107 ++++++++ napalm_vmware/provision/seed.py | 67 +++++ napalm_vmware/provision/transfer.py | 113 ++++++++ napalm_vmware/provisioning.py | 396 +++++++++++++++++++++++++++ napalm_vmware/vcenter.py | 3 +- pyproject.toml | 4 + tests/test_actions.py | 18 ++ tests/test_drivers.py | 19 +- tests/test_inventory.py | 12 + tests/test_provision_image.py | 107 ++++++++ tests/test_provision_ovf.py | 79 ++++++ tests/test_provision_placement.py | 114 ++++++++ tests/test_provision_seed.py | 80 ++++++ tests/test_provision_transfer.py | 63 +++++ tests/test_provisioning.py | 277 +++++++++++++++++++ tests/test_vcsim.py | 46 ++++ 25 files changed, 1851 insertions(+), 14 deletions(-) create mode 100644 napalm_vmware/provision/__init__.py create mode 100644 napalm_vmware/provision/image.py create mode 100644 napalm_vmware/provision/ovf.py create mode 100644 napalm_vmware/provision/placement.py create mode 100644 napalm_vmware/provision/seed.py create mode 100644 napalm_vmware/provision/transfer.py create mode 100644 napalm_vmware/provisioning.py create mode 100644 tests/test_provision_image.py create mode 100644 tests/test_provision_ovf.py create mode 100644 tests/test_provision_placement.py create mode 100644 tests/test_provision_seed.py create mode 100644 tests/test_provision_transfer.py create mode 100644 tests/test_provisioning.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 309d430..5875c46 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- VM provisioning (`create_vm_from_cloud_init`, `destroy_vm`, + `get_vm_status`, `get_network_targets`, `get_image_storages`) from qcow2/raw + cloud images: converted with qemu-img, imported over OVF/NFC, configured + by cloud-init from a NoCloud seed ISO. Tested against vcsim. +- `reboot_host` on ESXi, refused outside maintenance mode. +- Declares `open-vm-tools` as the guest agent for provisioned VMs. + ## [0.1.0] - 2026-09-24 ### Added diff --git a/README.md b/README.md index 40487be..f908950 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,10 @@ hardware**: see [Harvesting fixtures](#harvesting-fixtures). | `get_vm_storage_pools` | ✅ | ✅ | datastores | | `get_virtual_networks` | ✅ | ✅ (+ dvPortgroups) | port groups | | `get_device_warnings` | ✅ | ✅ | raw `{code, meta}` | -| VM provisioning, VIBs, updates, host reboot | — | — | out of scope for v1 | +| `reboot_host` | ✅ (maintenance mode only) | — | `RebootHost_Task` | +| `create_vm_from_cloud_init`, `destroy_vm`, `get_vm_status` | ✅ | ✅ | see [Provisioning](#provisioning) | +| `get_network_targets`, `get_image_storages` | ✅ | ✅ | port groups, datastores | +| VIBs, updates | — | — | not yet | Unverified assumptions, to be checked against real hardware: @@ -44,6 +47,9 @@ Unverified assumptions, to be checked against real hardware: - An account that may read the inventory. For power and snapshot actions it also needs *Virtual machine → Interaction → Power on/off/Reset/Suspend* and *Virtual machine → Snapshot management*. +- For provisioning: `qemu-img` (package `qemu-utils`) where the driver runs, + and HTTPS from there to every ESXi host that may receive a VM -- through a + vCenter the disk upload goes straight to the host, not via the vCenter. - **A paid license for any write.** On the free vSphere Hypervisor license the API is read-only; the driver reports `vmware_api_read_only` and turns the refusal into a readable error. @@ -69,7 +75,9 @@ driver.close() ``` `optional_args`: `port` (default 443), `verify_ssl` / `ssl_verify` (default -`True`; ESXi ships a self-signed certificate). Other keys are ignored. +`True`; ESXi ships a self-signed certificate), `image_cache_dir` (where +converted cloud images are kept; default a directory under the system temp +dir). Other keys are ignored. VMs are addressed by name, by `vmid`, or by MoRef (`vm-42`). A name shared by two VMs is refused rather than guessed. @@ -94,6 +102,32 @@ tools/sanitize.py tools/harvest-out/esxi8-dell.json > tests/fixtures/esxi8-dell. the ones the drivers read. `tools/harvest-out/` is gitignored. Read the sanitised file before committing it. +## Provisioning + +`create_vm_from_cloud_init` takes the same cloud images netOrk's catalog +offers for Proxmox (qcow2/raw): + +1. The image is downloaded where the driver runs, its checksum verified (one + retry), and converted with `qemu-img` to a streamOptimized VMDK. The VMDK + is cached by URL in `image_cache_dir`; the download is not kept. +2. A host is chosen that is connected, not in maintenance mode, and sees the + datastore and every requested port group (the named datastore, or the one + with the most free space). +3. The VM is created from a minimal OVF descriptor (PVSCSI disk, VMXNET3 + NICs) and the disk streamed in over NFC. +4. Requested MACs are pinned, the disk grown to `disk_resize_gb`, and a + NoCloud seed ISO (user-data, meta-data, network-config) uploaded next to + the VM's files and attached as a CD-ROM on a new SATA controller. +5. The VM is powered on. Any failure after step 3 removes the VM again. + +A port group fixes its VLAN, so `get_network_targets` reports each one with +`kind="portgroup"`, `vlan_aware=False` and its `fixed_vlan_tag`; a NIC asking +for a different `vlan_tag` is refused. The guest agent netOrk installs is +`open-vm-tools` (`GUEST_AGENT_PACKAGES`), which reports the IP address back. + +Unverified until #305: that each distribution's cloud kernel carries the +PVSCSI and VMXNET3 drivers. + ## Design notes **One seam.** Every read goes through `Inventory.collect(type, paths)`, a diff --git a/napalm_vmware/_inventory.py b/napalm_vmware/_inventory.py index c160824..5b4577e 100644 --- a/napalm_vmware/_inventory.py +++ b/napalm_vmware/_inventory.py @@ -56,9 +56,13 @@ class Inventory: finally: view.Destroy() - def properties(self, obj: Any, paths: Sequence[str]) -> dict[str, Any]: - """The given properties of one managed object; ``{}`` if it is gone.""" - rows = self._retrieve(_PC.ObjectSpec(obj=obj), type(obj), paths) + def properties(self, obj: Any, paths: Sequence[str], *, raw: bool = False) -> dict[str, Any]: + """The given properties of one managed object; ``{}`` if it is gone. + + ``raw=True`` keeps the pyVmomi objects -- for a caller that has to hand + a device back to vSphere whole, backing and all, in a reconfigure. + """ + rows = self._retrieve(_PC.ObjectSpec(obj=obj), type(obj), paths, raw=raw) return rows[0] if rows else {} def licenses(self) -> list[dict[str, Any]]: @@ -68,7 +72,9 @@ class Inventory: return [] return self.properties(manager, ["licenses"]).get("licenses", []) - def _retrieve(self, obj_spec: Any, vim_type: Any, paths: Sequence[str]) -> list[dict[str, Any]]: + def _retrieve( + self, obj_spec: Any, vim_type: Any, paths: Sequence[str], *, raw: bool = False + ) -> list[dict[str, Any]]: spec = _PC.FilterSpec( objectSet=[obj_spec], propSet=[_PC.PropertySpec(type=vim_type, pathSet=list(paths))] ) @@ -76,15 +82,15 @@ class Inventory: result = pc.RetrievePropertiesEx([spec], _PC.RetrieveOptions(maxObjects=_PAGE_SIZE)) rows: list[dict[str, Any]] = [] while result is not None: - rows.extend(_row(obj) for obj in result.objects) + rows.extend(_row(obj, raw) for obj in result.objects) if not result.token: break result = pc.ContinueRetrievePropertiesEx(token=result.token) return rows -def _row(obj_content: Any) -> dict[str, Any]: +def _row(obj_content: Any, raw: bool = False) -> dict[str, Any]: row: dict[str, Any] = {"_moref": obj_content.obj._moId} for prop in obj_content.propSet or []: - row[prop.name] = to_plain(prop.val) + row[prop.name] = prop.val if raw else to_plain(prop.val) return row diff --git a/napalm_vmware/base.py b/napalm_vmware/base.py index 8864d48..ca21301 100644 --- a/napalm_vmware/base.py +++ b/napalm_vmware/base.py @@ -2,6 +2,7 @@ from __future__ import annotations +from pathlib import Path from typing import Any, ClassVar from napalm.base.exceptions import ConnectionException @@ -22,6 +23,7 @@ from napalm_vmware.parse.storage import storage_pools from napalm_vmware.parse.vm_config import vm_config from napalm_vmware.parse.vms import vm_list from napalm_vmware.parse.warnings import host_warnings +from napalm_vmware.provision.image import DEFAULT_CACHE_DIR _DEFAULT_PORT = 443 #: ``about.apiType`` -> the driver that handles it, for a helpful refusal. @@ -55,6 +57,8 @@ class VmwareBaseDriver(HypervisorDriver): args = optional_args or {} self._port = int(args.get("port") or _DEFAULT_PORT) self._verify_ssl = bool(args.get("verify_ssl", args.get("ssl_verify", True))) + # Where converted cloud images are kept between provisioning jobs. + self._image_cache_dir = Path(args.get("image_cache_dir") or DEFAULT_CACHE_DIR) self._si: Any = None self._inventory: Any = None diff --git a/napalm_vmware/esxi.py b/napalm_vmware/esxi.py index bd42635..466fb25 100644 --- a/napalm_vmware/esxi.py +++ b/napalm_vmware/esxi.py @@ -14,9 +14,10 @@ from napalm_vmware.base import VmwareBaseDriver from napalm_vmware.parse.facts import esxi_facts from napalm_vmware.parse.interfaces import host_interfaces, host_interfaces_ip from napalm_vmware.parse.lldp import lldp_neighbors +from napalm_vmware.provisioning import VmwareProvisioningMixin -class VmwareEsxiDriver(VmwareActionsMixin, VmwareBaseDriver): +class VmwareEsxiDriver(VmwareProvisioningMixin, VmwareActionsMixin, VmwareBaseDriver): """One ESXi host. Its VMs, vmnics, vmkernel NICs, datastores and port groups.""" DRIVER_NAME = "vmware_esxi" @@ -46,6 +47,22 @@ class VmwareEsxiDriver(VmwareActionsMixin, VmwareBaseDriver): def get_interfaces_ip(self) -> dict[str, dict[str, Any]]: return host_interfaces_ip(self._host()) + def reboot_host(self) -> None: + """Restart the host, which has to be in maintenance mode. + + ``force=True`` would restart a host with running VMs by cutting their + power. Evacuating them -- or deciding not to -- is the operator's call, + made by entering maintenance mode in vSphere first. + """ + host = self._host() + if not host.get("runtime.inMaintenanceMode"): + raise RuntimeError( + f"{host.get('name', self.hostname)} is not in maintenance mode; " + "enter it in vSphere first so running VMs are not powered off" + ) + # The task finishes as the host goes down; there is nothing to wait for. + invoke(self._mo(vim.HostSystem, host["_moref"]).RebootHost_Task, force=False) + def get_lldp_neighbors(self) -> dict[str, list[dict[str, str]]]: network_system = self._host().get("configManager.networkSystem") if not network_system: diff --git a/napalm_vmware/paths.py b/napalm_vmware/paths.py index 3804d0a..8815a98 100644 --- a/napalm_vmware/paths.py +++ b/napalm_vmware/paths.py @@ -26,6 +26,9 @@ HOST = ( "config.autoStart", "configIssue", "configManager.networkSystem", + "parent", + "datastore", + "network", ) VM = ( @@ -47,6 +50,8 @@ VM = ( "summary.quickStats", "guest.net", "guest.ipAddress", + "guest.hostName", + "config.files.vmPathName", "guest.toolsStatus", "guest.toolsRunningStatus", "snapshot", @@ -66,6 +71,13 @@ DV_PORTGROUP = ( DV_SWITCH = ("name",) +# Placement of new VMs: which network objects exist, and the parent chain from +# a host's compute resource up to its datacenter. +NETWORK = ("name",) +COMPUTE_RESOURCE = ("parent", "resourcePool") +FOLDER = ("parent",) +DATACENTER = ("name", "vmFolder") + #: Managed object type name -> paths, in the order harvest.py dumps them. ALL = { "HostSystem": HOST, @@ -73,4 +85,8 @@ ALL = { "Datastore": DATASTORE, "DistributedVirtualPortgroup": DV_PORTGROUP, "DistributedVirtualSwitch": DV_SWITCH, + "Network": NETWORK, + "ComputeResource": COMPUTE_RESOURCE, + "Folder": FOLDER, + "Datacenter": DATACENTER, } diff --git a/napalm_vmware/provision/__init__.py b/napalm_vmware/provision/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/napalm_vmware/provision/image.py b/napalm_vmware/provision/image.py new file mode 100644 index 0000000..b9e8bc2 --- /dev/null +++ b/napalm_vmware/provision/image.py @@ -0,0 +1,135 @@ +"""Cloud images (qcow2/raw) converted to streamOptimized VMDKs, cached by URL. + +ESXi cannot boot a qcow2 and cannot download one itself, so the conversion +runs where the driver runs: download, verify, ``qemu-img convert``. The +result is kept, keyed by URL, so the next VM from the same image skips both +steps. Only the converted disk is kept; the download is deleted. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import subprocess +import tempfile +from collections.abc import Callable +from pathlib import Path + +import requests + +_CHUNK = 1024 * 1024 +DEFAULT_CACHE_DIR = Path(tempfile.gettempdir()) / "napalm-vmware-images" + +Fetch = Callable[[str, Path, float], None] +Convert = Callable[[Path, Path], None] +VirtualSize = Callable[[Path], int] + + +def _qemu_img() -> str: + path = shutil.which("qemu-img") + if path is None: + raise RuntimeError( + "qemu-img is not installed where netOrk runs the provisioning job; " + "it is needed to convert cloud images for VMware (package qemu-utils)" + ) + return path + + +def fetch(url: str, dest: Path, timeout: float) -> None: # pragma: no cover - network + with requests.get(url, stream=True, timeout=timeout) as response: + response.raise_for_status() + with dest.open("wb") as out: + for chunk in response.iter_content(_CHUNK): + out.write(chunk) + + +def convert_to_vmdk(src: Path, dst: Path) -> None: + subprocess.run( + [ + _qemu_img(), + "convert", + "-O", + "vmdk", + "-o", + "subformat=streamOptimized", + str(src), + str(dst), + ], + check=True, + capture_output=True, + ) + + +def virtual_size(path: Path) -> int: + """The disk size the image describes, in bytes (not the file size).""" + out = subprocess.run( + [_qemu_img(), "info", "--output", "json", str(path)], + check=True, + capture_output=True, + text=True, + ).stdout + return int(json.loads(out)["virtual-size"]) + + +def _digest(path: Path, algorithm: str) -> str: + h = hashlib.new(algorithm) + with path.open("rb") as fh: + for chunk in iter(lambda: fh.read(_CHUNK), b""): + h.update(chunk) + return h.hexdigest() + + +class ImageCache: + """Converted images in ``directory``, one ``.vmdk`` plus ``.json`` each.""" + + def __init__( + self, + directory: Path = DEFAULT_CACHE_DIR, + *, + fetch: Fetch = fetch, + convert: Convert = convert_to_vmdk, + virtual_size: VirtualSize = virtual_size, + ) -> None: + self._dir = Path(directory) + self._fetch = fetch + self._convert = convert + self._virtual_size = virtual_size + + def vmdk(self, url: str, checksum: str | None, timeout: float) -> tuple[Path, int]: + """``(path to the VMDK, virtual disk size in bytes)`` for ``url``.""" + self._dir.mkdir(parents=True, exist_ok=True) + key = hashlib.sha256(url.encode()).hexdigest()[:16] + vmdk, meta = self._dir / f"{key}.vmdk", self._dir / f"{key}.json" + if vmdk.exists() and meta.exists(): + return vmdk, int(json.loads(meta.read_text())["virtual_size"]) + + download = self._dir / f"{key}.download" + try: + self._download_verified(url, checksum, download, timeout) + size = self._virtual_size(download) + partial = self._dir / f"{key}.vmdk.partial" + self._convert(download, partial) + # Rename last: a VMDK that exists is a VMDK that is complete, even + # when two jobs convert the same image at once. + os.replace(partial, vmdk) + meta.write_text(json.dumps({"url": url, "virtual_size": size})) + finally: + download.unlink(missing_ok=True) + return vmdk, size + + def _download_verified( + self, url: str, checksum: str | None, dest: Path, timeout: float + ) -> None: + algorithm, _, expected = (checksum or "").rpartition(":") + algorithm = (algorithm or "sha256").lower() + for _attempt in (1, 2): + self._fetch(url, dest, timeout) + if not checksum: + return + actual = _digest(dest, algorithm) + if actual.lower() == expected.lower(): + return + dest.unlink(missing_ok=True) + raise RuntimeError(f"Checksum mismatch for {url}: expected {expected}, got {actual}") diff --git a/napalm_vmware/provision/ovf.py b/napalm_vmware/provision/ovf.py new file mode 100644 index 0000000..971d853 --- /dev/null +++ b/napalm_vmware/provision/ovf.py @@ -0,0 +1,120 @@ +"""A minimal OVF 1.0 descriptor for importing one converted cloud image. + +vSphere builds the VM from this (``OvfManager.CreateImportSpec``) and then +takes the disk contents over NFC. The descriptor only has to describe what +the image does not: CPU, memory, one disk and the NICs. + +Devices are the ones VMware recommends for a 64-bit Linux guest -- PVSCSI and +VMXNET3. Both drivers are in the mainline kernel; whether every distribution's +*cloud* kernel carries them is one of the things #305 checks on real hardware. +""" + +from __future__ import annotations + +from xml.sax.saxutils import escape, quoteattr + +_HW_VERSION = "vmx-13" # ESXi 6.5 and later +_GUEST_OS = "otherLinux64Guest" +_STREAM_OPTIMIZED = "http://www.vmware.com/interfaces/specifications/vmdk.html#streamOptimized" + + +def _item(**elements: str) -> str: + # CIM requires the rasd elements in alphabetical order. + body = "".join(f"{escape(v)}" for k, v in sorted(elements.items())) + return f"{body}" + + +def _nic_items(nic_count: int, first_instance: int) -> str: + return "".join( + _item( + AutomaticAllocation="true", + Connection=f"net{i}", + ElementName=f"Network adapter {i + 1}", + InstanceID=str(first_instance + i), + ResourceSubType="VmxNet3", + ResourceType="10", + ) + for i in range(nic_count) + ) + + +def ovf_descriptor( + name: str, + *, + cpu: int, + memory_mb: int, + capacity_bytes: int, + nic_count: int, + firmware: str = "bios", +) -> str: + networks = "".join( + f'net{i}' + for i in range(nic_count) + ) + hardware = "".join( + [ + _item( + AllocationUnits="hertz * 10^6", + ElementName=f"{cpu} virtual CPU(s)", + InstanceID="1", + ResourceType="3", + VirtualQuantity=str(cpu), + ), + _item( + AllocationUnits="byte * 2^20", + ElementName=f"{memory_mb} MB of memory", + InstanceID="2", + ResourceType="4", + VirtualQuantity=str(memory_mb), + ), + _item( + Address="0", + ElementName="SCSI controller 0", + InstanceID="3", + ResourceSubType="VirtualSCSI", + ResourceType="6", + ), + _item( + AddressOnParent="0", + ElementName="Hard disk 1", + HostResource="ovf:/disk/vmdisk1", + InstanceID="4", + Parent="3", + ResourceType="17", + ), + _nic_items(nic_count, first_instance=5), + ] + ) + firmware_config = ( + '' + if firmware == "efi" + else "" + ) + return ( + '' + '' + '' + "Virtual disks" + f'' + "" + f"Networks{networks}" + f"" + "A virtual machine" + f"{escape(name)}" + f'Guest OS' + "" + "Virtual hardware" + "Virtual Hardware Family" + "0" + f"{escape(name)}" + f"{_HW_VERSION}" + f"{hardware}{firmware_config}" + "" + "" + "" + ) diff --git a/napalm_vmware/provision/placement.py b/napalm_vmware/provision/placement.py new file mode 100644 index 0000000..7495f70 --- /dev/null +++ b/napalm_vmware/provision/placement.py @@ -0,0 +1,107 @@ +"""Where a new VM goes: a pure decision over plain inventory rows. + +A host qualifies when it is connected, not in maintenance mode, and sees both +a usable datastore and every requested network. Among those, the pair with +the most free space wins -- or the named datastore, when one is given. The +resource pool comes from the host's compute resource and the VM folder from +the datacenter above it; a standalone ESXi host has the same shape +(``ha-compute-res`` under ``ha-datacenter``), so one path serves both. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +Row = dict[str, Any] + + +@dataclass +class Inventory: + hosts: list[Row] + datastores: list[Row] + networks: list[Row] + compute_resources: list[Row] + folders: list[Row] = field(default_factory=list) + datacenters: list[Row] = field(default_factory=list) + + +@dataclass +class Placement: + host: str + host_name: str + datastore: str + datastore_name: str + resource_pool: str + folder: str + datacenter_name: str + networks: list[str] + + +def _usable_datastores(inv: Inventory, storage: str | None) -> dict[str, Row]: + usable = { + ds["_moref"]: ds + for ds in inv.datastores + if (ds.get("summary") or {}).get("accessible") + and (ds.get("summary") or {}).get("maintenanceMode", "normal") == "normal" + and (storage is None or ds.get("name") == storage) + } + if storage is not None and not usable: + raise ValueError(f"There is no usable datastore named {storage!r}") + return usable + + +def _host_networks( + host: Row, names_by_moref: dict[str, str], wanted: list[str] +) -> list[str] | None: + """The host's network moref for each wanted name, or None if one is missing.""" + by_name = {names_by_moref.get(m): m for m in host.get("network") or []} + found = [by_name.get(name) for name in wanted] + return None if None in found else [m for m in found if m] + + +def _datacenter(inv: Inventory, compute_resource: str) -> Row: + parents = {row["_moref"]: row.get("parent") for row in inv.compute_resources + inv.folders} + datacenters = {dc["_moref"]: dc for dc in inv.datacenters} + node: str | None = compute_resource + while node is not None and node not in datacenters: + node = parents.get(node) + if node is None: + raise RuntimeError(f"No datacenter above {compute_resource!r}") + return datacenters[node] + + +def choose_placement(inv: Inventory, storage: str | None, networks: list[str]) -> Placement: + usable = _usable_datastores(inv, storage) + names = {n["_moref"]: n.get("name", "") for n in inv.networks} + best: tuple[int, Row, Row, list[str]] | None = None + for host in sorted(inv.hosts, key=lambda h: h.get("name", "")): + if host.get("runtime.connectionState") != "connected" or host.get( + "runtime.inMaintenanceMode" + ): + continue + host_nets = _host_networks(host, names, networks) + if host_nets is None: + continue + for ds_ref in host.get("datastore") or []: + ds = usable.get(ds_ref) + free = int((ds or {}).get("summary", {}).get("freeSpace", 0)) + if ds is not None and (best is None or free > best[0]): + best = (free, host, ds, host_nets) + if best is None: + raise ValueError( + f"There is no host that sees datastore {storage or '(any)'} and networks {networks}" + ) + _, host, ds, host_nets = best + compute = next(c for c in inv.compute_resources if c["_moref"] == host.get("parent")) + dc = _datacenter(inv, compute["_moref"]) + return Placement( + host=host["_moref"], + host_name=host.get("name", ""), + datastore=ds["_moref"], + datastore_name=ds.get("name", ""), + resource_pool=compute["resourcePool"], + folder=dc["vmFolder"], + datacenter_name=dc.get("name", ""), + networks=host_nets, + ) diff --git a/napalm_vmware/provision/seed.py b/napalm_vmware/provision/seed.py new file mode 100644 index 0000000..2074a5f --- /dev/null +++ b/napalm_vmware/provision/seed.py @@ -0,0 +1,67 @@ +"""cloud-init's NoCloud seed: user-data, meta-data, network-config on an ISO. + +NoCloud rather than VMware's guestinfo datasource because it needs nothing in +the guest: guestinfo is read through open-vm-tools, which a generic cloud +image does not ship -- it is one of the things the user-data installs. +""" + +from __future__ import annotations + +import io +from typing import Any + +import pycdlib +import yaml + +#: The volume label cloud-init's NoCloud datasource looks for. +_LABEL = "cidata" + + +def user_data(cloud_init_config: dict[str, Any], ssh_public_keys: list[str] | None) -> str: + """The ``#cloud-config`` document, with ``ssh_public_keys`` merged in.""" + config = dict(cloud_init_config) + if ssh_public_keys: + keys = list(config.get("ssh_authorized_keys") or []) + keys += [k for k in ssh_public_keys if k not in keys] + config["ssh_authorized_keys"] = keys + return "#cloud-config\n" + yaml.safe_dump(config, sort_keys=False) + + +def meta_data(hostname: str, instance_id: str) -> str: + return yaml.safe_dump({"instance-id": instance_id, "local-hostname": hostname}, sort_keys=False) + + +def network_config(nics: list[tuple[str, bool]]) -> dict[str, Any] | None: + """Netplan v2 config: DHCP on every ``(mac, dhcp)`` NIC that asks for it. + + ``None`` when no NIC does -- cloud-init then leaves networking alone + rather than being told to configure nothing. + """ + ethernets = { + f"nic{i}": {"match": {"macaddress": mac}, "dhcp4": True} + for i, (mac, dhcp) in enumerate(nics) + if dhcp + } + return {"version": 2, "ethernets": ethernets} if ethernets else None + + +def nocloud_iso(user: str, meta: str, network: dict[str, Any] | None) -> bytes: + """An ISO 9660 image (Rock Ridge + Joliet) holding the seed files.""" + files = {"user-data": user, "meta-data": meta} + if network is not None: + files["network-config"] = yaml.safe_dump(network, sort_keys=False) + iso = pycdlib.PyCdlib() + iso.new(interchange_level=3, joliet=3, rock_ridge="1.09", vol_ident=_LABEL) + for index, (name, content) in enumerate(files.items()): + data = content.encode() + iso.add_fp( + io.BytesIO(data), + len(data), + f"/SEED{index}.;1", + rr_name=name, + joliet_path=f"/{name}", + ) + out = io.BytesIO() + iso.write_fp(out) + iso.close() + return out.getvalue() diff --git a/napalm_vmware/provision/transfer.py b/napalm_vmware/provision/transfer.py new file mode 100644 index 0000000..6f16703 --- /dev/null +++ b/napalm_vmware/provision/transfer.py @@ -0,0 +1,113 @@ +"""The two HTTP transfers provisioning needs: a disk over NFC, a file to a datastore. + +Both authenticate with the vSphere session's own cookie, so no second login +and no credentials beyond the ones the driver already holds. +""" + +from __future__ import annotations + +import time +from collections.abc import Callable +from pathlib import Path +from typing import Any +from urllib.parse import quote + +import requests + +#: vSphere drops an NFC lease that reports no progress for five minutes. +_PROGRESS_EVERY_SECONDS = 20.0 + + +class ProgressFile: + """A file body with a length, calling ``report(percent)`` as it is read. + + Sized so ``requests`` sends it with a Content-Length instead of chunked: + vSphere's NFC endpoint refuses a chunked disk upload. + """ + + def __init__( + self, + path: Path, + report: Callable[[int], None], + clock: Callable[[], float] = time.monotonic, + ) -> None: + self._fh = path.open("rb") + self._size = path.stat().st_size + self._sent = 0 + self._report = report + self._clock = clock + self._last = clock() + + def __len__(self) -> int: + return self._size + + def read(self, size: int = -1) -> bytes: + chunk = self._fh.read(size) + self._sent += len(chunk) + if chunk and self._clock() - self._last >= _PROGRESS_EVERY_SECONDS: + self._report(min(99, self._sent * 100 // (self._size or 1))) + self._last = self._clock() + return chunk + + def close(self) -> None: + self._fh.close() + + +def upload_disk( + url: str, vmdk: Path, cookie: str, verify: bool, report: Callable[[int], None] +) -> None: + """Stream a streamOptimized VMDK to an NFC lease's device URL.""" + body = ProgressFile(vmdk, report) + try: + response = requests.post( + url, + data=body, + headers={"Content-Type": "application/x-vnd.vmware-streamVmdk", "Cookie": cookie}, + verify=verify, + timeout=(30, 600), + ) + finally: + body.close() + response.raise_for_status() + + +def lease_url(url: str, hostname: str, port: int) -> str: + """An NFC device URL with its ``*`` host filled in. + + ESXi answers ``https://*/nfc/...``: "the host you are talking to". A port + other than 443 has to be carried over, or the upload misses a NAT'd host. + Through a vCenter the URL already names the ESXi host that takes the disk. + """ + host = hostname if port == 443 else f"{hostname}:{port}" + return url.replace("://*/", f"://{host}/", 1) + + +def datastore_url(base: str, path: str) -> str: + """``https://host/folder/`` -- the datastore file browser endpoint.""" + return f"{base}/folder/{quote(path)}" + + +def upload_file( + base: str, + datacenter: str, + datastore: str, + path: str, + data: bytes, + cookie: str, + verify: bool, +) -> None: + """Put ``data`` at ``[datastore] path``.""" + response = requests.put( + datastore_url(base, path), + params={"dcPath": datacenter, "dsName": datastore}, + data=data, + headers={"Content-Type": "application/octet-stream", "Cookie": cookie}, + verify=verify, + timeout=(30, 120), + ) + response.raise_for_status() + + +def session_cookie(si: Any) -> str: + """The ``vmware_soap_session`` cookie of a pyVmomi ServiceInstance.""" + return si._stub.cookie diff --git a/napalm_vmware/provisioning.py b/napalm_vmware/provisioning.py new file mode 100644 index 0000000..24770d0 --- /dev/null +++ b/napalm_vmware/provisioning.py @@ -0,0 +1,396 @@ +"""HypervisorDriver provisioning: new VMs from netOrk's cloud-image catalog. + +The image is downloaded and converted where the driver runs +(``provision/image.py``), imported over OVF/NFC, and configured by cloud-init +from a NoCloud ISO placed next to the VM's files. See the README for why this +path and not OVA templates or a Content Library. +""" + +from __future__ import annotations + +import logging +import posixpath +import time +import uuid +from pathlib import Path +from typing import Any + +from napalm_device_types.models import ( + NetworkTargetDict, + NICConfigDict, + StorageTargetDict, + VMProvisionResultDict, + VMStatusDict, +) +from pyVmomi import vim + +from napalm_vmware import paths +from napalm_vmware._plain import to_plain +from napalm_vmware._tasks import fault_message, invoke, wait_for_task, wait_until +from napalm_vmware.parse import vm_devices as dev +from napalm_vmware.parse.networks import virtual_networks +from napalm_vmware.parse.vms import vm_list +from napalm_vmware.provision import transfer +from napalm_vmware.provision.image import ImageCache +from napalm_vmware.provision.ovf import ovf_descriptor +from napalm_vmware.provision.placement import Inventory as PlacementInventory +from napalm_vmware.provision.placement import Placement, choose_placement +from napalm_vmware.provision.seed import meta_data, network_config, nocloud_iso, user_data + +logger = logging.getLogger(__name__) + +_GB = 1024**3 +_SEED_ISO = "cidata.iso" +_TASK_TIMEOUT = 300 +_STATUS = {"poweredOn": "running", "poweredOff": "stopped", "suspended": "suspended"} + + +class VmwareProvisioningMixin: + """Mixed into both drivers ahead of :class:`VmwareBaseDriver`.""" + + GUEST_AGENT_PACKAGES: tuple[str, ...] = ("open-vm-tools",) + # The unit is open-vm-tools on Debian/Ubuntu and vmtoolsd on EL/Fedora. + GUEST_AGENT_RUNCMD: tuple[str, ...] = ( + "systemctl enable --now open-vm-tools || systemctl enable --now vmtoolsd", + ) + + hostname: str + _port: int + _verify_ssl: bool + _image_cache_dir: Path + _si: Any + _inventory: Any + _mo: Any + _find_vm: Any + _hosts: Any + _index: Any + + # -- choosing targets -------------------------------------------------------- + + def get_image_storages(self) -> list[StorageTargetDict]: + result: list[StorageTargetDict] = [] + for ds in self._inventory.collect("Datastore", paths.DATASTORE): + summary = ds.get("summary") or {} + if ( + not summary.get("accessible") + or summary.get("maintenanceMode", "normal") != "normal" + ): + continue + result.append( + { + "name": ds.get("name", ""), + "type": str(summary.get("type", "")).lower(), + "total_gb": round(int(summary.get("capacity") or 0) / _GB, 1), + "available_gb": round(int(summary.get("freeSpace") or 0) / _GB, 1), + } + ) + return sorted(result, key=lambda s: s["name"]) + + def get_network_targets(self) -> list[NetworkTargetDict]: + networks = virtual_networks( + self._hosts(), + self._inventory.collect("DistributedVirtualPortgroup", paths.DV_PORTGROUP), + self._inventory.collect("DistributedVirtualSwitch", paths.DV_SWITCH), + ) + # The port group fixes the VLAN; a NIC cannot add a tag of its own. + return [ + { + "name": net["name"], + "kind": "portgroup", + "vlan_aware": False, + "fixed_vlan_tag": net["vlan_id"] or None, + } + for net in sorted(networks.values(), key=lambda n: n["name"]) + ] + + def _check_nics(self, nics: list[NICConfigDict]) -> None: + if not nics: + raise RuntimeError("A VM needs at least one network adapter") + targets = {t["name"]: t for t in self.get_network_targets()} + for nic in nics: + target = targets.get(nic["bridge"]) + if target is None: + raise RuntimeError(f"There is no port group named {nic['bridge']!r}") + if nic.get("trunk_vlan_tags"): + raise RuntimeError("VMware port groups cannot trunk per adapter") + tag = nic.get("vlan_tag") + if tag and tag != target.get("fixed_vlan_tag"): + raise RuntimeError( + f"Port group {nic['bridge']!r} carries VLAN " + f"{target.get('fixed_vlan_tag') or 'untagged'}, not {tag}; " + "choose a port group on the VLAN instead" + ) + + def _placement_inventory(self) -> PlacementInventory: + collect = self._inventory.collect + return PlacementInventory( + hosts=self._hosts(), + datastores=collect("Datastore", paths.DATASTORE), + networks=collect("Network", paths.NETWORK), + compute_resources=collect("ComputeResource", paths.COMPUTE_RESOURCE), + folders=collect("Folder", paths.FOLDER), + datacenters=collect("Datacenter", paths.DATACENTER), + ) + + # -- creating ---------------------------------------------------------------- + + def create_vm_from_cloud_init( + self, + name: str, + *, + image_url: str, + cpu: int, + memory: int, + nics: list[NICConfigDict], + cloud_init_config: dict[str, Any], + image_checksum: str | None = None, + ssh_public_keys: list[str] | None = None, + disk_resize_gb: int | None = None, + storage: str | None = None, + download_timeout: int = 300, + timeout: int = 180, + ) -> VMProvisionResultDict: + self._check_nics(nics) + try: + vmdk, image_size = ImageCache(self._image_cache_dir).vmdk( + image_url, image_checksum, download_timeout + ) + place = choose_placement( + self._placement_inventory(), storage, [n["bridge"] for n in nics] + ) + except ValueError as exc: + raise RuntimeError(str(exc)) from exc + descriptor = ovf_descriptor( + name, cpu=cpu, memory_mb=memory, capacity_bytes=image_size, nic_count=len(nics) + ) + vm_ref = self._import_ovf(name, descriptor, vmdk, place, timeout) + try: + self._finish_vm( + vm_ref, + name, + place, + nics, + user_data(cloud_init_config, ssh_public_keys), + disk_gb=disk_resize_gb, + image_size=image_size, + ) + self._run_task(self._mo(vim.VirtualMachine, vm_ref).PowerOnVM_Task) + except Exception as exc: + logger.warning("Provisioning %s failed after import, removing it: %s", name, exc) + self._discard(vm_ref) + raise RuntimeError(f"Provisioning {name!r} failed: {exc}") from exc + props = self._inventory.properties( + self._mo(vim.VirtualMachine, vm_ref), ["config.instanceUuid"] + ) + return {"vmid": props["config.instanceUuid"], "name": name, "node": place.host_name} + + def _run_task(self, call: Any, *args: Any, **kwargs: Any) -> None: + wait_for_task(self._inventory, invoke(call, *args, **kwargs), _TASK_TIMEOUT) + + def _import_ovf( # pragma: no cover - live NFC session; covered by tests/test_vcsim.py + self, name: str, descriptor: str, vmdk: Path, place: Placement, timeout: int + ) -> str: + """Create the VM from ``descriptor`` and stream ``vmdk`` into it; return its MoRef.""" + content = self._si.RetrieveContent() + pool = self._mo(vim.ResourcePool, place.resource_pool) + params = vim.OvfManager.CreateImportSpecParams( + entityName=name, + diskProvisioning="thin", + networkMapping=[ + vim.OvfManager.NetworkMapping(name=f"net{i}", network=self._mo(vim.Network, ref)) + for i, ref in enumerate(place.networks) + ], + ) + spec = invoke( + content.ovfManager.CreateImportSpec, + descriptor, + pool, + self._mo(vim.Datastore, place.datastore), + params, + ) + if spec.error: + raise RuntimeError("; ".join(e.msg or type(e).__name__ for e in spec.error)) + lease = invoke( + pool.ImportVApp, + spec.importSpec, + self._mo(vim.Folder, place.folder), + self._mo(vim.HostSystem, place.host), + ) + try: + wait_until( + lambda: ( + self._inventory.properties(lease, ["state"]).get("state") in ("ready", "error") + ), + timeout, + "the import lease", + ) + info = self._inventory.properties(lease, ["state", "error", "info"]) + if info.get("state") == "error": + raise RuntimeError(fault_message(info.get("error") or {})) + url = transfer.lease_url(info["info"]["deviceUrl"][0]["url"], self.hostname, self._port) + transfer.upload_disk( + url, + vmdk, + transfer.session_cookie(self._si), + self._verify_ssl, + report=lambda pct: lease.HttpNfcLeaseProgress(pct), + ) + lease.HttpNfcLeaseComplete() + except Exception: + try: + lease.HttpNfcLeaseAbort() + except Exception: # noqa: BLE001 - the original error is the one to report + pass + raise + return info["info"]["entity"] + + def _finish_vm( + self, + vm_ref: str, + name: str, + place: Placement, + nics: list[NICConfigDict], + user: str, + *, + disk_gb: int | None, + image_size: int, + ) -> None: + """MACs, disk size and the cloud-init seed: everything the OVF could not say.""" + vm_mo = self._mo(vim.VirtualMachine, vm_ref) + props = self._inventory.properties( + vm_mo, ["config.hardware.device", "config.files.vmPathName"], raw=True + ) + # Live device objects: an "edit" replaces the device with what is sent, + # so it has to be the whole device, backing included. + devices = list(props.get("config.hardware.device") or []) + adapters = [d for d in devices if dev.is_nic(to_plain(d))] + 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) + ] + 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))] + ), + ) + vm_dir = posixpath.dirname(str(props["config.files.vmPathName"]).split("] ", 1)[1]) + self._upload_seed(place, f"{vm_dir}/{_SEED_ISO}", seed) + iso_path = f"[{place.datastore_name}] {vm_dir}/{_SEED_ISO}" + changes += _seed_cdrom(iso_path) + self._run_task(vm_mo.ReconfigVM_Task, spec=vim.vm.ConfigSpec(deviceChange=changes)) + + def _upload_seed(self, place: Placement, path: str, data: bytes) -> None: # pragma: no cover + transfer.upload_file( + f"https://{self.hostname}:{self._port}", + place.datacenter_name, + place.datastore_name, + path, + data, + transfer.session_cookie(self._si), + self._verify_ssl, + ) + + def _discard(self, vm_ref: str) -> None: + vm_mo = self._mo(vim.VirtualMachine, vm_ref) + try: + state = self._inventory.properties(vm_mo, ["runtime.powerState"]).get( + "runtime.powerState" + ) + if state == "poweredOn": + self._run_task(vm_mo.PowerOffVM_Task) + self._run_task(vm_mo.Destroy_Task) + except Exception as exc: # noqa: BLE001 - report the provisioning error, not this one + logger.error("Could not remove half-provisioned VM %s: %s", vm_ref, exc) + + # -- after creation ------------------------------------------------------------ + + def destroy_vm(self, vmid: str, *, remove_disk: bool = True, timeout: int = 60) -> None: + try: + vm = self._find_vm(vmid) + except ValueError as exc: + raise RuntimeError(str(exc)) from exc + vm_mo = self._mo(vim.VirtualMachine, vm["_moref"]) + if vm.get("runtime.powerState") == "poweredOn": + self._run_task(vm_mo.PowerOffVM_Task) + if remove_disk: + self._run_task(vm_mo.Destroy_Task) + else: + invoke(vm_mo.UnregisterVM) + + def get_vm_status( + self, vmid: str, *, wait_for_ip: bool = False, timeout: int = 300, poll_interval: int = 5 + ) -> VMStatusDict: + deadline = time.monotonic() + timeout + while True: + status = self._vm_status(vmid) + if not wait_for_ip or status.get("ip_address"): + return status + if time.monotonic() >= deadline: + raise RuntimeError(f"VM {vmid} reported no IP address within {timeout}s") + time.sleep(poll_interval) + + def _vm_status(self, vmid: str) -> VMStatusDict: + try: + vm = self._find_vm(vmid) + except ValueError as exc: + raise RuntimeError(str(exc)) from exc + hosts = self._hosts() + (entry,) = vm_list([vm], hosts, self._index(hosts)) + status: VMStatusDict = {"status": _STATUS.get(vm.get("runtime.powerState", ""), "unknown")} + if entry["ipv4"]: + status["ip_address"] = entry["ipv4"] + if vm.get("guest.hostName"): + status["hostname"] = vm["guest.hostName"] + first_nic = next(iter(entry["interfaces"].values()), None) + if first_nic and first_nic["mac_address"]: + status["mac_address"] = first_nic["mac_address"] + return status + + +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): + if not nic.get("mac"): + continue + adapter.addressType = "manual" + adapter.macAddress = nic["mac"].lower() + changes.append(vim.vm.device.VirtualDeviceSpec(operation="edit", device=adapter)) + return changes + + +def _disk_growth(devices: list[Any], disk_gb: int | None, image_size: int) -> list[Any]: + if not disk_gb or disk_gb * _GB <= image_size: + return [] + disk = next(d for d in devices if isinstance(d, vim.vm.device.VirtualDisk)) + disk.capacityInKB = disk_gb * 1024 * 1024 + # Both fields, or the live object carries the old size in the other one; + # pyVmomi's stubs type capacityInBytes as None, hence setattr. + setattr(disk, "capacityInBytes", disk_gb * _GB) # noqa: B010 + return [vim.vm.device.VirtualDeviceSpec(operation="edit", device=disk)] + + +def _seed_cdrom(iso_path: str) -> list[Any]: + """A SATA controller and a CD-ROM with the seed ISO, both new. + + The OVF describes only a SCSI controller; a CD-ROM needs IDE or SATA. + Negative keys are placeholders vSphere replaces, and let the CD-ROM name + its not-yet-existing controller. + """ + controller = vim.vm.device.VirtualAHCIController(key=-101, busNumber=0) + cdrom = vim.vm.device.VirtualCdrom( + key=-102, + controllerKey=-101, + unitNumber=0, + backing=vim.vm.device.VirtualCdrom.IsoBackingInfo(fileName=iso_path), + connectable=vim.vm.device.VirtualDevice.ConnectInfo( + startConnected=True, connected=True, allowGuestControl=True + ), + ) + return [ + vim.vm.device.VirtualDeviceSpec(operation="add", device=controller), + vim.vm.device.VirtualDeviceSpec(operation="add", device=cdrom), + ] diff --git a/napalm_vmware/vcenter.py b/napalm_vmware/vcenter.py index 5211a08..8de0ef0 100644 --- a/napalm_vmware/vcenter.py +++ b/napalm_vmware/vcenter.py @@ -14,9 +14,10 @@ from napalm_device_types import FingerprintRule from napalm_vmware.actions import VmwareActionsMixin from napalm_vmware.base import VmwareBaseDriver from napalm_vmware.parse.facts import vcenter_facts +from napalm_vmware.provisioning import VmwareProvisioningMixin -class VmwareVcenterDriver(VmwareActionsMixin, VmwareBaseDriver): +class VmwareVcenterDriver(VmwareProvisioningMixin, VmwareActionsMixin, VmwareBaseDriver): """A vCenter and every VM, datastore and port group it manages.""" DRIVER_NAME = "vmware_vcenter" diff --git a/pyproject.toml b/pyproject.toml index f2a3c96..ca9251a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,6 +31,10 @@ dependencies = [ "napalm>=5.0.0", "napalm-device-types>=2.0.0", "pyvmomi>=8.0.1", + # Provisioning: download the image, write the NoCloud seed ISO and its YAML. + "requests>=2.31", + "pycdlib>=1.14", + "PyYAML>=6.0", ] [project.optional-dependencies] diff --git a/tests/test_actions.py b/tests/test_actions.py index 20d9a35..48dd61a 100644 --- a/tests/test_actions.py +++ b/tests/test_actions.py @@ -132,3 +132,21 @@ class TestSnapshots: 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") diff --git a/tests/test_drivers.py b/tests/test_drivers.py index 04f3a80..250ac71 100644 --- a/tests/test_drivers.py +++ b/tests/test_drivers.py @@ -80,9 +80,22 @@ class TestIdentity: assert callable(getattr(cls, method, None)) @pytest.mark.parametrize("cls", DRIVERS) - def test_does_not_claim_provisioning(self, cls): - """netOrk offers VM creation wherever this exists; v1 cannot do it.""" - assert not hasattr(cls, "create_vm_from_cloud_init") + @pytest.mark.parametrize( + "method", + [ + "create_vm_from_cloud_init", + "destroy_vm", + "get_vm_status", + "get_network_targets", + "get_image_storages", + ], + ) + def test_implements_provisioning(self, cls, method): + assert callable(getattr(cls, method, None)) + + @pytest.mark.parametrize("cls", DRIVERS) + def test_declares_its_guest_agent(self, cls): + assert cls.GUEST_AGENT_PACKAGES == ("open-vm-tools",) class TestOptionalArgs: diff --git a/tests/test_inventory.py b/tests/test_inventory.py index 10c8bc1..7834175 100644 --- a/tests/test_inventory.py +++ b/tests/test_inventory.py @@ -119,3 +119,15 @@ def test_properties_of_a_single_object(content): def test_properties_of_a_vanished_object(content): content.propertyCollector.RetrievePropertiesEx.return_value = None assert Inventory(content).properties(vim.Task("task-7", None), ["info.state"]) == {} + + +def test_raw_properties_keep_the_pyvmomi_objects(content): + disk = vim.vm.device.VirtualDisk(key=2000, capacityInKB=1024) + content.propertyCollector.RetrievePropertiesEx.return_value = SimpleNamespace( + objects=[_content(vim.VirtualMachine("vm-1", None), {"config.hardware.device": [disk]})], + token=None, + ) + props = Inventory(content).properties( + vim.VirtualMachine("vm-1", None), ["config.hardware.device"], raw=True + ) + assert props["config.hardware.device"][0] is disk diff --git a/tests/test_provision_image.py b/tests/test_provision_image.py new file mode 100644 index 0000000..c58db40 --- /dev/null +++ b/tests/test_provision_image.py @@ -0,0 +1,107 @@ +"""Turning a catalog cloud image (qcow2/raw) into a VMDK vSphere can import.""" + +from __future__ import annotations + +import hashlib +import shutil +import subprocess + +import pytest + +from napalm_vmware.provision import image as image_mod +from napalm_vmware.provision.image import ImageCache + +URL = "https://cloud.example.org/debian-13-genericcloud-amd64.qcow2" +CONTENT = b"qcow2-bytes" +SHA = hashlib.sha256(CONTENT).hexdigest() + + +class _Fake: + def __init__(self, payloads=(CONTENT,)): + self.payloads = list(payloads) + self.fetched = 0 + self.converted = 0 + + def fetch(self, url, dest, timeout): + dest.write_bytes(self.payloads[min(self.fetched, len(self.payloads) - 1)]) + self.fetched += 1 + + def convert(self, src, dst): + dst.write_bytes(b"vmdk:" + src.read_bytes()) + self.converted += 1 + + def size(self, path): + return 2 * 1024**3 + + +def _cache(tmp_path, fake): + return ImageCache(tmp_path, fetch=fake.fetch, convert=fake.convert, virtual_size=fake.size) + + +def test_first_use_downloads_and_converts(tmp_path): + fake = _Fake() + vmdk, size = _cache(tmp_path, fake).vmdk(URL, f"sha256:{SHA}", timeout=60) + assert vmdk.read_bytes() == b"vmdk:" + CONTENT + assert size == 2 * 1024**3 + assert (fake.fetched, fake.converted) == (1, 1) + + +def test_second_use_is_served_from_the_cache(tmp_path): + fake = _Fake() + _cache(tmp_path, fake).vmdk(URL, None, timeout=60) + vmdk, size = _cache(tmp_path, fake).vmdk(URL, None, timeout=60) + assert (fake.fetched, fake.converted) == (1, 1) + assert size == 2 * 1024**3 + + +def test_the_download_is_not_kept_after_conversion(tmp_path): + _cache(tmp_path, _Fake()).vmdk(URL, None, timeout=60) + assert [p.suffix for p in tmp_path.iterdir()] and not list(tmp_path.glob("*.download")) + + +def test_a_corrupt_download_is_retried_once(tmp_path): + fake = _Fake(payloads=(b"truncated", CONTENT)) + vmdk, _ = _cache(tmp_path, fake).vmdk(URL, f"sha256:{SHA}", timeout=60) + assert fake.fetched == 2 + assert vmdk.read_bytes() == b"vmdk:" + CONTENT + + +def test_a_persistent_mismatch_is_an_error(tmp_path): + fake = _Fake(payloads=(b"wrong",)) + with pytest.raises(RuntimeError, match="Checksum mismatch"): + _cache(tmp_path, fake).vmdk(URL, f"sha256:{SHA}", timeout=60) + assert fake.converted == 0 + assert not list(tmp_path.glob("*.vmdk")) + + +def test_bare_hex_checksum_means_sha256(tmp_path): + _cache(tmp_path, _Fake()).vmdk(URL, SHA, timeout=60) + + +def test_different_urls_do_not_share_a_cache_entry(tmp_path): + fake = _Fake() + cache = _cache(tmp_path, fake) + cache.vmdk(URL, None, timeout=60) + cache.vmdk(URL + "?v=2", None, timeout=60) + assert fake.converted == 2 + + +def test_missing_qemu_img_is_explained(monkeypatch, tmp_path): + monkeypatch.setattr(image_mod.shutil, "which", lambda name: None) + with pytest.raises(RuntimeError, match="qemu-img"): + image_mod.convert_to_vmdk(tmp_path / "a", tmp_path / "b") + + +@pytest.mark.skipif(not shutil.which("qemu-img"), reason="qemu-img not installed") +def test_real_conversion_round_trip(tmp_path): + raw = tmp_path / "disk.raw" + subprocess.run( + ["qemu-img", "create", "-f", "raw", str(raw), "64M"], check=True, capture_output=True + ) + vmdk = tmp_path / "disk.vmdk" + image_mod.convert_to_vmdk(raw, vmdk) + info = subprocess.run( + ["qemu-img", "info", str(vmdk)], check=True, capture_output=True, text=True + ).stdout + assert "streamOptimized" in info + assert image_mod.virtual_size(raw) == 64 * 1024**2 diff --git a/tests/test_provision_ovf.py b/tests/test_provision_ovf.py new file mode 100644 index 0000000..eaeabda --- /dev/null +++ b/tests/test_provision_ovf.py @@ -0,0 +1,79 @@ +"""The OVF descriptor an import is built from.""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET + +import pytest + +from napalm_vmware.provision.ovf import ovf_descriptor + +NS = { + "ovf": "http://schemas.dmtf.org/ovf/envelope/1", + "rasd": "http://schemas.dmtf.org/wbem/wscim/1/cim-schema/2/CIM_ResourceAllocationSettingData", + "vmw": "http://www.vmware.com/schema/ovf", +} +OVF = "{http://schemas.dmtf.org/ovf/envelope/1}" +RASD = "{" + NS["rasd"] + "}" + + +@pytest.fixture +def root(): + return ET.fromstring( + ovf_descriptor("web01", cpu=2, memory_mb=4096, capacity_bytes=10 * 1024**3, nic_count=2) + ) + + +def _items(root): + return root.findall(".//ovf:VirtualHardwareSection/ovf:Item", NS) + + +def _by_type(root, resource_type): + return [i for i in _items(root) if i.find("rasd:ResourceType", NS).text == str(resource_type)] + + +def test_cpu_and_memory(root): + (cpu,) = _by_type(root, 3) + (mem,) = _by_type(root, 4) + assert cpu.find("rasd:VirtualQuantity", NS).text == "2" + assert mem.find("rasd:VirtualQuantity", NS).text == "4096" + + +def test_disk_references_the_stream_optimized_file(root): + disk = root.find(".//ovf:DiskSection/ovf:Disk", NS) + assert disk.get(f"{OVF}capacity") == str(10 * 1024**3) + assert disk.get(f"{OVF}format").endswith("#streamOptimized") + (file_ref,) = root.findall(".//ovf:References/ovf:File", NS) + assert disk.get(f"{OVF}fileRef") == file_ref.get(f"{OVF}id") + (item,) = _by_type(root, 17) + assert item.find("rasd:HostResource", NS).text == f"ovf:/disk/{disk.get(f'{OVF}diskId')}" + + +def test_one_vmxnet3_nic_per_network(root): + nics = _by_type(root, 10) + assert [n.find("rasd:Connection", NS).text for n in nics] == ["net0", "net1"] + assert {n.find("rasd:ResourceSubType", NS).text for n in nics} == {"VmxNet3"} + networks = root.findall(".//ovf:NetworkSection/ovf:Network", NS) + assert [n.get(f"{OVF}name") for n in networks] == ["net0", "net1"] + + +def test_rasd_elements_are_in_schema_order(root): + """vSphere rejects an Item whose CIM elements are out of order.""" + for item in _items(root): + names = [child.tag.removeprefix(RASD) for child in item] + assert names == sorted(names), names + + +def test_name_is_escaped(): + ET.fromstring(ovf_descriptor("a str: + reader = pycdlib.PyCdlib() + reader.open_fp(io.BytesIO(iso)) + out = io.BytesIO() + reader.get_file_from_iso_fp(out, rr_path=f"/{name}") + reader.close() + return out.getvalue().decode() + + +class TestIso: + def test_carries_the_three_files_under_their_cloud_init_names(self): + iso = nocloud_iso("#cloud-config\n{}\n", "instance-id: x\n", {"version": 2}) + assert _read(iso, "user-data") == "#cloud-config\n{}\n" + assert _read(iso, "meta-data") == "instance-id: x\n" + assert yaml.safe_load(_read(iso, "network-config")) == {"version": 2} + + def test_volume_label_is_cidata(self): + """cloud-init's NoCloud datasource finds the seed by this label.""" + reader = pycdlib.PyCdlib() + reader.open_fp(io.BytesIO(nocloud_iso("u", "m", None))) + assert reader.pvd.volume_identifier.decode().strip() == "cidata" + reader.close() + + def test_without_network_config(self): + reader = pycdlib.PyCdlib() + reader.open_fp(io.BytesIO(nocloud_iso("u", "m", None))) + names = [ + c.rock_ridge.name().decode() for c in reader.list_children(rr_path="/") if c.rock_ridge + ] + reader.close() + assert "network-config" not in names diff --git a/tests/test_provision_transfer.py b/tests/test_provision_transfer.py new file mode 100644 index 0000000..49943dd --- /dev/null +++ b/tests/test_provision_transfer.py @@ -0,0 +1,63 @@ +"""NFC upload body with lease progress, lease URLs and datastore URLs.""" + +from __future__ import annotations + +from napalm_vmware.provision.transfer import ProgressFile, datastore_url, lease_url + + +class _Clock: + def __init__(self): + self.now = 0.0 + + def __call__(self): + self.now += 15.0 # each read "takes" 15 s + return self.now + + +class TestProgressFile: + def test_has_a_length_so_requests_does_not_chunk(self, tmp_path): + """A generator body is sent chunked; with a Content-Length header too, + vSphere's NFC endpoint answers 500. A sized body avoids both.""" + path = tmp_path / "disk.vmdk" + path.write_bytes(b"x" * 1000) + body = ProgressFile(path, lambda pct: None) + assert len(body) == 1000 + body.close() + + def test_reads_everything_and_reports_rising_progress(self, tmp_path): + path = tmp_path / "disk.vmdk" + path.write_bytes(b"x" * (3 * 1024 * 1024 + 5)) + reports = [] + body = ProgressFile(path, reports.append, clock=_Clock()) + data = b"".join(iter(lambda: body.read(1024 * 1024), b"")) + body.close() + assert len(data) == path.stat().st_size + assert reports and reports == sorted(reports) and all(0 <= r < 100 for r in reports) + + +class TestLeaseUrl: + def test_star_becomes_the_host_on_the_default_port(self): + assert ( + lease_url("https://*/nfc/x/disk-0.vmdk", "esx01", 443) + == "https://esx01/nfc/x/disk-0.vmdk" + ) + + def test_star_carries_a_non_default_port(self): + assert ( + lease_url("https://*/nfc/x/d.vmdk", "10.0.0.5", 8443) + == "https://10.0.0.5:8443/nfc/x/d.vmdk" + ) + + def test_a_named_host_is_left_alone(self): + """Through vCenter the URL names the ESXi host that takes the disk.""" + assert ( + lease_url("https://esx02.lan/nfc/x/d.vmdk", "vc01", 443) + == "https://esx02.lan/nfc/x/d.vmdk" + ) + + +def test_datastore_url_quotes_the_path(): + assert ( + datastore_url("https://esx01:443", "web 01/cidata.iso") + == "https://esx01:443/folder/web%2001/cidata.iso" + ) diff --git a/tests/test_provisioning.py b/tests/test_provisioning.py new file mode 100644 index 0000000..941cae3 --- /dev/null +++ b/tests/test_provisioning.py @@ -0,0 +1,277 @@ +"""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 "2048" in descriptor + assert descriptor.count("10") == 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_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) + + +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) diff --git a/tests/test_vcsim.py b/tests/test_vcsim.py index 41f58fa..8a53ba2 100644 --- a/tests/test_vcsim.py +++ b/tests/test_vcsim.py @@ -103,3 +103,49 @@ def test_wrong_driver_is_refused(): 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()) -- 2.54.0