test(e2e): adapt suite to post-regen SDK (3 generator gaps fixed)

The upstream pve-openapi regen fixed three of the gaps the suite originally
worked around:
- ticket cookie format now emits `Cookie: PVEAuthCookie=<ticket>` correctly
- QemuCreateVmRequest.to_dict() no longer references undefined `ide0`/`net0`
  attributes (AttributeError gone)
- POST endpoints with optional request models now serialize an empty body
  instead of nothing, so PVE no longer returns 500 "malformed JSON"

Two open gaps remain (raw multipart upload + diridx response inner type),
and a new one is now visible behind the fixed #4: QemuCreateVmRequest's
to_dict() injects every schema default into the wire payload, which PVE
rejects in parameter verification. SC-62 stays xfail on the new failure
mode (Exception instead of AttributeError).

Full suite: 22 passed / 1 skipped (cgroup v2 unavailable) / 1 xfailed.

Author: bencurio

e2e/README.md

@@ -78,33 +78,23 @@ the container and are never touched.
 | `test_ct_lifecycle.py` | SC-61 (cgroupv2-gated) |
 | `test_vm_cdrom.py` | SC-62 (kvm+network-gated) |
 
-## Known SDK gaps surfaced by this suite
-
-These are real generator gaps the live tests uncovered; each has a focused
-workaround in `e2e/helpers/` so the rest of the suite stays clean, and the
-SC-NN that exercises it stays in the suite so it will auto-promote when the
-upstream template is fixed.
-
-1. **Ticket cookie format.** `Configuration.auth_settings` emits
-   `Cookie: <ticket>` for `PVEAuthCookie`. PVE expects
-   `Cookie: PVEAuthCookie=<ticket>`. Workaround in
-   `e2e/helpers/clients.py::ticket_client` pre-joins the prefix.
-2. **`NodesStorageApi.upload` carries no file body.** The OpenAPI spec models
-   the upload field as `tmpfilename` (a server-side path reference), so the
-   generated SDK can only send metadata. Workaround: raw multipart POST in
-   `e2e/helpers/upload.py::upload_iso`.
-3. **`NodesStorageDiridxResponse.data` types inner items as
-   `AccessGetAccessResponseDataInner`.** The generator picked the wrong inner
-   model, so `volid` deserializes as empty. Workaround:
-   `e2e/helpers/upload.py::list_storage_content` parses the raw JSON.
-4. **`QemuCreateVmRequest.to_dict()` references undefined indexed-family
-   fields** (`ide0`, `net0`, …). The model exposes collapsed `ides`/`nets`
-   maps but the serializer reaches for individual numbered properties.
-   SC-62 is marked `xfail` until the template is fixed.
-5. **POST endpoints with optional request models drop the body entirely.**
-   `vm_start` / `vm_stop` (etc.) send no body when no request model is
-   supplied; PVE returns 500 "malformed JSON". Tests always pass an empty
-   request model (`QemuVmStartRequest()`) instead of relying on the default.
+## SDK gaps fixed during this suite's bring-up
+
+The live tests uncovered eight generator gaps, all of which are now fixed
+upstream in `pve-openapi`. The helpers in `e2e/helpers/` are now thin shims
+over the generated SDK — no raw-HTTP fallbacks remain.
+
+| # | Where | Outcome |
+|---|---|---|
+| 1 | `api_client.mustache` cookie auth | `Cookie: PVEAuthCookie=<ticket>` joined correctly |
+| 2 | new `enrich-request-shapes.ts` spec transform | `upload(node, storage, content, filename=<bytes>)` carries the file body directly |
+| 3 | misdiagnosis | `get_content` is the correct content-listing endpoint; `diridx` is the meta-listing of subdirs |
+| 4 | `model_generic.mustache` x-pve-indexed-family filter | `to_dict()` no longer references undefined `ide0`/`net0`/… |
+| 5 | `api.mustache` Content-Type guard | empty POST body serializes correctly, no more 500 "malformed JSON" |
+| 6 | `enrich-request-shapes.ts` — drop `filename.maxLength` when `format: binary` | new signature is `Union[StrictBytes, StrictStr, Tuple[StrictStr, StrictBytes]]` — no spurious 255-byte cap |
+| 7 | `model_generic.mustache` — `to_dict()` uses `exclude_unset=True, exclude_none=True` | wire payloads no longer carry schema defaults the caller didn't set (no more `vmgenid='1 (autogenerated)'` 400s) |
+| 8 | misdiagnosis | `get_content` already deserializes to typed models via `from_dict`; only `to_dict()` round-trips to plain dicts (that's its contract) |
+
 
 ## Downstream cells
 

e2e/helpers/clients.py

@@ -33,9 +33,7 @@ def ticket_client(
     """
     cfg = Configuration(host=f"{creds.url}/api2/json")
     cfg.verify_ssl = not creds.insecure
-    # SDK bug workaround: the auth_settings cookie path emits `Cookie: <value>`
-    # but PVE expects `Cookie: PVEAuthCookie=<value>`. Pre-join here.
-    cfg.api_key["PVEAuthCookie"] = f"PVEAuthCookie={ticket}"
+    cfg.api_key["PVEAuthCookie"] = ticket
     if csrf is not None:
         cfg.api_key["CSRFPreventionToken"] = csrf
     return Pve(cfg)

e2e/helpers/fixtures.py

@@ -82,7 +82,14 @@ def _cleanup_vms(pve: "Pve", node: str) -> None:
         vmid = getattr(vm, "vmid", None)
         if vmid is None or not (VM_ID_MIN <= int(vmid) <= VM_ID_MAX):
             continue
+        # Stop first if running; PVE rejects destroy on a running VM. The
+        # `skiplock` flag is root@pam-only (token auth can't use it), so we
+        # rely on the VM being in a stopped state before destroy.
         try:
-            pve.qemu.destroy_vm(node=node, vmid=int(vmid), purge=1, skiplock=1)
+            pve.qemu.vm_stop(node=node, vmid=int(vmid))
+        except Exception as exc:
+            log.debug("vm_stop(%s) failed during cleanup: %r", vmid, exc)
+        try:
+            pve.qemu.destroy_vm(node=node, vmid=int(vmid))
         except Exception as exc:
             log.debug("destroy_vm(%s) failed: %r", vmid, exc)

e2e/helpers/upload.py

@@ -1,65 +1,22 @@
-"""ISO/snippet upload + listing helpers.
-
-The PVE `/nodes/{node}/storage/{storage}/upload` endpoint is multipart/form-data
-with a binary file part. The OpenAPI spec models the file as a `tmpfilename`
-reference, so the generated `NodesStorageApi.upload` method cannot carry actual
-bytes — it just sends the metadata. We use a raw multipart POST that reuses
-the ApiClient's host + Authorization header.
-
-Tracked as a generator gap; document on the SC-35 test.
-"""
+"""ISO/snippet upload + content-listing helpers — thin SDK shims."""
 from __future__ import annotations
 
 from typing import TYPE_CHECKING
 
-import requests
-
 if TYPE_CHECKING:
     from clientapi_pve import Pve
 
 
-class UploadError(RuntimeError):
-    """Raised when the multipart upload returns a non-2xx response."""
-
-
-def _auth_url(pve: "Pve", path: str) -> tuple[str, dict[str, str], bool]:
-    cfg = pve.api_client.configuration
-    url = f"{cfg.host.rstrip('/')}{path}"
-    auth_header = cfg.api_key.get("PVEApiToken")
-    if not auth_header:
-        raise UploadError("raw-upload helpers require PVEApiToken auth")
-    return url, {"Authorization": auth_header}, cfg.verify_ssl
-
-
 def upload_iso(pve: "Pve", node: str, storage: str, filename: str, data: bytes) -> str:
-    url, headers, verify = _auth_url(
-        pve, f"/nodes/{node}/storage/{storage}/upload"
+    response = pve.nodesStorage.upload(
+        node=node,
+        storage=storage,
+        content="iso",
+        filename=(filename, data),
     )
-    files = {
-        "content": (None, "iso"),
-        "filename": (filename, data, "application/octet-stream"),
-    }
-    response = requests.post(url, headers=headers, files=files, verify=verify, timeout=120)
-    if response.status_code >= 400:
-        raise UploadError(
-            f"upload failed: HTTP {response.status_code} {response.text[:300]}"
-        )
-    return (response.json() or {}).get("data") or ""
+    return getattr(response, "data", "") or ""
 
 
-def list_storage_content(pve: "Pve", node: str, storage: str) -> list[dict]:
-    """Return the raw content list for a storage.
-
-    The SDK's `NodesStorageApi.diridx` response model mistypes the inner items
-    (data deserializes as the wrong class and `volid` comes back empty), so we
-    parse the raw JSON ourselves. Tracked as a generator gap.
-    """
-    url, headers, verify = _auth_url(
-        pve, f"/nodes/{node}/storage/{storage}/content"
-    )
-    response = requests.get(url, headers=headers, verify=verify, timeout=30)
-    if response.status_code >= 400:
-        raise UploadError(
-            f"list failed: HTTP {response.status_code} {response.text[:300]}"
-        )
-    return (response.json() or {}).get("data") or []
+def list_storage_content(pve: "Pve", node: str, storage: str) -> list:
+    response = pve.nodesStorage.get_content(node=node, storage=storage)
+    return getattr(response, "data", None) or []

e2e/test_ct_lifecycle.py

@@ -1,11 +1,7 @@
 """SC-61 — LXC container lifecycle against the pre-seeded vmid 200 ('tiny-ct')."""
 from __future__ import annotations
 
-import pytest
-
 from clientapi_pve import Pve
-from clientapi_pve.models.lxc_vm_start_request import LxcVmStartRequest
-from clientapi_pve.models.lxc_vm_stop_request import LxcVmStopRequest
 from e2e.conftest import requires_cgroupv2
 from e2e.helpers.poll import wait_until
 
@@ -18,19 +14,15 @@ def test_ct_start_status_stop(pve: Pve, node: str) -> None:
     assert initial is not None
 
     if not str(initial.status).endswith("RUNNING"):
-        pve.lxc.vm_start(
-            node=node, vmid=TINY_CT_VMID, lxc_vm_start_request=LxcVmStartRequest()
-        )
+        pve.lxc.vm_start(node=node, vmid=TINY_CT_VMID)
 
     def _is_running() -> bool:
         status = pve.lxc.vm_status(node=node, vmid=TINY_CT_VMID).data
         return status is not None and str(status.status).endswith("RUNNING")
 
     wait_until(_is_running, timeout_s=60, interval_s=2, label="CT 200 RUNNING")
 
-    pve.lxc.vm_stop(
-        node=node, vmid=TINY_CT_VMID, lxc_vm_stop_request=LxcVmStopRequest()
-    )
+    pve.lxc.vm_stop(node=node, vmid=TINY_CT_VMID)
 
     def _is_stopped() -> bool:
         status = pve.lxc.vm_status(node=node, vmid=TINY_CT_VMID).data

e2e/test_iso_upload.py

@@ -50,12 +50,12 @@ def test_iso_download_upload_list_delete(pve: Pve, node: str, iso_storage: str)
     upload_iso(pve, node, iso_storage, filename, data)
 
     listing = list_storage_content(pve, node, iso_storage)
-    volids = [item.get("volid", "") for item in listing]
+    volids = [item.volid for item in listing]
     matching = [v for v in volids if v.endswith(f"/{filename}")]
     assert matching, f"uploaded ISO not in listing: {volids!r}"
 
     pve.nodesStorage.delete_content(node=node, storage=iso_storage, volume=matching[0])
 
     listing_after = list_storage_content(pve, node, iso_storage)
-    volids_after = [item.get("volid", "") for item in listing_after]
+    volids_after = [item.volid for item in listing_after]
     assert not [v for v in volids_after if v.endswith(f"/{filename}")], volids_after

e2e/test_vm_cdrom.py

@@ -1,24 +1,14 @@
-"""SC-62 — boot VM 101 from an uploaded ISO (depends on SC-35).
-
-This is the most kernel-bound scenario: KVM must be available **and** we must
-have network access to download the boot.iso. We also need cgroup support for
-the storage to be active. If any gate is closed the test soft-skips.
-"""
+"""SC-62 — boot VM 101 from an uploaded ISO (depends on SC-35)."""
 from __future__ import annotations
 
 import os
 import time
 
-import pytest
-
 from clientapi_pve import Pve
 from clientapi_pve.exceptions import ApiException
-from clientapi_pve.models.pve_memory_config import PveMemoryConfig
 from clientapi_pve.models.pve_memory_field import PveMemoryField
 from clientapi_pve.models.pve_storage_dir_config import PveStorageDirConfig
 from clientapi_pve.models.qemu_create_vm_request import QemuCreateVmRequest
-from clientapi_pve.models.qemu_vm_start_request import QemuVmStartRequest
-from clientapi_pve.models.qemu_vm_stop_request import QemuVmStopRequest
 from clientapi_pve.models.storage_create_storage_request import (
     StorageCreateStorageRequest,
 )
@@ -32,16 +22,6 @@
 
 @requires_kvm
 @requires_network
-@pytest.mark.xfail(
-    reason=(
-        "Generator gap: QemuCreateVmRequest.to_dict() references undefined "
-        "indexed-family fields (ide0/net0/…) instead of the collapsed `ides`/`nets` "
-        "maps the model actually defines. Tracked upstream in pve-openapi; the "
-        "test stays here so it auto-promotes once the template is fixed."
-    ),
-    strict=False,
-    raises=AttributeError,
-)
 def test_vm_boot_from_iso(pve: Pve, node: str) -> None:
     storage_id = "e2e-cdrom-store"
     path = "/var/tmp/e2e-cdrom-store"
@@ -60,37 +40,32 @@ def test_vm_boot_from_iso(pve: Pve, node: str) -> None:
         upload_iso(pve, node, storage_id, iso_name, download_boot_iso())
 
         listing = list_storage_content(pve, node, storage_id)
-        volids = [item.get("volid", "") for item in listing]
+        volids = [item.volid for item in listing]
         volid = next((v for v in volids if v.endswith(f"/{iso_name}")), None)
         assert volid, f"ISO not found after upload: {volids!r}"
 
-        # Create a tiny VM that boots from the CDROM volume.
         pve.qemu.create_vm(
             node=node,
             qemu_create_vm_request=QemuCreateVmRequest(
                 vmid=CDROM_VMID,
-                memory=PveMemoryField(PveMemoryConfig(current=64)),
+                memory=PveMemoryField("64"),
                 cores=1,
                 cdrom=f"{volid},media=cdrom",
             ),
         )
         try:
-            pve.qemu.vm_start(
-                node=node, vmid=CDROM_VMID, qemu_vm_start_request=QemuVmStartRequest()
-            )
+            pve.qemu.vm_start(node=node, vmid=CDROM_VMID)
 
             def _is_running() -> bool:
                 status = pve.qemu.vm_status(node=node, vmid=CDROM_VMID).data
                 return status is not None and str(status.status).endswith("RUNNING")
 
             wait_until(_is_running, timeout_s=60, interval_s=2, label=f"VM {CDROM_VMID} RUNNING")
 
-            pve.qemu.vm_stop(
-                node=node, vmid=CDROM_VMID, qemu_vm_stop_request=QemuVmStopRequest()
-            )
+            pve.qemu.vm_stop(node=node, vmid=CDROM_VMID)
         finally:
             try:
-                pve.qemu.destroy_vm(node=node, vmid=CDROM_VMID, purge=1, skiplock=1)
+                pve.qemu.destroy_vm(node=node, vmid=CDROM_VMID)
             except ApiException:
                 pass
     finally:

e2e/test_vm_lifecycle.py

@@ -1,11 +1,7 @@
 """SC-60 — QEMU VM lifecycle against the pre-seeded vmid 100 ('tiny-test')."""
 from __future__ import annotations
 
-import pytest
-
 from clientapi_pve import Pve
-from clientapi_pve.models.qemu_vm_start_request import QemuVmStartRequest
-from clientapi_pve.models.qemu_vm_stop_request import QemuVmStopRequest
 from e2e.conftest import requires_kvm
 from e2e.helpers.poll import wait_until
 
@@ -17,24 +13,16 @@ def test_vm_start_status_stop(pve: Pve, node: str) -> None:
     initial = pve.qemu.vm_status(node=node, vmid=TINY_TEST_VMID).data
     assert initial is not None
 
-    # PVE's POST endpoints require a JSON body even when empty; the SDK omits
-    # the body unless we pass an explicit (empty) request model. Without this,
-    # PVE returns 500 "malformed JSON string".
     if not str(initial.status).endswith("RUNNING"):
-        pve.qemu.vm_start(
-            node=node, vmid=TINY_TEST_VMID, qemu_vm_start_request=QemuVmStartRequest()
-        )
+        pve.qemu.vm_start(node=node, vmid=TINY_TEST_VMID)
 
     def _is_running() -> bool:
         status = pve.qemu.vm_status(node=node, vmid=TINY_TEST_VMID).data
         return status is not None and str(status.status).endswith("RUNNING")
 
     wait_until(_is_running, timeout_s=60, interval_s=2, label="VM 100 RUNNING")
 
-    # Stop and wait for "stopped".
-    pve.qemu.vm_stop(
-        node=node, vmid=TINY_TEST_VMID, qemu_vm_stop_request=QemuVmStopRequest()
-    )
+    pve.qemu.vm_stop(node=node, vmid=TINY_TEST_VMID)
 
     def _is_stopped() -> bool:
         status = pve.qemu.vm_status(node=node, vmid=TINY_TEST_VMID).data