fix(vm_provision_mixin): upload Cloud-Init snippet as a real multipart file

proxmoxer only builds a multipart request for io.IOBase values passed
as kwargs; a plain filename string (plus a nonexistent "data" field,
as the old code sent) goes out as an ordinary form-urlencoded POST
instead. Real Proxmox's /storage/{s}/upload endpoint expects an actual
file upload for "filename" and responds to anything else by closing
the connection with no HTTP response at all.

Found live: the VM shell, disk import, and node-scoped storage
selection all succeeded, then create_vm_from_cloud_init failed with
requests.exceptions.ConnectionError / RemoteDisconnected right at the
snippet upload step.

napalm_proxmox/vm_provision_mixin.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import io
 import logging
 import time
 import yaml
@@ -337,11 +338,18 @@ class ProxmoxVMProvisionMixin:
             filename = f"{vmid}-user-data.yaml"
             _logger.debug(f"Uploading Cloud-Init snippet {filename} to {snippet_storage}")
 
-            # Upload to snippet storage
+            # Upload to snippet storage. Proxmox's upload endpoint expects the
+            # "filename" parameter to BE the file (multipart), not a name
+            # string with separate content — proxmoxer only builds a
+            # multipart request when the value is an io.IOBase instance,
+            # otherwise it silently sends everything as a plain
+            # form-urlencoded POST, which real Proxmox rejects by dropping
+            # the connection (RemoteDisconnected, no HTTP response at all).
+            file_obj = io.BytesIO(user_data_yaml.encode("utf-8"))
+            file_obj.name = filename
             self._node_api().storage(snippet_storage).upload.post(
                 content="snippets",
-                filename=filename,
+                filename=file_obj,
-                data=user_data_yaml,
             )
 
             # Step 7: Configure Cloud-Init references and SSH keys

tests/test_vm_provision_mixin.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import io
+
 import pytest
 from unittest.mock import MagicMock, patch
 from napalm_proxmox.driver import ProxmoxDriver
@@ -130,6 +132,15 @@ def test_create_vm_from_cloud_init_single_nic():
     assert "tag=10" in net_call_args[1]["net0"]
     assert "vmbr0" in net_call_args[1]["net0"]
 
+    # Regression: the snippet must be uploaded as an actual file (io.IOBase),
+    # not a plain filename string with a separate "data" field — proxmoxer
+    # only builds a real multipart request for io.IOBase values, and real
+    # Proxmox drops the connection outright for anything else (see
+    # test_create_vm_uploads_snippet_as_file_object for the dedicated check).
+    upload_kwargs = mock_storage.upload.post.call_args[1]
+    assert "data" not in upload_kwargs
+    assert isinstance(upload_kwargs["filename"], io.IOBase)
+
 
 def test_create_vm_from_cloud_init_dual_nic_trunk():
     """create_vm_from_cloud_init with dual NICs: net0 DHCP + net1 trunk (no DHCP)."""
@@ -883,3 +894,75 @@ def test_get_image_storages_excludes_storage_restricted_to_other_nodes():
     targets = mixin.get_image_storages()
 
     assert [t["name"] for t in targets] == ["local-zfs"]
+
+
+# ---------------------------------------------------------------------------
+# Snippet upload — must be a real multipart file, not a filename+data string pair
+# ---------------------------------------------------------------------------
+
+
+def test_create_vm_uploads_snippet_as_file_object_with_correct_content():
+    """Regression: real Proxmox's /storage/{s}/upload endpoint expects the
+    "filename" parameter to be the file itself (multipart). proxmoxer only
+    builds a multipart request when the value is an io.IOBase instance —
+    passing a plain string (with a separate, nonexistent "data" field, as the
+    old code did) makes proxmoxer send a normal form-urlencoded POST instead,
+    which real Proxmox responds to by closing the connection outright
+    (observed live: requests.exceptions.ConnectionError /
+    RemoteDisconnected('Remote end closed connection without response'),
+    after the VM shell and disk import had already succeeded)."""
+    mixin = ProxmoxVMProvisionMixin()
+    mixin._node_name = "pve1"
+
+    mock_api = MagicMock()
+    mock_api.cluster.nextid.get.return_value = 106
+
+    mock_node = MagicMock()
+    mock_node.storage.get.return_value = [
+        {"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir", "enabled": 1},
+        {"storage": "local", "type": "dir", "content": "snippets", "enabled": 1},
+    ]
+    mixin._api = mock_api
+    mixin._node_api = MagicMock(return_value=mock_node)
+    mixin._download_cloud_image = MagicMock(return_value="/var/lib/vz/template/netork-images/debian-12.qcow2")
+    mixin._run_node_command = MagicMock(return_value="")
+
+    mock_vm = MagicMock()
+    mock_node.qemu.return_value = mock_vm
+    mock_node.qemu.post.return_value = None
+    mock_vm.config.post.return_value = None
+    mock_vm.config.get.return_value = {
+        "unused0": "local-zfs:vm-106-disk-0",
+        "scsi0": "local-zfs:vm-106-disk-0",
+    }
+    mock_vm.status.start.post.return_value = "UPID:pve1:132:start"
+
+    mock_task = MagicMock()
+    mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
+    mock_node.tasks.return_value = mock_task
+
+    mock_storage = MagicMock()
+    mock_storage.upload.post.return_value = {"filename": "local:snippets/106-user-data.yaml"}
+    mock_node.storage.return_value = mock_storage
+
+    with patch("time.sleep"):
+        mixin.create_vm_from_cloud_init(
+            name="upload-shape-vm",
+            image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
+            cpu=2,
+            memory=2048,
+            nics=[{"bridge": "vmbr0"}],
+            cloud_init_config={"hostname": "upload-shape-vm", "chpasswd": {"expire": False}},
+        )
+
+    upload_call = mock_storage.upload.post.call_args
+    assert upload_call[1]["content"] == "snippets"
+    assert "data" not in upload_call[1]
+
+    file_obj = upload_call[1]["filename"]
+    assert isinstance(file_obj, io.IOBase)
+    assert file_obj.name == "106-user-data.yaml"
+    file_obj.seek(0)
+    content = file_obj.read().decode("utf-8")
+    assert content.startswith("#cloud-config\n")
+    assert "hostname: upload-shape-vm" in content