"""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, ): """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.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