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,34 @@ 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 originally uncovered five generator gaps; all five are now
+fixed upstream in `pve-openapi`:
+
+| # | 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" |
+
+The helpers in `e2e/helpers/` are now thin shims over the generated SDK.
+
+### Remaining edges
+
+- **Upload `max_length=255` mis-applied to file body.** The
+  `enrich-request-shapes.ts` transform leaves the original `filename`'s
+  `MaxLen(255)` constraint on the bytes element of the `Tuple[str, bytes]`
+  form — meaning the SDK rejects any file bigger than 255 bytes at the
+  Pydantic layer. `e2e/helpers/upload.py::upload_iso` keeps a raw multipart
+  fallback until the constraint is moved off the body.
+- **`QemuCreateVmRequest.to_dict()` schema-defaults injection.** Independent
+  of the indexed-family fix, `to_dict()` still emits every field's schema
+  default (`vmgenid='1 (autogenerated)'`, `vcpus=0`, `cpulimit=0`, …) and
+  PVE rejects the call in parameter verification. SC-62 stays `xfail` until
+  `to_dict()` honours `exclude_unset`.
 
 ## 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/upload.py

@@ -1,12 +1,13 @@
-"""ISO/snippet upload + listing helpers.
+"""ISO/snippet upload + content-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.
+The upload helper falls back to a raw multipart POST: the SDK's generated
+`NodesStorageApi.upload` signature wrongly applies the original `filename`'s
+`max_length=255` validator to the **bytes content** in the `Tuple[str, bytes]`
+form, so files larger than 255 bytes are rejected at the Pydantic layer.
 
-Tracked as a generator gap; document on the SC-35 test.
+The listing helper goes through the SDK's `get_content` but normalizes each
+entry: the current generator declares `data: List[Inner]` but actually
+deserializes inner items as plain dicts.
 """
 from __future__ import annotations
 
@@ -22,44 +23,36 @@ 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]:
+def upload_iso(pve: "Pve", node: str, storage: str, filename: str, data: bytes) -> str:
     cfg = pve.api_client.configuration
-    url = f"{cfg.host.rstrip('/')}{path}"
+    url = f"{cfg.host.rstrip('/')}/nodes/{node}/storage/{storage}/upload"
     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"
+        raise UploadError("upload_iso requires PVEApiToken auth on the client")
+    response = requests.post(
+        url,
+        headers={"Authorization": auth_header},
+        files={
+            "content": (None, "iso"),
+            "filename": (filename, data, "application/octet-stream"),
+        },
+        verify=cfg.verify_ssl,
+        timeout=120,
     )
-    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 ""
 
 
-def list_storage_content(pve: "Pve", node: str, storage: str) -> list[dict]:
-    """Return the raw content list for a storage.
+def _item_volid(item) -> str:
+    """Extract `volid` from a listing entry that may be a dict or a model."""
+    if isinstance(item, dict):
+        return item.get("volid") or ""
+    return getattr(item, "volid", "") or ""
 
-    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

@@ -14,7 +14,7 @@
 )
 from e2e.conftest import requires_network
 from e2e.helpers.iso import download_boot_iso
-from e2e.helpers.upload import list_storage_content, upload_iso
+from e2e.helpers.upload import _item_volid, list_storage_content, upload_iso
 
 
 @pytest.fixture
@@ -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(item) 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(item) 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,31 +1,24 @@
-"""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,
 )
+import pytest
+
 from e2e.conftest import requires_kvm, requires_network
 from e2e.helpers.iso import download_boot_iso
 from e2e.helpers.poll import wait_until
-from e2e.helpers.upload import list_storage_content, upload_iso
+from e2e.helpers.upload import _item_volid, list_storage_content, upload_iso
 
 CDROM_VMID = 101
 
@@ -34,13 +27,13 @@
 @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."
+        "Generator gap: QemuCreateVmRequest.to_dict() still injects every "
+        "schema default into the wire payload (vmgenid='1 (autogenerated)', "
+        "vcpus=0, cpulimit=0, …); PVE rejects them in parameter verification. "
+        "Independent of the indexed-family fix in #4 — tracked separately."
     ),
     strict=False,
-    raises=AttributeError,
+    raises=Exception,
 )
 def test_vm_boot_from_iso(pve: Pve, node: str) -> None:
     storage_id = "e2e-cdrom-store"
@@ -60,11 +53,10 @@ 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(item) 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(
@@ -75,19 +67,15 @@ def test_vm_boot_from_iso(pve: Pve, node: str) -> None:
             ),
         )
         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)

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