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

Merge pull request 'chore: adopt the NAPALM driver standard' (#33) from chore/driver-standard into main

Browse Source
This commit is contained in:
christianmanivong committed 2026-10-23 12:17:27 +00:00
2 parents 1f6859de08 711d5f2705
commit 984ae958d8
32 files changed
+365 -311
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitea/workflows/ci.yml
+17
View File
@@ -0,0 +1,17 @@
name: CI
on:
push:
branches: [main, master]
tags: ["v*"]
pull_request:
jobs:
driver:
uses: NAPALM/driver-standard/.gitea/workflows/driver.yml@main
with:
standard-ref: main
# Until napalm-device-types 4.22.0 (AptProxyMixin, #53) is on PyPI.
pre-install: "git+https://git.netork.io/NAPALM/napalm-device-types.git"
secrets:
PYPI_TOKEN: ${{ secrets.PYPI_TOKEN }}
.github/workflows/ci.yml
-48
View File
@@ -1,48 +0,0 @@
name: CI
on:
push:
branches: ["**"]
pull_request:
branches: ["**"]
jobs:
test:
runs-on: ubuntu-latest
strategy:
fail-fast: false
matrix:
python-version: ["3.10", "3.11", "3.12"]
steps:
- name: Checkout
uses: actions/checkout@v4
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
cache: pip
- name: Install package with dev extras
run: |
python -m pip install --upgrade pip
# napalm-device-types lives in git.netork.io/NAPALM, not on PyPI: without this
# pip looks there, finds an unrelated 0.1.0 and the job dies before any test.
python -m pip install "napalm-device-types @ git+https://git.netork.io/NAPALM/napalm-device-types.git"
python -m pip install -e ".[dev]"
- name: Run unit tests
run: |
python -m pytest -q --tb=short
- name: Build wheel and sdist
run: |
python -m pip install build
python -m build
- name: Upload dist artifacts
# v4 refuses to run on Gitea ("not currently supported on GHES").
uses: actions/upload-artifact@v3
with:
name: dist-${{ matrix.python-version }}
path: dist/*
CHANGELOG.md
+14
View File
@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [0.8.0] - 2026-10-23
The first release on PyPI. It collects everything since 0.1.0.
### Changed
- The driver follows the NAPALM driver standard
(https://git.netork.io/NAPALM/driver-standard): CI on Gitea runs ruff, mypy, the
tests on Python 3.10–3.13 and at the lowest declared versions, bandit, pip-audit and
a build check, and a release tag publishes to PyPI. The package now ships `py.typed`.
**Python 3.9 is no longer supported**; napalm-device-types never ran on it.
- Exceptions the driver deliberately ignores (closing the node's SSH session, the
`is_alive()` probe) are now logged at debug level.
- The project links point at git.netork.io.
### Added
- apt's proxy (napalm-device-types' `AptProxyMixin`, NetOrk/netork#1397): `get_package_proxy()`, `get_package_proxy_report()`, `check_package_proxy(url, expect)` and the fix `configure_package_proxy(url)`, as on any Debian host. The node is logged in as root, so the file is written and other proxy lines are commented out over the exec path. Requires napalm-device-types 4.22.0; version 0.7.0.
- `get_service_packages()` (napalm-device-types' `ServicePackagesMixin`, NetOrk/netork#1236): which installed packages each systemd unit of the node is made of — the one shipping its unit file and the one shipping its `ExecStart` binary. Read-only, over the exec path. Requires napalm-device-types 4.20.0; version 0.6.0.
README.md
+6 -5
View File
@@ -12,8 +12,9 @@ It supports all three networking domains of Proxmox VE:
## Requirements
- Python ≥ 3.9
- Python ≥ 3.10
- NAPALM ≥ 5.0.0
- napalm-device-types ≥ 4.22.0
- proxmoxer ≥ 2.0.0
- netaddr ≥ 0.9.0
- requests ≥ 2.31.0
@@ -27,7 +28,7 @@ pip install napalm-proxmox
Or directly from source:
```bash
git clone https://github.com/example/napalm-proxmox.git
git clone https://git.netork.io/NAPALM/napalm-proxmox.git
cd napalm-proxmox
pip install -e .
```
@@ -45,9 +46,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
+25 -33
View File
@@ -32,8 +32,7 @@ from __future__ import annotations
import logging
from typing import Any
logger = logging.getLogger(__name__)
from napalm.base.exceptions import ConnectionException
from napalm_device_types import (
AptProxyMixin,
EnsurePackageMixin,
@@ -46,29 +45,28 @@ 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
logger = logging.getLogger(__name__)
# --------------------------------------------------------------------------- #
# 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,8 @@ class ProxmoxDriver(
self._api: ProxmoxAPI | None = None
self._node_name: str = ""
self._ssh_client: "paramiko.SSHClient | None" = None
# paramiko.SSHClient; paramiko is imported lazily (see _node_ssh).
self._ssh_client: Any = None
# Candidate config (merge/replace)
self._candidate_config: str = ""
@@ -211,9 +211,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 +229,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"),
@@ -262,7 +258,7 @@ class ProxmoxDriver(
try:
self._ssh_client.close()
except Exception:
pass
logger.debug("%s: closing the SSH session: ignored", self.hostname, exc_info=True)
self._ssh_client = None
def is_alive(self) -> _JsonDict:
@@ -279,7 +275,7 @@ class ProxmoxDriver(
self._api.version.get()
alive = True
except Exception:
pass
logger.debug("%s: liveness probe failed", self.hostname, exc_info=True)
return {"is_alive": alive}
# ------------------------------------------------------------------ #
@@ -305,7 +301,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 +354,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
@@ -436,12 +433,9 @@ class ProxmoxDriver(
uptime = float(status.get("uptime", 0))
dns_search = dns.get("search", "")
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
+9 -8
View File
@@ -174,8 +174,13 @@ class ProxmoxInterfaceMixin:
parts = line.split()
if len(parts) < 6:
continue
ip_addr, _, flags, mac, _, iface = (
parts[0], parts[1], parts[2], parts[3], parts[4], parts[5]
ip_addr, _, _flags, mac, _, iface = (
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/py.typed
View File
Whitespace-only changes.
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
+53 -47
View File
@@ -12,7 +12,10 @@
# See the License for the specific language governing permissions and
# limitations under the License.
"""System-level NAPALM getters for Proxmox VE nodes (environment, NTP, SNMP, users, packages, services, updates, disk SMART)."""
"""System-level NAPALM getters for Proxmox VE nodes.
Environment, NTP, SNMP, users, packages, services, updates and disk SMART.
"""
from __future__ import annotations
@@ -25,8 +28,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 +178,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 = ""
@@ -228,7 +227,6 @@ class ProxmoxSystemMixin:
if not uid:
continue
# Proxmox roles: Administrator -> 15, otherwise 1
groups = user.get("groups", "") or ""
level = 1
try:
roles = self._api.access.users(uid).get() or {} # type: ignore[union-attr]
@@ -238,7 +236,8 @@ class ProxmoxSystemMixin:
logger.debug("Failed to fetch roles for user %s: %s", uid, exc)
result[uid] = {
"level": level,
"password": "",
# NAPALM's get_users shape; the password is never read.
"password": "", # nosec B105
"sshkeys": [],
}
@@ -256,7 +255,8 @@ class ProxmoxSystemMixin:
if uname not in result and uid_int < 1000 or uid_int == 0:
result[uname] = {
"level": 15 if uid_int == 0 else 0,
"password": "",
# NAPALM's get_users shape; the password is never read.
"password": "", # nosec B105
"sshkeys": [],
}
@@ -344,17 +344,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,11 +445,12 @@ 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(
f"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y {pkg_args} 2>&1 || true"
"DEBIAN_FRONTEND=noninteractive apt-get install --only-upgrade -y "
f"{pkg_args} 2>&1 || true"
)
return {"success": True, "output": output}
@@ -477,7 +480,8 @@ class ProxmoxSystemMixin:
port = 161
try:
conf = self._exec_ssh_command(
"grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf 2>/dev/null | head -5"
"grep -E '^[[:space:]]*(ro|rw)?community' /etc/snmp/snmpd.conf "
"2>/dev/null | head -5"
)
for line in conf.splitlines():
parts = line.split()
@@ -526,9 +530,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 +561,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 +618,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 +637,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 +654,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 +702,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 +741,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 +777,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):
@@ -810,7 +815,8 @@ class ProxmoxSystemMixin:
"timezone": None,
"ntp_servers": [],
"ssh_port": 22,
"ssh_password_auth": False,
# A flag, not a password.
"ssh_password_auth": False, # nosec B105
}
# Cluster name and node list from /cluster/status
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
+55 -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,16 @@ 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
+11 -11
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):
@@ -939,16 +939,16 @@ class ProxmoxVMProvisionMixin:
raise
except Exception as 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}") from 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
+51 -21
View File
@@ -1,28 +1,26 @@
[build-system]
requires = ["setuptools>=61", "wheel"]
requires = ["setuptools>=77"]
build-backend = "setuptools.build_meta"
[project]
name = "napalm-proxmox"
version = "0.7.0"
version = "0.8.0"
description = "NAPALM driver for Proxmox VE (classic Linux networking, SDN, OVS)"
readme = "README.md"
license = { text = "Apache-2.0" }
license = "Apache-2.0"
authors = [{ name = "Christian Manivong" }]
keywords = ["napalm", "proxmox", "network", "sdn", "ovs", "automation"]
classifiers = [
"Development Status :: 3 - Alpha",
"Intended Audience :: Developers",
"Intended Audience :: System Administrators",
"License :: OSI Approved :: Apache Software License",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Topic :: System :: Networking",
]
requires-python = ">=3.9"
requires-python = ">=3.10"
dependencies = [
"napalm>=5.0.0",
"napalm_device_types>=4.22.0",
@@ -33,34 +31,66 @@ dependencies = [
]
[project.optional-dependencies]
dev = [
"pytest>=7.4",
"pytest-cov>=4.1",
"pytest-mock>=3.12",
"responses>=0.24",
"ruff>=0.1",
"mypy>=1.7",
]
dev = ["pytest>=8", "pytest-cov>=5", "mypy>=1.11", "ruff>=0.6"]
[project.entry-points."napalm.drivers"]
proxmox = "napalm_proxmox:ProxmoxDriver"
[project.urls]
Repository = "https://git.netork.io/NAPALM/napalm-proxmox"
Issues = "https://git.netork.io/NAPALM/napalm-proxmox/issues"
Changelog = "https://git.netork.io/NAPALM/napalm-proxmox/src/branch/main/CHANGELOG.md"
[tool.setuptools.packages.find]
where = ["."]
include = ["napalm_proxmox*"]
[tool.setuptools.package-data]
napalm_proxmox = ["py.typed"]
[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.mypy]
python_version = "3.10"
strict = true
# napalm itself is untyped, so every driver subclasses Any.
disallow_subclassing_any = false
[[tool.mypy.overrides]]
module = ["napalm.*", "proxmoxer.*", "paramiko.*", "netaddr.*", "yaml.*"]
ignore_missing_imports = true
[[tool.mypy.overrides]]
# Not yet under strict mypy (quality baseline); may only shrink.
module = [
"napalm_proxmox.config_mixin",
"napalm_proxmox.driver",
"napalm_proxmox.interfaces_mixin",
"napalm_proxmox.lldp_mixin",
"napalm_proxmox.routing_mixin",
"napalm_proxmox.sdn_mixin",
"napalm_proxmox.system_mixin",
"napalm_proxmox.utils",
"napalm_proxmox.vm_contract_mixin",
"napalm_proxmox.vm_mixin",
"napalm_proxmox.vm_provision_mixin",
]
ignore_errors = true
[tool.pytest.ini_options]
testpaths = ["tests"]
addopts = "--cov=napalm_proxmox --cov-report=term-missing"
[tool.mypy]
python_version = "3.9"
warn_return_any = true
warn_unused_configs = true
[tool.coverage.run]
source = ["napalm_proxmox"]
[tool.coverage.report]
fail_under = 68
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/quality_baseline.json
+65
View File
@@ -0,0 +1,65 @@
{
"coverage_floor": 68.0,
"function_complexity": {
"napalm_proxmox/driver.py::ProxmoxDriver._resolve_node": 13,
"napalm_proxmox/driver.py::ProxmoxDriver.get_facts": 16,
"napalm_proxmox/interfaces_mixin.py::ProxmoxInterfaceMixin.get_interfaces_counters": 14,
"napalm_proxmox/interfaces_mixin.py::ProxmoxInterfaceMixin.get_mac_address_table": 22,
"napalm_proxmox/routing_mixin.py::ProxmoxRoutingMixin.get_route_to": 13,
"napalm_proxmox/routing_mixin.py::ProxmoxRoutingMixin.ping": 12,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._find_switch_uplink": 13,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._get_ovs_bridge_for_port": 13,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._get_vm_vlan_tags": 11,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._is_cluster_master": 13,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.get_network_instances": 13,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.get_vlans": 16,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.set_vlan": 13,
"napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_disk_smart": 22,
"napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_packages": 16,
"napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_users": 15,
"napalm_proxmox/utils.py::addresses_from_node_network": 15,
"napalm_proxmox/vm_contract_mixin.py::parse_vm_config": 16,
"napalm_proxmox/vm_mixin.py::ProxmoxVMMixin._get_vm_disk_and_boot": 16,
"napalm_proxmox/vm_mixin.py::ProxmoxVMMixin.get_vm_interfaces": 37,
"napalm_proxmox/vm_mixin.py::ProxmoxVMMixin.get_vms": 21,
"napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.create_vm_from_cloud_init": 26,
"napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.get_vm_status": 16
},
"function_lines": {
"napalm_proxmox/driver.py::ProxmoxDriver.get_facts": 82,
"napalm_proxmox/interfaces_mixin.py::ProxmoxInterfaceMixin.get_mac_address_table": 71,
"napalm_proxmox/lldp_mixin.py::ProxmoxLLDPMixin.get_lldp_neighbors_detail": 52,
"napalm_proxmox/routing_mixin.py::ProxmoxRoutingMixin.ping": 78,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin._get_vm_vlan_tags": 54,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.get_vlans": 66,
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.set_vlan": 70,
"napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_disk_smart": 74,
"napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_packages": 66,
"napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_users": 51,
"napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.set_hostname": 124,
"napalm_proxmox/vm_mixin.py::ProxmoxVMMixin.get_vm_interfaces": 179,
"napalm_proxmox/vm_mixin.py::ProxmoxVMMixin.get_vms": 114,
"napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin._download_cloud_image": 61,
"napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.create_vm_from_cloud_init": 232,
"napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.destroy_vm": 65,
"napalm_proxmox/vm_provision_mixin.py::ProxmoxVMProvisionMixin.get_vm_status": 92
},
"mypy_relaxed": [
"napalm_proxmox.config_mixin",
"napalm_proxmox.driver",
"napalm_proxmox.interfaces_mixin",
"napalm_proxmox.lldp_mixin",
"napalm_proxmox.routing_mixin",
"napalm_proxmox.sdn_mixin",
"napalm_proxmox.system_mixin",
"napalm_proxmox.utils",
"napalm_proxmox.vm_contract_mixin",
"napalm_proxmox.vm_mixin",
"napalm_proxmox.vm_provision_mixin"
],
"python_file_lines": {},
"untyped_public": [
"napalm_proxmox/sdn_mixin.py::ProxmoxSDNMixin.set_vlan",
"napalm_proxmox/system_mixin.py::ProxmoxSystemMixin.get_snmp_config"
]
}
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
+5 -4
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:
@@ -85,7 +86,7 @@ class TestGetUsers:
def test_user_has_required_keys(self, driver):
driver._node_api().execute.post.return_value = {"data": ""}
result = driver.get_users()
for user, data in result.items():
for _user, data in result.items():
assert "level" in data
assert "password" in data
assert "sshkeys" in data
@@ -248,7 +249,7 @@ class TestGetRouteTo:
def test_route_fields(self, driver):
driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
result = driver.get_route_to()
for prefix, routes in result.items():
for _prefix, routes in result.items():
for route in routes:
assert "protocol" in route
assert "next_hop" in route
@@ -257,7 +258,7 @@ class TestGetRouteTo:
def test_protocol_filter(self, driver):
driver._node_api().execute.post.return_value = {"data": self.ROUTE_OUTPUT}
result = driver.get_route_to(protocol="dhcp")
for prefix, routes in result.items():
for _prefix, routes in result.items():
for route in routes:
assert "dhcp" in route["protocol"].lower()
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