NAPALM/napalm-proxmox
1
0 Fork 0
Code Issues 14 Pull Requests Actions Packages Projects Releases Wiki Activity

chore: adopt the NAPALM driver standard #33

Merged
christianmanivong merged 4 commits from chore/driver-standard into main 2026-10-23 12:17:27 +00:00
Conversation 0 Commits 4 Files Changed 32
+195 -225
27 changed files with 195 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 3b873d91fe - Show all commits

No files matched your search

README.md
+3 -3
View File
@@ -45,9 +45,9 @@ device = driver(
username="root",
password="secret",
optional_args={
"realm": "pam", # default: "pam"
"port": 8006, # default: 8006
"verify_ssl": True, # default: True
"realm": "pam", # default: "pam"
"port": 8006, # default: 8006
"verify_ssl": True, # default: True
},
)
napalm_proxmox/config_mixin.py
+2 -7
View File
@@ -19,7 +19,6 @@ from __future__ import annotations
import re
from typing import Any
_JsonDict = dict[str, Any]
@@ -44,13 +43,9 @@ class ProxmoxConfigMixin:
configs: _JsonDict = {"running": "", "candidate": "", "startup": ""}
if retrieve in ("running", "all", "startup"):
raw = self._exec_ssh_command(
"cat /etc/network/interfaces 2>/dev/null || true"
)
raw = self._exec_ssh_command("cat /etc/network/interfaces 2>/dev/null || true")
# Append SDN config if available
sdn_raw = self._exec_ssh_command(
"cat /etc/pve/sdn/vnets.cfg 2>/dev/null || true"
)
sdn_raw = self._exec_ssh_command("cat /etc/pve/sdn/vnets.cfg 2>/dev/null || true")
running = raw
if sdn_raw:
running += "\n# === SDN VNets ===\n" + sdn_raw
napalm_proxmox/driver.py
+20 -28
View File
@@ -34,6 +34,7 @@ from typing import Any
logger = logging.getLogger(__name__)
from napalm.base.exceptions import ConnectionException
from napalm_device_types import (
AptProxyMixin,
EnsurePackageMixin,
@@ -46,29 +47,26 @@ from napalm_device_types import (
PortSpec,
ServicePackagesMixin,
SnmpAgentMixin,
SystemIdentityMixin,
SystemdServicesMixin,
SystemIdentityMixin,
)
from napalm.base.exceptions import ConnectionException
from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
try:
from proxmoxer import ProxmoxAPI
except ImportError as exc:
raise ImportError(
"proxmoxer is required: pip install proxmoxer"
) from exc
raise ImportError("proxmoxer is required: pip install proxmoxer") from exc
from napalm_proxmox.interfaces_mixin import ProxmoxInterfaceMixin
from napalm_proxmox.sdn_mixin import ProxmoxSDNMixin
from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
from napalm_proxmox.config_mixin import ProxmoxConfigMixin
from napalm_proxmox.vm_mixin import ProxmoxVMMixin
from napalm_proxmox.vm_contract_mixin import ProxmoxVMContractMixin
from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
from napalm_proxmox.interfaces_mixin import ProxmoxInterfaceMixin
from napalm_proxmox.lldp_mixin import ProxmoxLLDPMixin
from napalm_proxmox.routing_mixin import ProxmoxRoutingMixin
from napalm_proxmox.sdn_mixin import ProxmoxSDNMixin
from napalm_proxmox.system_mixin import ProxmoxSystemMixin
from napalm_proxmox.vm_contract_mixin import ProxmoxVMContractMixin
from napalm_proxmox.vm_mixin import ProxmoxVMMixin
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
from napalm_proxmox.vm_snapshot_mixin import ProxmoxVMSnapshotMixin
# --------------------------------------------------------------------------- #
# Type aliases
@@ -144,7 +142,8 @@ class ProxmoxDriver(
self._port: int = self.optional_args.get("port", 8006)
self._verify_ssl: bool = self.optional_args.get(
"verify_ssl", self.optional_args.get("ssl_verify", self.optional_args.get("verify", True))
"verify_ssl",
self.optional_args.get("ssl_verify", self.optional_args.get("verify", True)),
)
self._realm: str = self.optional_args.get("realm", "pam")
self._token_name: str | None = self.optional_args.get("token_name")
@@ -157,7 +156,7 @@ class ProxmoxDriver(
self._api: ProxmoxAPI | None = None
self._node_name: str = ""
self._ssh_client: "paramiko.SSHClient | None" = None
self._ssh_client: paramiko.SSHClient | None = None
# Candidate config (merge/replace)
self._candidate_config: str = ""
@@ -211,9 +210,7 @@ class ProxmoxDriver(
self._resolve_node()
except Exception as exc:
raise ConnectionException(
f"Cannot connect to {self.hostname}: {exc}"
) from exc
raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc
def _resolve_node(self) -> str:
"""Resolve the PVE node this connection talks to.
@@ -231,9 +228,7 @@ class ProxmoxDriver(
self._node_name = self._node
return self._node_name
members = [
s for s in (self._api.cluster.status.get() or []) if s.get("type") == "node"
]
members = [s for s in (self._api.cluster.status.get() or []) if s.get("type") == "node"]
short_host = self.hostname.split(".")[0]
for pick in (
lambda s: s.get("local"),
@@ -305,7 +300,6 @@ class ProxmoxDriver(
falls back to a direct paramiko SSH connection.
"""
import base64 as _b64
import time as _time
# Try API exec endpoint
try:
@@ -359,7 +353,9 @@ class ProxmoxDriver(
connect_kwargs["password"] = ssh_pass
client = paramiko.SSHClient()
try:
HostKeyCheck(self.optional_args, self.hostname, 22).connect(client, **connect_kwargs)
HostKeyCheck(self.optional_args, self.hostname, 22).connect(
client, **connect_kwargs
)
except Exception:
client.close()
raise
@@ -439,9 +435,7 @@ class ProxmoxDriver(
hostname = self._node_name
fqdn = f"{self._node_name}.{dns_search}" if dns_search else self.hostname
iface_list = sorted(
iface["iface"] for iface in network if iface.get("iface")
)
iface_list = sorted(iface["iface"] for iface in network if iface.get("iface"))
pve_version = version.get("version", "")
release = version.get("release", "")
@@ -465,7 +459,7 @@ class ProxmoxDriver(
"n=$(cat /sys/class/dmi/id/product_name 2>/dev/null); "
"r=$(cat /sys/class/dmi/id/product_version 2>/dev/null); "
"s=$(cat /sys/class/dmi/id/product_serial 2>/dev/null); "
"printf '%s\\n%s\\n%s\\n%s\\n' \"$v\" \"$n\" \"$r\" \"$s\""
'printf \'%s\\n%s\\n%s\\n%s\\n\' "$v" "$n" "$r" "$s"'
)
lines = self._exec_ssh_command(dmi_cmd).splitlines()
if len(lines) >= 4:
@@ -515,5 +509,3 @@ class ProxmoxDriver(
def cli(self, commands: list[str], encoding: str = "text") -> dict[str, str]:
"""Execute a list of shell commands and return their outputs."""
return {cmd: self._exec_ssh_command(cmd) for cmd in commands}
napalm_proxmox/interfaces_mixin.py
+8 -7
View File
@@ -175,7 +175,12 @@ class ProxmoxInterfaceMixin:
if len(parts) < 6:
continue
ip_addr, _, flags, mac, _, iface = (
parts[0], parts[1], parts[2], parts[3], parts[4], parts[5]
parts[0],
parts[1],
parts[2],
parts[3],
parts[4],
parts[5],
)
if mac in ("00:00:00:00:00:00", ""):
continue
@@ -202,9 +207,7 @@ class ProxmoxInterfaceMixin:
if iface.get("type") in ("bridge",) and iface.get("iface")
]
for bridge in linux_bridges:
raw = self._exec_ssh_command(
f"bridge fdb show br {bridge} 2>/dev/null || true"
)
raw = self._exec_ssh_command(f"bridge fdb show br {bridge} 2>/dev/null || true")
for line in raw.splitlines():
parts = line.split()
if len(parts) < 3:
@@ -240,9 +243,7 @@ class ProxmoxInterfaceMixin:
if iface.get("type") in ("OVSBridge",) and iface.get("iface")
]
for bridge in ovs_bridges:
raw = self._exec_ssh_command(
f"ovs-appctl fdb/show {bridge} 2>/dev/null || true"
)
raw = self._exec_ssh_command(f"ovs-appctl fdb/show {bridge} 2>/dev/null || true")
for line in raw.splitlines():
parts = line.split()
if len(parts) < 4:
napalm_proxmox/lldp_mixin.py
+2 -7
View File
@@ -19,7 +19,6 @@ from __future__ import annotations
import re
from typing import Any
_JsonDict = dict[str, Any]
@@ -29,9 +28,7 @@ class ProxmoxLLDPMixin:
def get_lldp_neighbors(self) -> dict[str, list[_JsonDict]]:
"""Return LLDP neighbours (requires lldpd on the Proxmox node)."""
result: dict[str, list[_JsonDict]] = {}
raw = self._exec_ssh_command(
"lldpcli show neighbors summary 2>/dev/null || true"
)
raw = self._exec_ssh_command("lldpcli show neighbors summary 2>/dev/null || true")
current_iface = ""
for line in raw.splitlines():
m_iface = re.match(r"^\s*Interface:\s+(\S+?),?\s", line)
@@ -54,9 +51,7 @@ class ProxmoxLLDPMixin:
def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[_JsonDict]]:
"""Return detailed LLDP neighbour information."""
result: dict[str, list[_JsonDict]] = {}
raw = self._exec_ssh_command(
"lldpcli show neighbors details 2>/dev/null || true"
)
raw = self._exec_ssh_command("lldpcli show neighbors details 2>/dev/null || true")
current_iface = ""
current_entry: _JsonDict = {}
napalm_proxmox/routing_mixin.py
+2 -7
View File
@@ -21,7 +21,6 @@ from typing import Any
import napalm.base.constants as C
_JsonDict = dict[str, Any]
@@ -92,12 +91,8 @@ class ProxmoxRoutingMixin:
# Individual probe lines
probes = []
for m_probe in re.finditer(
r"icmp_seq=\d+.*?time=([\d.]+) ms.*?from ([\d.a-fA-F:]+)", raw
):
probes.append(
{"ip_address": m_probe.group(2), "rtt": float(m_probe.group(1))}
)
for m_probe in re.finditer(r"icmp_seq=\d+.*?time=([\d.]+) ms.*?from ([\d.a-fA-F:]+)", raw):
probes.append({"ip_address": m_probe.group(2), "rtt": float(m_probe.group(1))})
return {
"success": {
napalm_proxmox/sdn_mixin.py
+5 -12
View File
@@ -153,7 +153,7 @@ class ProxmoxSDNMixin:
tags.setdefault(str(tag), set()).add(bridge)
try:
for vm in (self._node_api().qemu.get() or []):
for vm in self._node_api().qemu.get() or []:
vmid = int(vm.get("vmid", 0))
try:
config = self._node_api().qemu(vmid).config.get() or {}
@@ -164,7 +164,7 @@ class ProxmoxSDNMixin:
logger.warning("_get_vm_vlan_tags: failed to list QEMU VMs: %s", exc)
try:
for ct in (self._node_api().lxc.get() or []):
for ct in self._node_api().lxc.get() or []:
vmid = int(ct.get("vmid", 0))
try:
config = self._node_api().lxc(vmid).config.get() or {}
@@ -238,9 +238,7 @@ class ProxmoxSDNMixin:
2. LLDP basic fallback: first physical port with any LLDP neighbour.
"""
network = {
iface["iface"]: iface
for iface in self._get_node_network()
if iface.get("iface")
iface["iface"]: iface for iface in self._get_node_network() if iface.get("iface")
}
try:
@@ -309,10 +307,7 @@ class ProxmoxSDNMixin:
return True
min_id = min(int(n.get("nodeid", 9999)) for n in online_nodes)
for n in online_nodes:
if (
n.get("name") == self._node_name
and int(n.get("nodeid", 9999)) == min_id
):
if n.get("name") == self._node_name and int(n.get("nodeid", 9999)) == min_id:
return True
return False
except Exception as exc:
@@ -361,9 +356,7 @@ class ProxmoxSDNMixin:
vlan_id: VLAN identifier (1-4094).
config: Dict that may contain ``"name"`` for the VLAN alias.
"""
name: str = (
(config.get("name") or f"VLAN{vlan_id}") if config else f"VLAN{vlan_id}"
)
name: str = (config.get("name") or f"VLAN{vlan_id}") if config else f"VLAN{vlan_id}"
vnet_id = f"vlan{vlan_id:04d}"
uplink = self._find_switch_uplink()
napalm_proxmox/system_mixin.py
+39 -40
View File
@@ -25,8 +25,6 @@ from typing import Any
from napalm_device_types import APT_UPGRADABLE_COMMAND, parse_apt_upgradable
from napalm_proxmox import utils
logger = logging.getLogger(__name__)
_JsonDict = dict[str, Any]
@@ -177,9 +175,7 @@ class ProxmoxSystemMixin:
Proxmox does not expose SNMP configuration via the REST API.
We read /etc/snmp/snmpd.conf via exec if available.
"""
raw = self._exec_ssh_command(
"cat /etc/snmp/snmpd.conf 2>/dev/null || true"
)
raw = self._exec_ssh_command("cat /etc/snmp/snmpd.conf 2>/dev/null || true")
communities: dict[str, _JsonDict] = {}
location = ""
contact = ""
@@ -344,17 +340,19 @@ class ProxmoxSystemMixin:
if not name or status != "installed":
continue
size_bytes = int(size_kb) * 1024 if size_kb.isdigit() else 0
result.append({
"name": name,
"version": version,
"installed": True,
"description": "",
"size": size_bytes,
"source": "pve",
"source_package": source_package,
"source_version": source_version,
"upgrade_version": upgradable.get(name, ""),
})
result.append(
{
"name": name,
"version": version,
"installed": True,
"description": "",
"size": size_bytes,
"source": "pve",
"source_package": source_package,
"source_version": source_version,
"upgrade_version": upgradable.get(name, ""),
}
)
return result
# ------------------------------------------------------------------ #
@@ -443,7 +441,7 @@ class ProxmoxSystemMixin:
def apply_updates(self, packages: list[str]) -> _JsonDict:
"""Upgrade the given packages via ``apt-get install`` over SSH."""
for pkg in packages:
if not re.match(r'^[a-zA-Z0-9_\-\+\.]+$', pkg):
if not re.match(r"^[a-zA-Z0-9_\-\+\.]+$", pkg):
raise ValueError(f"Invalid package name: {pkg!r}")
pkg_args = " ".join(packages)
output = self._exec_ssh_command(
@@ -526,9 +524,9 @@ class ProxmoxSystemMixin:
encoded = base64.b64encode(content.encode()).decode()
# No ``exit`` here: ``_rc`` reads the status after the command's braces.
return self._step(
f"umask 077 && t=$(mktemp) && echo {encoded} | base64 -d > \"$t\" && "
f"install -m 644 -o root -g root \"$t\" {shlex.quote(path)}; "
"s=$?; rm -f \"$t\"; [ \"$s\" -eq 0 ]",
f'umask 077 && t=$(mktemp) && echo {encoded} | base64 -d > "$t" && '
f'install -m 644 -o root -g root "$t" {shlex.quote(path)}; '
's=$?; rm -f "$t"; [ "$s" -eq 0 ]',
f"wrote {path}",
)
@@ -557,9 +555,9 @@ class ProxmoxSystemMixin:
def _write_snmp_agent_config(self, settings: dict) -> dict:
encoded = base64.b64encode(render_snmpd_conf(settings).encode()).decode()
return self._step(
f"umask 077 && t=$(mktemp) && echo {encoded} | base64 -d > \"$t\" && "
f"install -m 644 -o root -g root \"$t\" /etc/snmp/snmpd.conf; "
"s=$?; rm -f \"$t\"; exit $s",
f'umask 077 && t=$(mktemp) && echo {encoded} | base64 -d > "$t" && '
f'install -m 644 -o root -g root "$t" /etc/snmp/snmpd.conf; '
's=$?; rm -f "$t"; exit $s',
"wrote /etc/snmp/snmpd.conf",
)
@@ -614,9 +612,11 @@ class ProxmoxSystemMixin:
fqdn = new_hostname
# Derive old short hostname for substitution in /etc/hosts
old_short = self._exec_ssh_command(
"cat /etc/hostname 2>/dev/null || hostname -s 2>/dev/null"
).strip().split(".")[0]
old_short = (
self._exec_ssh_command("cat /etc/hostname 2>/dev/null || hostname -s 2>/dev/null")
.strip()
.split(".")[0]
)
# ── 1. /etc/hostname ─────────────────────────────────────────────
hostname_b64 = _b64.b64encode(f"{short_hostname}\n".encode()).decode()
@@ -631,6 +631,7 @@ class ProxmoxSystemMixin:
# Both "hostname" and "hostname.domain" forms are covered because
# the new value is the full fqdn when a domain was given.
import re as _re
new_replace = fqdn if domain else short_hostname
hosts_new = _re.sub(
r"\b" + _re.escape(old_short) + r"\b",
@@ -647,16 +648,12 @@ class ProxmoxSystemMixin:
# ── 3. /etc/mailname ─────────────────────────────────────────────
mailname_b64 = _b64.b64encode(f"{fqdn}\n".encode()).decode()
self._exec_ssh_command(
f"[ -f /etc/mailname ]"
f" && echo {mailname_b64} | base64 -d > /etc/mailname"
f" || true"
f"[ -f /etc/mailname ] && echo {mailname_b64} | base64 -d > /etc/mailname || true"
)
# ── 4. Postfix myhostname ─────────────────────────────────────────
self._exec_ssh_command(
f"command -v postconf >/dev/null 2>&1"
f" && postconf -e 'myhostname = {fqdn}'"
f" || true"
f"command -v postconf >/dev/null 2>&1 && postconf -e 'myhostname = {fqdn}' || true"
)
# ── 5. Apply hostname at runtime ─────────────────────────────────
@@ -699,7 +696,9 @@ class ProxmoxSystemMixin:
logger.info(
"Proxmox set_hostname: %s → %s (fqdn: %s). "
"Reboot required for node name to update in web UI / cluster.",
old_short, short_hostname, fqdn,
old_short,
short_hostname,
fqdn,
)
# ------------------------------------------------------------------ #
@@ -736,12 +735,12 @@ class ProxmoxSystemMixin:
if not dev:
continue
entry: dict = {
"model": disk.get("model", ""),
"serial": disk.get("serial", ""),
"type": disk.get("type", ""),
"size": disk.get("size", 0),
"health": disk.get("health", "unknown").lower(),
"wearout": disk.get("wearout"), # NVMe wear indicator 0-100
"model": disk.get("model", ""),
"serial": disk.get("serial", ""),
"type": disk.get("type", ""),
"size": disk.get("size", 0),
"health": disk.get("health", "unknown").lower(),
"wearout": disk.get("wearout"), # NVMe wear indicator 0-100
"temperature": None,
"percentage_used": None,
"available_spare": None,
@@ -772,7 +771,7 @@ class ProxmoxSystemMixin:
# HDD-specific SMART attributes
for attr in smart.get("attributes", []):
name = attr.get("name", "").lower()
raw = attr.get("raw", "")
raw = attr.get("raw", "")
try:
raw_int = int(str(raw).split()[0])
except (ValueError, TypeError):
napalm_proxmox/utils.py
+3 -1
View File
@@ -14,7 +14,6 @@ from typing import Any
from netaddr import EUI, AddrFormatError
# ---------------------------------------------------------------------------
# MAC address helpers
# ---------------------------------------------------------------------------
@@ -47,6 +46,7 @@ def normalize_mac(mac: str) -> str:
# IP / prefix helpers
# ---------------------------------------------------------------------------
def prefix_length(cidr: str) -> int:
"""Return prefix length from a CIDR string such as '192.168.1.0/24' → 24."""
try:
@@ -141,6 +141,7 @@ def speed_mbps(iface_data: dict[str, Any]) -> float:
# ARP / neighbour table helpers
# ---------------------------------------------------------------------------
def parse_arp_output(raw: str) -> list[dict[str, Any]]:
"""Parse Linux ``arp -n`` text output into NAPALM ARP dicts.
@@ -202,6 +203,7 @@ def sdn_zone_to_network_instance(zone: dict[str, Any]) -> dict[str, Any]:
# OVS helpers
# ---------------------------------------------------------------------------
def parse_ovs_vsctl_show(raw: str) -> dict[str, Any]:
"""Very lightweight parser for ``ovs-vsctl show`` output.
napalm_proxmox/vm_mixin.py
+53 -52
View File
@@ -20,8 +20,6 @@ import logging
import re
from typing import Any
from napalm_proxmox import utils
logger = logging.getLogger(__name__)
_JsonDict = dict[str, Any]
@@ -30,9 +28,7 @@ _JsonDict = dict[str, Any]
class ProxmoxVMMixin:
"""Mixin providing VM and container NAPALM methods."""
def get_vm_interfaces(
self, vmid: int, vm_type: str
) -> tuple[dict[str, _JsonDict], bool, bool]:
def get_vm_interfaces(self, vmid: int, vm_type: str) -> tuple[dict[str, _JsonDict], bool, bool]:
"""Return network interfaces for a single VM or LXC container.
Returns a 3-tuple ``(interfaces, agent_running, agent_enabled)``:
@@ -105,7 +101,7 @@ class ProxmoxVMMixin:
logger.debug("get_vm_interfaces: LXC %s config failed: %s", vmid, exc)
try:
for iface in (self._node_api().lxc(vmid).interfaces.get() or []):
for iface in self._node_api().lxc(vmid).interfaces.get() or []:
name = iface.get("name", "")
if not name or name == "lo":
continue
@@ -233,7 +229,7 @@ class ProxmoxVMMixin:
# QEMU VMs
try:
for vm in (self._node_api().qemu.get() or []):
for vm in self._node_api().qemu.get() or []:
vmid = int(vm.get("vmid", 0))
name = vm.get("name", f"vm-{vmid}")
status = vm.get("status", "unknown")
@@ -256,30 +252,32 @@ class ProxmoxVMMixin:
)
disks, onboot = self._get_vm_disk_and_boot(vmid, "qemu")
result.append({
"vmid": str(vmid),
"name": name,
"type": "vm",
"status": status,
"vcpus": vcpus,
"memory": memory_mb,
"cpu_usage": round(cpu_usage, 4),
"memory_usage": memory_usage_mb,
"uptime": uptime,
"node": self._node_name,
"interfaces": interfaces,
"ipv4": ipv4,
"agent_enabled": agent_enabled,
"agent_running": agent_running,
"disks": disks,
"onboot": onboot,
})
result.append(
{
"vmid": str(vmid),
"name": name,
"type": "vm",
"status": status,
"vcpus": vcpus,
"memory": memory_mb,
"cpu_usage": round(cpu_usage, 4),
"memory_usage": memory_usage_mb,
"uptime": uptime,
"node": self._node_name,
"interfaces": interfaces,
"ipv4": ipv4,
"agent_enabled": agent_enabled,
"agent_running": agent_running,
"disks": disks,
"onboot": onboot,
}
)
except Exception as exc:
logger.warning("get_vms: failed to list QEMU VMs: %s", exc)
# LXC containers
try:
for ct in (self._node_api().lxc.get() or []):
for ct in self._node_api().lxc.get() or []:
vmid = int(ct.get("vmid", 0))
name = ct.get("name", f"ct-{vmid}")
status = ct.get("status", "unknown")
@@ -300,24 +298,26 @@ class ProxmoxVMMixin:
)
disks, onboot = self._get_vm_disk_and_boot(vmid, "lxc")
result.append({
"vmid": str(vmid),
"name": name,
"type": "container",
"status": status,
"vcpus": vcpus,
"memory": memory_mb,
"cpu_usage": round(cpu_usage, 4),
"memory_usage": memory_usage_mb,
"uptime": uptime,
"node": self._node_name,
"interfaces": interfaces,
"ipv4": ipv4,
"agent_enabled": agent_enabled,
"agent_running": agent_running,
"disks": disks,
"onboot": onboot,
})
result.append(
{
"vmid": str(vmid),
"name": name,
"type": "container",
"status": status,
"vcpus": vcpus,
"memory": memory_mb,
"cpu_usage": round(cpu_usage, 4),
"memory_usage": memory_usage_mb,
"uptime": uptime,
"node": self._node_name,
"interfaces": interfaces,
"ipv4": ipv4,
"agent_enabled": agent_enabled,
"agent_running": agent_running,
"disks": disks,
"onboot": onboot,
}
)
except Exception as exc:
logger.warning("get_vms: failed to list LXC containers: %s", exc)
@@ -328,9 +328,7 @@ class ProxmoxVMMixin:
# Disk-key prefix for LXC: rootfs, mp (mount points)
_DISK_KEYS_CT = re.compile(r"^(rootfs|mp\d+)$")
def _get_vm_disk_and_boot(
self, vmid: int, vm_type: str
) -> tuple[list[dict], bool]:
def _get_vm_disk_and_boot(self, vmid: int, vm_type: str) -> tuple[list[dict], bool]:
"""Return (disks, onboot) for a VM or container.
Fetches config from /nodes/{node}/{qemu|lxc}/{vmid}/config.
@@ -376,8 +374,8 @@ class ProxmoxVMMixin:
logger.debug("Failed to fetch disk/boot config for vmid %s: %s", vmid, exc)
return [], False
_POWER_ACTIONS_VM = {'start', 'stop', 'shutdown', 'reboot', 'reset'}
_POWER_ACTIONS_CT = {'start', 'stop', 'shutdown', 'reboot'}
_POWER_ACTIONS_VM = {"start", "stop", "shutdown", "reboot", "reset"}
_POWER_ACTIONS_CT = {"start", "stop", "shutdown", "reboot"}
def power_vm(self, vmid: int, vm_type: str, action: str) -> dict:
"""Send a power action to a VM or container on this node.
@@ -385,11 +383,14 @@ class ProxmoxVMMixin:
Supported actions for VMs: start, stop, shutdown, reboot, reset
Supported actions for containers: start, stop, shutdown, reboot
"""
allowed = self._POWER_ACTIONS_VM if vm_type == 'vm' else self._POWER_ACTIONS_CT
allowed = self._POWER_ACTIONS_VM if vm_type == "vm" else self._POWER_ACTIONS_CT
if action not in allowed:
return {"success": False, "error": f"Action '{action}' not supported for {vm_type} (allowed: {sorted(allowed)})"}
return {
"success": False,
"error": f"Action '{action}' not supported for {vm_type} (allowed: {sorted(allowed)})",
}
try:
vm_api = self._node_api().qemu(vmid) if vm_type == 'vm' else self._node_api().lxc(vmid)
vm_api = self._node_api().qemu(vmid) if vm_type == "vm" else self._node_api().lxc(vmid)
task_id = getattr(vm_api.status, action).post()
return {"success": True, "task_id": task_id or ""}
except Exception as exc:
napalm_proxmox/vm_provision_mixin.py
+10 -10
View File
@@ -8,7 +8,7 @@ import ipaddress
import logging
import re
import time
from typing import TYPE_CHECKING, Any, Dict, List
from typing import TYPE_CHECKING, Any
from urllib.parse import quote
import yaml
@@ -257,7 +257,7 @@ class ProxmoxVMProvisionMixin:
timeout=30,
)
def _nic_macs(self, vmid: int, nics: List[Dict[str, Any]]) -> list[tuple[str, bool]]:
def _nic_macs(self, vmid: int, nics: list[dict[str, Any]]) -> list[tuple[str, bool]]:
"""``(mac, dhcp)`` per NIC, with the MAC Proxmox assigned where none was given."""
config = self._node_api().qemu(vmid).config.get()
macs = []
@@ -429,9 +429,9 @@ class ProxmoxVMProvisionMixin:
)
return path
def get_image_storages(self) -> List[StorageTargetDict]:
def get_image_storages(self) -> list[StorageTargetDict]:
"""List node-available storage pools suitable for a new VM's root disk."""
targets: List[StorageTargetDict] = []
targets: list[StorageTargetDict] = []
for storage in self._node_api().storage.get():
content = storage.get("content", "")
if "images" not in content or storage.get("enabled", 1) == 0:
@@ -538,10 +538,10 @@ class ProxmoxVMProvisionMixin:
image_url: str,
cpu: int,
memory: int,
nics: List[Dict[str, Any]],
cloud_init_config: Dict[str, Any],
nics: list[dict[str, Any]],
cloud_init_config: dict[str, Any],
image_checksum: str | None = None,
ssh_public_keys: List[str] | None = None,
ssh_public_keys: list[str] | None = None,
disk_resize_gb: int | None = None,
storage: str | None = None,
cpu_type: str | None = None,
@@ -633,7 +633,7 @@ class ProxmoxVMProvisionMixin:
# Step 4: Configure network interfaces (CPU/memory already set at shell creation)
_logger.info(f"Configuring {len(nics)} NIC(s) for VM {vmid}")
config_args: Dict[str, Any] = {}
config_args: dict[str, Any] = {}
# Build NIC config strings generically
for i, nic in enumerate(nics):
@@ -941,14 +941,14 @@ class ProxmoxVMProvisionMixin:
_logger.exception(f"Error getting status for VM {vmid}: {e}")
raise RuntimeError(f"Failed to get VM {vmid} status: {e}")
def get_network_targets(self) -> List[NetworkTargetDict]:
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] = []
targets: list[NetworkTargetDict] = []
for iface in self._get_node_network():
iface_type = iface.get("type")
pyproject.toml
+6 -2
View File
@@ -51,10 +51,14 @@ include = ["napalm_proxmox*"]
[tool.ruff]
line-length = 100
target-version = "py39"
target-version = "py310"
[tool.ruff.lint]
select = ["E", "F", "W", "I", "UP"]
select = ["E", "F", "I", "UP", "B"]
[tool.ruff.lint.per-file-ignores]
# Recorded device output keeps its lines.
"tests/**" = ["E501"]
[tool.pytest.ini_options]
testpaths = ["tests"]
tests/conftest.py
+3 -2
View File
@@ -2,11 +2,11 @@
from __future__ import annotations
import pytest
from unittest.mock import MagicMock, patch
from napalm_proxmox.driver import ProxmoxDriver
import pytest
from napalm_proxmox.driver import ProxmoxDriver
# ---------------------------------------------------------------------------
# Sample API payloads
@@ -124,6 +124,7 @@ PVE_USERS = [
# Driver fixture
# ---------------------------------------------------------------------------
def _build_mock_api(
nodes=None,
node_status=None,
tests/test_connection.py
+6 -9
View File
@@ -2,12 +2,13 @@
from __future__ import annotations
import pytest
from unittest.mock import MagicMock, patch
from unittest.mock import patch
import pytest
from napalm.base.exceptions import ConnectionException
from napalm_proxmox.driver import ProxmoxDriver
from tests.conftest import _build_mock_api, NODES_LIST
from tests.conftest import _build_mock_api
class TestOpen:
@@ -101,9 +102,7 @@ def _open_with(hostname, **api_kwargs):
class TestClusterNodeResolution:
def test_local_node_wins_over_first_listed(self):
drv = _open_with(
"172.22.8.120", nodes=CLUSTER_NODES_LIST, cluster_status=CLUSTER_STATUS
)
drv = _open_with("172.22.8.120", nodes=CLUSTER_NODES_LIST, cluster_status=CLUSTER_STATUS)
assert drv._node_name == "pve-dual"
def test_matches_by_ip_without_local_flag(self):
@@ -118,9 +117,7 @@ class TestClusterNodeResolution:
def test_matches_short_name_of_fqdn(self):
status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
drv = _open_with(
"pve-01.mgmt.example.com", nodes=CLUSTER_NODES_LIST, cluster_status=status
)
drv = _open_with("pve-01.mgmt.example.com", nodes=CLUSTER_NODES_LIST, cluster_status=status)
assert drv._node_name == "pve-01"
def test_single_node_without_cluster_status(self):
tests/test_environment.py
+3 -8
View File
@@ -2,11 +2,12 @@
from __future__ import annotations
import pytest
from tests.conftest import _build_mock_api, NODE_STATUS
from unittest.mock import patch
import pytest
from napalm_proxmox.driver import ProxmoxDriver
from tests.conftest import _build_mock_api
class TestGetEnvironment:
@@ -25,9 +26,6 @@ class TestGetEnvironment:
assert isinstance(result["fans"], dict)
def test_temperature_with_sensors(self):
from tests.conftest import _build_mock_api, NODE_STATUS
from unittest.mock import patch
from napalm_proxmox.driver import ProxmoxDriver
sensors_data = [
{"name": "CPU Package", "value": "55.0"},
@@ -49,9 +47,6 @@ class TestGetEnvironment:
assert result["temperature"]["GPU Core"]["is_critical"] is False
def test_temperature_critical(self):
from tests.conftest import _build_mock_api
from unittest.mock import patch
from napalm_proxmox.driver import ProxmoxDriver
sensors_data = [{"name": "NVMe", "value": "98.0"}]
mock_api = _build_mock_api(sensors=sensors_data)
tests/test_get_facts.py
+1 -3
View File
@@ -3,10 +3,8 @@
from __future__ import annotations
import pytest
from tests.conftest import _build_mock_api, NODE_NETWORK
from unittest.mock import patch
from napalm_proxmox.driver import ProxmoxDriver
from tests.conftest import NODE_NETWORK
@pytest.fixture
tests/test_interfaces.py
+13 -9
View File
@@ -2,11 +2,7 @@
from __future__ import annotations
import pytest
from tests.conftest import _build_mock_api, NODE_NETWORK
from unittest.mock import patch
from napalm_proxmox.driver import ProxmoxDriver
from tests.conftest import NODE_NETWORK
class TestGetInterfaces:
@@ -89,10 +85,18 @@ class TestGetInterfacesCounters:
result = driver.get_interfaces_counters()
eth0 = result["eth0"]
for field in (
"tx_errors", "rx_errors", "tx_discards", "rx_discards",
"tx_octets", "rx_octets", "tx_unicast_packets", "rx_unicast_packets",
"tx_multicast_packets", "rx_multicast_packets",
"tx_broadcast_packets", "rx_broadcast_packets",
"tx_errors",
"rx_errors",
"tx_discards",
"rx_discards",
"tx_octets",
"rx_octets",
"tx_unicast_packets",
"rx_unicast_packets",
"tx_multicast_packets",
"rx_multicast_packets",
"tx_broadcast_packets",
"rx_broadcast_packets",
):
assert field in eth0, f"Missing field: {field}"
tests/test_kernel_facts.py
+1 -1
View File
@@ -12,9 +12,9 @@ import gzip
from unittest.mock import patch
import pytest
from napalm_device_types import KernelFactsMixin
from napalm_device_types.kernel import KERNEL_FACTS_COMMAND
from napalm_proxmox.driver import ProxmoxDriver
REPORT = (
tests/test_licenses.py
+5 -2
View File
@@ -58,5 +58,8 @@ def test_an_unreadable_subscription_reports_nothing():
def test_no_warnings_of_its_own():
"""lldpd, the updates and the subscription are facts the caller reads
(get_packages, get_available_updates, get_licenses)."""
assert not any("get_device_warnings" in vars(c) for c in ProxmoxDriver.__mro__[:-1]
if c.__module__.startswith("napalm_proxmox"))
assert not any(
"get_device_warnings" in vars(c)
for c in ProxmoxDriver.__mro__[:-1]
if c.__module__.startswith("napalm_proxmox")
)
tests/test_listening_sockets.py
+1
View File
@@ -11,6 +11,7 @@ from __future__ import annotations
from unittest.mock import patch
from napalm_device_types import ListeningSocketsMixin
from napalm_proxmox.driver import ProxmoxDriver
WIRE = (
tests/test_misc.py
+2 -1
View File
@@ -2,8 +2,9 @@
from __future__ import annotations
from unittest.mock import patch
import pytest
from unittest.mock import MagicMock, patch
class TestGetNTPServers:
tests/test_ovs_and_arp.py
+1 -6
View File
@@ -2,12 +2,10 @@
from __future__ import annotations
import pytest
from unittest.mock import patch
from napalm_proxmox import utils
PROC_NET_ARP = """\
IP address HW type Flags HW address Mask Device
10.0.0.254 0x1 0x2 aa:bb:cc:dd:ee:ff * eth0
@@ -137,10 +135,7 @@ class TestGetMACAddressTable:
"cc:dd:ee:ff:00:11 dev eth0 vlan 20 master vmbr0\n"
)
OVS_APPCTL = (
" LOCAL 0 aa:bb:cc:dd:ee:03 0\n"
" 1 10 11:22:33:44:55:66 100\n"
)
OVS_APPCTL = " LOCAL 0 aa:bb:cc:dd:ee:03 0\n 1 10 11:22:33:44:55:66 100\n"
def test_linux_bridge_entries(self, driver):
def _exec(cmd):
tests/test_sdn.py
+2 -3
View File
@@ -2,8 +2,7 @@
from __future__ import annotations
import pytest
from tests.conftest import _build_mock_api, SDN_ZONES, SDN_VNETS
from tests.conftest import _build_mock_api
class TestGetVlans:
@@ -52,8 +51,8 @@ class TestGetVlans:
def test_empty_sdn_returns_dict(self):
from unittest.mock import patch
from napalm_proxmox.driver import ProxmoxDriver
from tests.conftest import _build_mock_api
mock_api = _build_mock_api(sdn_vnets=[], sdn_zones=[])
drv = ProxmoxDriver("pve1", "root", "secret", optional_args={"verify_ssl": False})
tests/test_service_packages.py
+1 -1
View File
@@ -11,9 +11,9 @@ from __future__ import annotations
from unittest.mock import patch
import pytest
from napalm_device_types import ServicePackagesMixin
from napalm_device_types.service_packages import SERVICE_PACKAGES_COMMAND
from napalm_proxmox.driver import ProxmoxDriver
WIRE = (
tests/test_system_identity.py
+1
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
from unittest.mock import patch
from napalm_device_types import SYSTEM_IDENTITY_COMMAND, SystemIdentityMixin
from napalm_proxmox.driver import ProxmoxDriver
# pve-dual, a Supermicro X10DRL-i: the product serial is the board vendor's default.
tests/test_utils.py
-1
View File
@@ -2,7 +2,6 @@
from __future__ import annotations
import pytest
from napalm_proxmox import utils
tests/test_vm_provision_mixin.py
+2 -3
View File
@@ -3,12 +3,11 @@
from __future__ import annotations
import base64
from unittest.mock import MagicMock, patch
import pytest
from unittest.mock import MagicMock, patch
from napalm_proxmox.driver import ProxmoxDriver
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
from napalm_proxmox.vm_provision_mixin import ProxmoxVMProvisionMixin
# ---------------------------------------------------------------------------
# Tests