Christian Manivong
abce85d6ef
_resolve_node() took the first entry of GET /nodes. In a cluster that lists every member, so a node polled without an explicit `node` driver argument talked to whichever member came first: pve-dual reported pve-02's name and VMs, and netOrk's VM sync moved pve-02's VM devices over to it. Resolve through GET /cluster/status instead: the entry marked local, then a match by IP or (short) name, then the sole node of a standalone host, and otherwise raise rather than guess. The lookup no longer swallows API errors either. A TLS verification failure used to leave the IP as the node name, so open() succeeded and every getter failed quietly while the poll reported success with empty data. It now surfaces as a ConnectionException from open(). Refs NetOrk/netork#417, NetOrk/netork#418
223 lines
6.0 KiB
Python
223 lines
6.0 KiB
Python
"""Shared fixtures and mock helpers for napalm-proxmox tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from napalm_proxmox.driver import ProxmoxDriver
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Sample API payloads
|
|
# ---------------------------------------------------------------------------
|
|
|
|
NODE_STATUS = {
|
|
"uptime": 123456,
|
|
"model": "PowerEdge R640",
|
|
"cpu": 0.12,
|
|
"memory": {"total": 137438953472, "used": 68719476736},
|
|
}
|
|
|
|
VERSION_INFO = {"version": "8.2.4", "release": "8", "repoid": "abc123"}
|
|
|
|
NODE_NETWORK = [
|
|
{
|
|
"iface": "eth0",
|
|
"type": "eth",
|
|
"active": 1,
|
|
"autostart": 1,
|
|
"hwaddr": "aa:bb:cc:dd:ee:01",
|
|
"speed": 1000,
|
|
"mtu": 1500,
|
|
"address": "10.0.0.1",
|
|
"netmask": "255.255.255.0",
|
|
"comments": "Management",
|
|
},
|
|
{
|
|
"iface": "eth1",
|
|
"type": "eth",
|
|
"active": 1,
|
|
"autostart": 1,
|
|
"hwaddr": "aa:bb:cc:dd:ee:02",
|
|
"speed": 10000,
|
|
"mtu": 9000,
|
|
"comments": "",
|
|
},
|
|
{
|
|
"iface": "vmbr0",
|
|
"type": "bridge",
|
|
"active": 1,
|
|
"autostart": 1,
|
|
"hwaddr": "aa:bb:cc:dd:ee:03",
|
|
"speed": None,
|
|
"mtu": 1500,
|
|
"address": "192.168.1.1",
|
|
"netmask": "255.255.255.0",
|
|
"comments": "Linux bridge",
|
|
},
|
|
{
|
|
"iface": "vmbr1",
|
|
"type": "OVSBridge",
|
|
"active": 0,
|
|
"autostart": 1,
|
|
"hwaddr": "",
|
|
"speed": None,
|
|
"mtu": 1500,
|
|
"comments": "OVS bridge",
|
|
},
|
|
]
|
|
|
|
SDN_ZONES = [
|
|
{"zone": "zone-vlan", "type": "vlan", "nodes": "pve1"},
|
|
{"zone": "zone-vxlan", "type": "vxlan", "nodes": "pve1,pve2", "vrf-vxlan": 10000},
|
|
{"zone": "zone-evpn", "type": "evpn", "nodes": "pve1,pve2", "vrf-vxlan": 10001},
|
|
]
|
|
|
|
SDN_VNETS = [
|
|
{"vnet": "vnet0", "zone": "zone-vlan", "tag": 10},
|
|
{"vnet": "vnet1", "zone": "zone-vlan", "tag": 20},
|
|
{"vnet": "vnet2", "zone": "zone-vxlan", "tag": 100000},
|
|
]
|
|
|
|
SDN_SUBNETS_VNET0 = [
|
|
{"cidr": "10.10.0.0/24", "gateway": "10.10.0.1", "vnet": "vnet0"},
|
|
]
|
|
|
|
SDN_SUBNETS_VNET1 = [
|
|
{"cidr": "10.20.0.0/24", "gateway": "10.20.0.1", "vnet": "vnet1"},
|
|
{"cidr": "2001:db8:20::/64", "gateway": "2001:db8:20::1", "vnet": "vnet1"},
|
|
]
|
|
|
|
SDN_SUBNETS_VNET2: list = []
|
|
|
|
# A DNS *search domain*, not an FQDN. It used to read "pve1.example.com",
|
|
# which made get_facts build "pve1.pve1.example.com" and looked like a
|
|
# driver bug rather than bad test data.
|
|
DNS_INFO = {"search": "example.com", "dns1": "8.8.8.8"}
|
|
|
|
NTP_INFO = {"server": "pool.ntp.org,time.cloudflare.com"}
|
|
|
|
NODE_NETSTAT = [
|
|
{
|
|
"dev": "eth0",
|
|
"rx_bytes": 1000000,
|
|
"tx_bytes": 500000,
|
|
"rx_packets": 1000,
|
|
"tx_packets": 900,
|
|
"rx_errs": 0,
|
|
"tx_errs": 0,
|
|
"rx_drop": 0,
|
|
"tx_drop": 1,
|
|
}
|
|
]
|
|
|
|
NODES_LIST = [{"node": "pve1", "status": "online"}]
|
|
|
|
PVE_USERS = [
|
|
{"userid": "root@pam", "groups": ""},
|
|
{"userid": "napalm@pam", "groups": ""},
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Driver fixture
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _build_mock_api(
|
|
nodes=None,
|
|
node_status=None,
|
|
version=None,
|
|
network=None,
|
|
sdn_zones=None,
|
|
sdn_vnets=None,
|
|
sdn_subnets=None,
|
|
dns=None,
|
|
ntp=None,
|
|
netstat=None,
|
|
pve_users=None,
|
|
exec_return="",
|
|
sensors=None,
|
|
cluster_status=None,
|
|
):
|
|
"""Build a MagicMock ProxmoxAPI with pre-configured return values."""
|
|
api = MagicMock()
|
|
|
|
# Nodes list
|
|
api.nodes.get.return_value = nodes or NODES_LIST
|
|
|
|
# Version
|
|
api.version.get.return_value = version or VERSION_INFO
|
|
|
|
# Node sub-resource
|
|
node_res = MagicMock()
|
|
api.nodes.return_value = node_res
|
|
|
|
node_res.status.get.return_value = node_status or NODE_STATUS
|
|
node_res.network.get.return_value = network if network is not None else NODE_NETWORK
|
|
node_res.dns.get.return_value = dns or DNS_INFO
|
|
node_res.ntp.get.return_value = ntp or NTP_INFO
|
|
node_res.netstat.get.return_value = netstat or NODE_NETSTAT
|
|
node_res.execute.post.return_value = {"data": exec_return}
|
|
node_res.hardware.sensors.get.return_value = sensors or []
|
|
node_res.time.get.return_value = {}
|
|
node_res.subscription.get.return_value = {}
|
|
|
|
# SDN
|
|
cluster = MagicMock()
|
|
api.cluster = cluster
|
|
cluster.status.get.return_value = cluster_status or []
|
|
cluster.sdn.zones.get.return_value = sdn_zones or SDN_ZONES
|
|
cluster.sdn.vnets.get.return_value = sdn_vnets or SDN_VNETS
|
|
|
|
# SDN subnets per vnet
|
|
subnets_map = sdn_subnets or {
|
|
"vnet0": SDN_SUBNETS_VNET0,
|
|
"vnet1": SDN_SUBNETS_VNET1,
|
|
"vnet2": SDN_SUBNETS_VNET2,
|
|
}
|
|
|
|
def _vnet_resource(vnet_id):
|
|
vnet_res = MagicMock()
|
|
vnet_res.subnets.get.return_value = subnets_map.get(vnet_id, [])
|
|
return vnet_res
|
|
|
|
cluster.sdn.vnets.side_effect = _vnet_resource
|
|
|
|
# Access / users
|
|
api.access.users.get.return_value = pve_users or PVE_USERS
|
|
api.access.users.return_value.get.return_value = {}
|
|
|
|
return api
|
|
|
|
|
|
@pytest.fixture
|
|
def driver():
|
|
"""Return an open ProxmoxDriver with a mocked API."""
|
|
drv = ProxmoxDriver(
|
|
hostname="pve1.example.com",
|
|
username="root",
|
|
password="secret",
|
|
optional_args={"verify_ssl": False, "realm": "pam"},
|
|
)
|
|
mock_api = _build_mock_api()
|
|
with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
|
|
drv.open()
|
|
return drv
|
|
|
|
|
|
@pytest.fixture
|
|
def driver_with_exec(request):
|
|
"""Return a driver whose exec endpoint returns a parametrised string."""
|
|
exec_output = getattr(request, "param", "")
|
|
drv = ProxmoxDriver(
|
|
hostname="pve1.example.com",
|
|
username="root",
|
|
password="secret",
|
|
optional_args={"verify_ssl": False},
|
|
)
|
|
mock_api = _build_mock_api(exec_return=exec_output)
|
|
with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
|
|
drv.open()
|
|
return drv
|