napalm-proxmox/tests/test_vm_provision_mixin.py

"""Tests for ProxmoxVMProvisionMixin — VM creation, destruction, status polling."""

from __future__ import annotations

import pytest
from unittest.mock import MagicMock, patch
from napalm_proxmox.driver import ProxmoxDriver
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin


# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------


def test_wait_for_task_success():
    """_wait_for_task succeeds when task reaches stopped status with exitstatus OK."""
    mixin = ProxmoxVMProvisionMixin()

    # Mock the _node_api() to return a mock that supports task polling
    mock_node = MagicMock()
    mock_task_status = MagicMock()
    mock_task_status.status.get.side_effect = [
        {"status": "running", "exitstatus": None},
        {"status": "stopped", "exitstatus": "OK"},
    ]
    mock_node.tasks.return_value = mock_task_status
    mixin._node_api = MagicMock(return_value=mock_node)

    # Should complete without raising
    with patch("time.sleep"):  # Speed up polling
        result = mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=30)
    assert result is None


def test_wait_for_task_error():
    """_wait_for_task raises RuntimeError when task exits with non-OK status."""
    mixin = ProxmoxVMProvisionMixin()

    mock_node = MagicMock()
    mock_task_status = MagicMock()
    mock_task_status.status.get.return_value = {"status": "stopped", "exitstatus": "FAILED"}
    mock_node.tasks.return_value = mock_task_status
    mixin._node_api = MagicMock(return_value=mock_node)

    with pytest.raises(RuntimeError, match="FAILED"):
        with patch("time.sleep"):
            mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=30)


def test_wait_for_task_timeout():
    """_wait_for_task raises RuntimeError on timeout."""
    mixin = ProxmoxVMProvisionMixin()

    mock_node = MagicMock()
    mock_task_status = MagicMock()
    mock_task_status.status.get.return_value = {"status": "running"}  # Always running
    mock_node.tasks.return_value = mock_task_status
    mixin._node_api = MagicMock(return_value=mock_node)

    with pytest.raises(RuntimeError, match="timed out"):
        with patch("napalm_proxmox.vm_provision_mixin.time.time") as mock_time:
            # Simulate time passing: return incremented values to exceed timeout quickly
            mock_time.side_effect = [0, 2, 4]  # After 2 iterations, time > timeout=1
            with patch("napalm_proxmox.vm_provision_mixin.time.sleep"):
                mixin._wait_for_task("UPID:pve1:123:456:789:clone:100:root@pam:", timeout=1)


def test_create_vm_from_cloud_init_single_nic():
    """create_vm_from_cloud_init with single NIC, DHCP enabled (generic happy path)."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"

    # Mock API hierarchy
    mock_api = MagicMock()
    mock_api.cluster.nextid.get.return_value = 101

    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
        {"storage": "snippets", "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 operations
    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-lvm:vm-101-disk-0",
        "scsi0": "local-lvm:vm-101-disk-0",
    }
    mock_vm.status.start.post.return_value = "UPID:pve1:124:start"

    # Mock task completion
    mock_task = MagicMock()
    mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
    mock_node.tasks.return_value = mock_task

    # Mock storage upload
    mock_storage = MagicMock()
    mock_storage.upload.post.return_value = {"filename": "snippets:snippets/101-user-data.yaml"}
    mock_node.storage.return_value = mock_storage

    with patch("time.sleep"):
        result = mixin.create_vm_from_cloud_init(
            name="test-vm",
            image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
            cpu=2,
            memory=2048,
            nics=[{"bridge": "vmbr0", "vlan_tag": 10}],
            cloud_init_config={"hostname": "test-vm"},
            timeout=120,
        )

    assert result["vmid"] == "101"
    assert result["name"] == "test-vm"
    assert result["node"] == "pve1"
    assert mock_node.qemu.post.called
    mixin._download_cloud_image.assert_called_once()

    # Verify NIC config was set correctly: net0 with tag=10, DHCP enabled
    net_call_args = next(
        c for c in mock_vm.config.post.call_args_list if "net0" in c[1]
    )
    assert "tag=10" in net_call_args[1]["net0"]
    assert "vmbr0" in net_call_args[1]["net0"]


def test_create_vm_from_cloud_init_dual_nic_trunk():
    """create_vm_from_cloud_init with dual NICs: net0 DHCP + net1 trunk (no DHCP)."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"

    # Mock API hierarchy
    mock_api = MagicMock()
    mock_api.cluster.nextid.get.return_value = 102

    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
        {"storage": "snippets", "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 operations
    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-lvm:vm-102-disk-0",
        "scsi0": "local-lvm:vm-102-disk-0",
    }
    mock_vm.status.start.post.return_value = "UPID:pve1:126:start"

    # Mock task completion
    mock_task = MagicMock()
    mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
    mock_node.tasks.return_value = mock_task

    # Mock storage upload
    mock_storage = MagicMock()
    mock_storage.upload.post.return_value = {"filename": "snippets:snippets/102-user-data.yaml"}
    mock_node.storage.return_value = mock_storage

    with patch("time.sleep"):
        result = mixin.create_vm_from_cloud_init(
            name="wireshark-sat-1",
            image_url="https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
            cpu=4,
            memory=4096,
            nics=[
                {"bridge": "vmbr0", "vlan_tag": 10},  # net0: mgmt with DHCP
                {"bridge": "vmbr1", "trunk_vlan_tags": [20, 30], "dhcp": False},  # net1: trunk, no DHCP
            ],
            cloud_init_config={"hostname": "sat-1", "runcmd": ["custom cmd"]},
            timeout=120,
        )

    assert result["vmid"] == "102"
    assert result["name"] == "wireshark-sat-1"

    # Verify both NICs configured
    net_call_args = next(
        c for c in mock_vm.config.post.call_args_list if "net0" in c[1]
    )
    assert "net1" in net_call_args[1]
    assert "tag=10" in net_call_args[1]["net0"]
    assert "trunks=20;30" in net_call_args[1]["net1"]
    assert "vmbr1" in net_call_args[1]["net1"]

    # Verify DHCP config: ipconfig0 yes, ipconfig1 no
    cloud_init_call_args = next(
        c for c in mock_vm.config.post.call_args_list if "ipconfig0" in c[1]
    )
    assert cloud_init_call_args[1]["ipconfig0"] == "ip=dhcp"
    assert "ipconfig1" not in cloud_init_call_args[1]  # net1 has no DHCP


def test_create_vm_missing_snippet_storage():
    """create_vm_from_cloud_init raises ValueError if snippet storage unavailable."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"

    # Mock API with images storage but no snippet storage
    mock_api = MagicMock()
    mock_api.cluster.nextid.get.return_value = 101

    mixin._api = mock_api

    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1}
    ]
    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-lvm:vm-101-disk-0"}

    # Mock task to allow clone to complete
    mock_task = MagicMock()
    mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
    mock_node.tasks.return_value = mock_task

    with pytest.raises(ValueError, match="snippet"):
        with patch("napalm_proxmox.vm_provision_mixin.time.sleep"):
            mixin.create_vm_from_cloud_init(
                name="test-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={},
            )


def test_create_vm_with_disk_resize():
    """create_vm_from_cloud_init resizes disk when disk_resize_gb is set."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"

    # Mock API hierarchy
    mock_api = MagicMock()
    mock_api.cluster.nextid.get.return_value = 103

    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
        {"storage": "snippets", "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 operations
    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-lvm:vm-103-disk-0",
        "scsi0": "local-lvm:vm-103-disk-0",
    }
    mock_vm.resize.put.return_value = None
    mock_vm.status.start.post.return_value = "UPID:pve1:128:start"

    # Mock task completion
    mock_task = MagicMock()
    mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
    mock_node.tasks.return_value = mock_task

    # Mock storage upload
    mock_storage = MagicMock()
    mock_storage.upload.post.return_value = {"filename": "snippets:snippets/103-user-data.yaml"}
    mock_node.storage.return_value = mock_storage

    with patch("time.sleep"):
        result = mixin.create_vm_from_cloud_init(
            name="big-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": "big-vm"},
            disk_resize_gb=100,
            timeout=120,
        )

    assert result["vmid"] == "103"

    # Verify resize was called
    mock_vm.resize.put.assert_called_once()
    call_kwargs = mock_vm.resize.put.call_args[1]
    assert call_kwargs["disk"] == "scsi0"
    assert call_kwargs["size"] == "100G"


def test_get_vm_status_with_wait_for_ip():
    """get_vm_status(wait_for_ip=True) polls guest-agent until IP acquired."""
    mixin = ProxmoxVMProvisionMixin()

    mock_node = MagicMock()
    mixin._node_api = MagicMock(return_value=mock_node)

    # Mock VM config
    mock_vm = MagicMock()
    mock_node.qemu.return_value = mock_vm
    mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0,tag=10"}

    # Mock guest-agent: first no IP, then with IP
    mock_agent = MagicMock()
    mock_agent.network_get_interfaces.get.side_effect = [
        {"result": [{"name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00"}]},
        {
            "result": [
                {
                    "name": "eth0",
                    "hardware-address": "aa:bb:cc:dd:ee:00",
                    "ip-addresses": [{"ip-address": "10.0.0.100", "ip-address-type": "ipv4"}],
                }
            ]
        },
    ]
    mock_vm.agent = mock_agent

    with patch("time.sleep"):
        result = mixin.get_vm_status("101", wait_for_ip=True, timeout=30, poll_interval=1)

    assert result["status"] == "running"
    assert result["ip_address"] == "10.0.0.100"
    assert result["mac_address"] == "aa:bb:cc:dd:ee:00"


def test_get_vm_status_timeout_waiting_for_ip():
    """get_vm_status raises RuntimeError if IP acquisition times out."""
    mixin = ProxmoxVMProvisionMixin()

    mock_node = MagicMock()
    mixin._node_api = MagicMock(return_value=mock_node)

    # Mock VM config
    mock_vm = MagicMock()
    mock_node.qemu.return_value = mock_vm
    mock_vm.config.get.return_value = {"net0": "virtio,bridge=vmbr0"}

    # Mock guest-agent that never returns IP
    mock_agent = MagicMock()
    mock_agent.network_get_interfaces.get.return_value = {
        "result": [{"name": "eth0", "hardware-address": "aa:bb:cc:dd:ee:00"}]
    }
    mock_vm.agent = mock_agent

    with pytest.raises(RuntimeError, match="timeout|IP"):
        with patch("time.sleep"):
            mixin.get_vm_status("101", wait_for_ip=True, timeout=1, poll_interval=0.5)


def test_destroy_vm_success():
    """destroy_vm stops running VM, deletes it, cleans up snippets."""
    mixin = ProxmoxVMProvisionMixin()

    mock_node = MagicMock()
    mixin._node_api = MagicMock(return_value=mock_node)

    # Mock VM operations
    mock_vm = MagicMock()
    mock_node.qemu.return_value = mock_vm
    mock_vm.config.get.return_value = {"cicustom": "user=snippets:snippets/101-user-data.yaml"}
    mock_vm.status.stop.post.return_value = "UPID:pve1:125:stop"
    mock_vm.delete.return_value = None

    # Mock task completion
    mock_task = MagicMock()
    mock_task.status.get.return_value = {"status": "stopped", "exitstatus": "OK"}
    mock_node.tasks.return_value = mock_task

    # Mock snippet deletion (best-effort)
    mock_storage = MagicMock()
    mock_node.storage.return_value = mock_storage
    mock_content = MagicMock()
    mock_storage.content.return_value = mock_content
    mock_content.delete.return_value = None

    with patch("time.sleep"):
        mixin.destroy_vm("101", remove_disk=True, timeout=60)

    assert mock_vm.delete.called


def test_destroy_vm_already_stopped():
    """destroy_vm succeeds even if VM already stopped."""
    mixin = ProxmoxVMProvisionMixin()

    mock_node = MagicMock()
    mixin._node_api = MagicMock(return_value=mock_node)

    # Mock VM operations
    mock_vm = MagicMock()
    mock_node.qemu.return_value = mock_vm
    mock_vm.config.get.return_value = {}
    mock_vm.delete.return_value = None

    with patch("time.sleep"):
        mixin.destroy_vm("101", remove_disk=True, timeout=60)

    # Verify delete was called (stop may fail or not be needed)
    assert mock_vm.delete.called


# ---------------------------------------------------------------------------
# get_network_targets
# ---------------------------------------------------------------------------


def test_get_network_targets_filters_to_bridges_and_vnets():
    """get_network_targets excludes physical NICs/bonds; includes bridges + vnets."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._get_node_network = MagicMock(
        return_value=[
            {"iface": "eth0", "type": "eth"},
            {"iface": "bond0", "type": "bond"},
            {"iface": "vmbr0", "type": "bridge"},
            {"iface": "vmbr1", "type": "OVSBridge"},
        ]
    )
    mixin._get_sdn_vnets = MagicMock(
        return_value=[{"vnet": "vnet0", "zone": "zone-vlan", "tag": 10}]
    )

    targets = mixin.get_network_targets()
    names = {t["name"] for t in targets}

    assert names == {"vmbr0", "vmbr1", "vnet0"}
    assert "eth0" not in names
    assert "bond0" not in names


def test_get_network_targets_linux_bridge_vlan_aware_flag():
    """A Linux bridge's vlan_aware reflects its bridge_vlan_aware config flag."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._get_node_network = MagicMock(
        return_value=[
            {"iface": "vmbr0", "type": "bridge", "bridge_vlan_aware": 1},
            {"iface": "vmbr2", "type": "bridge"},  # no flag -> not vlan aware
        ]
    )
    mixin._get_sdn_vnets = MagicMock(return_value=[])

    targets = {t["name"]: t for t in mixin.get_network_targets()}

    assert targets["vmbr0"]["kind"] == "bridge"
    assert targets["vmbr0"]["vlan_aware"] is True
    assert targets["vmbr2"]["vlan_aware"] is False


def test_get_network_targets_ovs_bridge_always_vlan_aware():
    """An OVS bridge is always vlan_aware, regardless of bridge_vlan_aware."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._get_node_network = MagicMock(
        return_value=[{"iface": "vmbr1", "type": "OVSBridge"}]
    )
    mixin._get_sdn_vnets = MagicMock(return_value=[])

    targets = mixin.get_network_targets()

    assert targets[0]["kind"] == "bridge"
    assert targets[0]["vlan_aware"] is True


def test_get_network_targets_sdn_vnet_never_vlan_aware():
    """An SDN vnet is never vlan_aware — its VLAN is already fixed by zone/tag."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._get_node_network = MagicMock(return_value=[])
    mixin._get_sdn_vnets = MagicMock(
        return_value=[
            {"vnet": "vnet0", "zone": "zone-vlan", "tag": 10},
            {"vnet": "vnet1", "zone": "zone-vlan", "tag": 20},
        ]
    )

    targets = mixin.get_network_targets()

    assert len(targets) == 2
    assert all(t["kind"] == "vnet" for t in targets)
    assert all(t["vlan_aware"] is False for t in targets)


def test_get_network_targets_vnet_exposes_fixed_vlan_tag():
    """A vnet's fixed_vlan_tag surfaces its SDN tag for display purposes."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._get_node_network = MagicMock(return_value=[])
    mixin._get_sdn_vnets = MagicMock(
        return_value=[{"vnet": "vnet0", "zone": "zone-vlan", "tag": 10}]
    )

    targets = mixin.get_network_targets()

    assert targets[0]["fixed_vlan_tag"] == 10


def test_get_network_targets_vnet_without_tag_has_none_fixed_vlan_tag():
    """A vnet with no tag (e.g. VXLAN/EVPN zone) reports fixed_vlan_tag=None."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._get_node_network = MagicMock(return_value=[])
    mixin._get_sdn_vnets = MagicMock(
        return_value=[{"vnet": "vnet2", "zone": "zone-vxlan"}]
    )

    targets = mixin.get_network_targets()

    assert targets[0]["fixed_vlan_tag"] is None


def test_get_network_targets_bridge_has_no_fixed_vlan_tag():
    """Bridges never carry a single fixed VLAN tag — only vnets do."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._get_node_network = MagicMock(
        return_value=[{"iface": "vmbr0", "type": "bridge", "bridge_vlan_aware": 1}]
    )
    mixin._get_sdn_vnets = MagicMock(return_value=[])

    targets = mixin.get_network_targets()

    assert "fixed_vlan_tag" not in targets[0] or targets[0]["fixed_vlan_tag"] is None


# ---------------------------------------------------------------------------
# _run_node_command
# ---------------------------------------------------------------------------


def _mock_ssh_exec(exit_status: int, stdout_text: str = "", stderr_text: str = ""):
    """Build a mock (stdin, stdout, stderr) tuple as returned by exec_command."""
    mock_stdin = MagicMock()
    mock_stdout = MagicMock()
    mock_stderr = MagicMock()
    mock_stdout.channel.recv_exit_status.return_value = exit_status
    mock_stdout.read.return_value = stdout_text.encode()
    mock_stderr.read.return_value = stderr_text.encode()
    return mock_stdin, mock_stdout, mock_stderr


def test_run_node_command_success_returns_stdout():
    mixin = ProxmoxVMProvisionMixin()
    mixin._ssh_client = MagicMock()
    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(0, stdout_text="hello\n")

    result = mixin._run_node_command("echo hello", timeout=10)

    assert result == "hello"


def test_run_node_command_nonzero_exit_raises():
    mixin = ProxmoxVMProvisionMixin()
    mixin._ssh_client = MagicMock()
    mixin._ssh_client.exec_command.return_value = _mock_ssh_exec(
        1, stderr_text="No such file or directory"
    )

    with pytest.raises(RuntimeError, match="No such file or directory"):
        mixin._run_node_command("cat /nonexistent", timeout=10)


# ---------------------------------------------------------------------------
# _download_cloud_image
# ---------------------------------------------------------------------------


def test_download_cloud_image_skips_download_when_cached():
    mixin = ProxmoxVMProvisionMixin()
    mixin._run_node_command = MagicMock(return_value="EXISTS")

    path = mixin._download_cloud_image(
        "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
        None,
        timeout=300,
    )

    assert path.endswith("debian-12-genericcloud-amd64.qcow2")
    # Only the existence check ran — no wget call
    assert mixin._run_node_command.call_count == 1
    assert "wget" not in mixin._run_node_command.call_args_list[0][0][0]


def test_download_cloud_image_downloads_when_missing():
    mixin = ProxmoxVMProvisionMixin()
    mixin._run_node_command = MagicMock(return_value="MISSING")

    mixin._download_cloud_image(
        "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
        None,
        timeout=300,
    )

    commands = [c[0][0] for c in mixin._run_node_command.call_args_list]
    assert any("wget" in cmd for cmd in commands)


def test_download_cloud_image_verifies_matching_checksum():
    mixin = ProxmoxVMProvisionMixin()
    mixin._run_node_command = MagicMock(
        side_effect=["EXISTS", "abc123"]  # existence check, then checksum
    )

    path = mixin._download_cloud_image(
        "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
        "sha256:abc123",
        timeout=300,
    )

    assert path.endswith("debian-12-genericcloud-amd64.qcow2")


def test_download_cloud_image_checksum_mismatch_raises_and_removes_file():
    mixin = ProxmoxVMProvisionMixin()
    mixin._run_node_command = MagicMock(
        side_effect=["EXISTS", "wrong-checksum", ""]  # existence, checksum, rm
    )

    with pytest.raises(RuntimeError, match="Checksum mismatch"):
        mixin._download_cloud_image(
            "https://cloud.debian.org/images/cloud/bookworm/latest/debian-12-genericcloud-amd64.qcow2",
            "sha256:abc123",
            timeout=300,
        )

    commands = [c[0][0] for c in mixin._run_node_command.call_args_list]
    assert any(cmd.startswith("rm -f") for cmd in commands)


# ---------------------------------------------------------------------------
# _find_default_image_storage
# ---------------------------------------------------------------------------


def test_find_default_image_storage_treats_absent_enabled_as_enabled():
    """Proxmox omits "enabled" entirely for storages never explicitly toggled —
    absent must mean enabled, not disabled. Regression: this was previously
    treated as falsy, causing every real Proxmox server to report no usable
    image storage even when several existed."""
    mixin = ProxmoxVMProvisionMixin()
    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir"},
    ]
    mixin._node_api = MagicMock(return_value=mock_node)

    storage = mixin._find_default_image_storage()

    assert storage == "local-lvm"


def test_find_default_image_storage_excludes_explicitly_disabled():
    mixin = ProxmoxVMProvisionMixin()
    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "old-storage", "type": "dir", "content": "images", "enabled": 0},
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
    ]
    mixin._node_api = MagicMock(return_value=mock_node)

    storage = mixin._find_default_image_storage()

    assert storage == "local-lvm"


def test_find_default_image_storage_raises_when_none_found():
    mixin = ProxmoxVMProvisionMixin()
    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "local", "type": "dir", "content": "backup,iso,vztmpl"},
    ]
    mixin._node_api = MagicMock(return_value=mock_node)

    with pytest.raises(ValueError, match="images"):
        mixin._find_default_image_storage()


def test_find_default_image_storage_excludes_storage_restricted_to_other_nodes():
    """Regression: a storage configured cluster-wide but restricted via 'nodes'
    to other cluster members must not be picked — /nodes/{node}/storage (unlike
    the cluster-wide /storage endpoint) only lists what's actually available on
    this node, so querying it is what makes this exclusion happen naturally.
    This was the real-world bug: local-lvm (nodes=pve-garden) was picked for a
    VM on pve-02, and 'qm importdisk' failed with 'storage not available on
    node', leaving a VM shell with no disk attached."""
    mixin = ProxmoxVMProvisionMixin()
    mock_node = MagicMock()
    # /nodes/pve-02/storage never even lists local-lvm — Proxmox itself filters
    # node-restricted storages out of this endpoint.
    mock_node.storage.get.return_value = [
        {"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir"},
    ]
    mixin._node_api = MagicMock(return_value=mock_node)

    storage = mixin._find_default_image_storage()

    assert storage == "local-zfs"


def test_create_vm_from_cloud_init_with_real_world_storage_shape():
    """Regression: real Proxmox servers omit "enabled" for never-toggled storages
    (observed live: local-lvm/local-zfs/fast-zfs all had content=images but no
    "enabled" key at all) — this used to make create_vm_from_cloud_init fail
    with "No storage with content='images' found" even though usable storage
    existed."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"

    mock_api = MagicMock()
    mock_api.cluster.nextid.get.return_value = 104

    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir"},
        {"storage": "local-zfs", "type": "zfspool", "content": "rootdir,images"},
        {"storage": "local", "type": "dir", "content": "backup,iso,vztmpl"},
        {"storage": "snippets", "type": "dir", "content": "snippets"},
    ]
    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-lvm:vm-104-disk-0",
        "scsi0": "local-lvm:vm-104-disk-0",
    }
    mock_vm.status.start.post.return_value = "UPID:pve1:130: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": "snippets:snippets/104-user-data.yaml"}
    mock_node.storage.return_value = mock_storage

    with patch("time.sleep"):
        result = mixin.create_vm_from_cloud_init(
            name="real-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": "real-shape-vm"},
        )

    assert result["vmid"] == "104"


def test_create_vm_from_cloud_init_explicit_storage_skips_auto_detect():
    """When storage= is given explicitly, it's used directly and
    _find_default_image_storage is never consulted (so an explicit choice
    always wins, even if auto-detect would have picked something else)."""
    mixin = ProxmoxVMProvisionMixin()
    mixin._node_name = "pve1"

    mock_api = MagicMock()
    mock_api.cluster.nextid.get.return_value = 105

    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "local-lvm", "type": "lvmthin", "content": "images,rootdir", "enabled": 1},
        {"storage": "fast-zfs", "type": "zfspool", "content": "images,rootdir", "enabled": 1},
        {"storage": "snippets", "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._find_default_image_storage = MagicMock(side_effect=AssertionError("should not be called"))
    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": "fast-zfs:vm-105-disk-0",
        "scsi0": "fast-zfs:vm-105-disk-0",
    }
    mock_vm.status.start.post.return_value = "UPID:pve1:131: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": "snippets:snippets/105-user-data.yaml"}
    mock_node.storage.return_value = mock_storage

    with patch("time.sleep"):
        result = mixin.create_vm_from_cloud_init(
            name="explicit-storage-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": "explicit-storage-vm"},
            storage="fast-zfs",
        )

    assert result["vmid"] == "105"
    mixin._find_default_image_storage.assert_not_called()
    import_cmd = next(
        c[0][0] for c in mixin._run_node_command.call_args_list if "qm importdisk" in c[0][0]
    )
    assert "fast-zfs" in import_cmd
    assert "local-lvm" not in import_cmd


# ---------------------------------------------------------------------------
# get_image_storages
# ---------------------------------------------------------------------------


def test_get_image_storages_filters_to_images_content():
    mixin = ProxmoxVMProvisionMixin()
    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {
            "storage": "local-zfs",
            "type": "zfspool",
            "content": "images,rootdir",
            "total": 100 * 1024**3,
            "avail": 40 * 1024**3,
        },
        {"storage": "local", "type": "dir", "content": "backup,iso,vztmpl,snippets"},
    ]
    mixin._node_api = MagicMock(return_value=mock_node)

    targets = mixin.get_image_storages()

    assert len(targets) == 1
    assert targets[0]["name"] == "local-zfs"
    assert targets[0]["type"] == "zfspool"
    assert targets[0]["total_gb"] == 100.0
    assert targets[0]["available_gb"] == 40.0


def test_get_image_storages_excludes_disabled_and_inactive():
    mixin = ProxmoxVMProvisionMixin()
    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "disabled-store", "type": "dir", "content": "images", "enabled": 0},
        {"storage": "inactive-store", "type": "nfs", "content": "images", "active": 0},
        {"storage": "ok-store", "type": "dir", "content": "images"},
    ]
    mixin._node_api = MagicMock(return_value=mock_node)

    targets = mixin.get_image_storages()

    assert [t["name"] for t in targets] == ["ok-store"]


def test_get_image_storages_excludes_storage_restricted_to_other_nodes():
    """Node-scoped query naturally excludes storages Proxmox itself doesn't
    list for this node (e.g. restricted via 'nodes' to other cluster members)."""
    mixin = ProxmoxVMProvisionMixin()
    mock_node = MagicMock()
    mock_node.storage.get.return_value = [
        {"storage": "local-zfs", "type": "zfspool", "content": "images,rootdir"},
    ]
    mixin._node_api = MagicMock(return_value=mock_node)

    targets = mixin.get_image_storages()

    assert [t["name"] for t in targets] == ["local-zfs"]