feat(vm_provision_mixin): implement get_network_targets()
Filters _get_node_network() to bridge/OVSBridge types only (excludes physical NICs, bonds), plus SDN vnets from _get_sdn_vnets(). vlan_aware: Linux bridge reflects its bridge_vlan_aware config flag; OVS bridge always true; SDN vnet always false (VLAN already fixed by the vnet's zone/tag). 4 new tests: bridge/vnet filtering, Linux bridge vlan_aware flag, OVS bridge always vlan_aware, SDN vnet never vlan_aware. All 15 tests in the file pass.
This commit is contained in:
@@ -8,7 +8,7 @@ import yaml
|
|||||||
from typing import Any, Dict, List
|
from typing import Any, Dict, List
|
||||||
from urllib.parse import quote
|
from urllib.parse import quote
|
||||||
|
|
||||||
from napalm_device_types.models import VMProvisionResultDict, VMStatusDict
|
from napalm_device_types.models import NetworkTargetDict, VMProvisionResultDict, VMStatusDict
|
||||||
|
|
||||||
_logger = logging.getLogger(__name__)
|
_logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -387,3 +387,34 @@ class ProxmoxVMProvisionMixin:
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
_logger.exception(f"Error getting status for VM {vmid}: {e}")
|
_logger.exception(f"Error getting status for VM {vmid}: {e}")
|
||||||
raise RuntimeError(f"Failed to get VM {vmid} status: {e}")
|
raise RuntimeError(f"Failed to get VM {vmid} status: {e}")
|
||||||
|
|
||||||
|
def get_network_targets(self) -> List[NetworkTargetDict]:
|
||||||
|
"""
|
||||||
|
List selectable network targets (bridges + SDN vnets) for a new VM's NIC.
|
||||||
|
|
||||||
|
Excludes physical NICs, bonds, and other non-bridge interface types —
|
||||||
|
those are never valid ``NICConfigDict.bridge`` values on Proxmox.
|
||||||
|
"""
|
||||||
|
targets: List[NetworkTargetDict] = []
|
||||||
|
|
||||||
|
for iface in self._get_node_network():
|
||||||
|
iface_type = iface.get("type")
|
||||||
|
name = iface.get("iface", "")
|
||||||
|
if not name:
|
||||||
|
continue
|
||||||
|
|
||||||
|
if iface_type == "bridge":
|
||||||
|
vlan_aware = bool(int(iface.get("bridge_vlan_aware", 0) or 0))
|
||||||
|
targets.append({"name": name, "kind": "bridge", "vlan_aware": vlan_aware})
|
||||||
|
elif iface_type == "OVSBridge":
|
||||||
|
# OVS bridges tag per-port regardless of a dedicated "VLAN aware" setting.
|
||||||
|
targets.append({"name": name, "kind": "bridge", "vlan_aware": True})
|
||||||
|
|
||||||
|
for vnet in self._get_sdn_vnets():
|
||||||
|
name = vnet.get("vnet", "")
|
||||||
|
if not name:
|
||||||
|
continue
|
||||||
|
# A vnet's VLAN is already fixed by its zone/tag — no separate vlan_tag applies.
|
||||||
|
targets.append({"name": name, "kind": "vnet", "vlan_aware": False})
|
||||||
|
|
||||||
|
return targets
|
||||||
|
|||||||
@@ -394,3 +394,81 @@ def test_destroy_vm_already_stopped():
|
|||||||
|
|
||||||
# Verify delete was called (stop may fail or not be needed)
|
# Verify delete was called (stop may fail or not be needed)
|
||||||
assert mock_vm.delete.called
|
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)
|
||||||
|
|||||||
Reference in New Issue
Block a user