diff --git a/.gitea/workflows/ci.yml b/.gitea/workflows/ci.yml new file mode 100644 index 0000000..34f9b57 --- /dev/null +++ b/.gitea/workflows/ci.yml @@ -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 }} diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml deleted file mode 100644 index a7bc4d4..0000000 --- a/.github/workflows/ci.yml +++ /dev/null @@ -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/* \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 7dc9fd7..b21fd40 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/README.md b/README.md index 48bfbba..c99de0a 100644 --- a/README.md +++ b/README.md @@ -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 }, ) diff --git a/napalm_proxmox/config_mixin.py b/napalm_proxmox/config_mixin.py index a3baa9d..fc72eba 100644 --- a/napalm_proxmox/config_mixin.py +++ b/napalm_proxmox/config_mixin.py @@ -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 diff --git a/napalm_proxmox/driver.py b/napalm_proxmox/driver.py index fdff1ec..f759947 100644 --- a/napalm_proxmox/driver.py +++ b/napalm_proxmox/driver.py @@ -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} - - diff --git a/napalm_proxmox/interfaces_mixin.py b/napalm_proxmox/interfaces_mixin.py index 782f194..c8a19f9 100644 --- a/napalm_proxmox/interfaces_mixin.py +++ b/napalm_proxmox/interfaces_mixin.py @@ -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: diff --git a/napalm_proxmox/lldp_mixin.py b/napalm_proxmox/lldp_mixin.py index 83da0da..cd10613 100644 --- a/napalm_proxmox/lldp_mixin.py +++ b/napalm_proxmox/lldp_mixin.py @@ -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 = {} diff --git a/napalm_proxmox/py.typed b/napalm_proxmox/py.typed new file mode 100644 index 0000000..e69de29 diff --git a/napalm_proxmox/routing_mixin.py b/napalm_proxmox/routing_mixin.py index 6b8808c..edf474e 100644 --- a/napalm_proxmox/routing_mixin.py +++ b/napalm_proxmox/routing_mixin.py @@ -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": { diff --git a/napalm_proxmox/sdn_mixin.py b/napalm_proxmox/sdn_mixin.py index f77143e..80bd327 100644 --- a/napalm_proxmox/sdn_mixin.py +++ b/napalm_proxmox/sdn_mixin.py @@ -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() diff --git a/napalm_proxmox/system_mixin.py b/napalm_proxmox/system_mixin.py index ce962ef..0ce4cb6 100644 --- a/napalm_proxmox/system_mixin.py +++ b/napalm_proxmox/system_mixin.py @@ -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 diff --git a/napalm_proxmox/utils.py b/napalm_proxmox/utils.py index 79446ad..63cbcff 100644 --- a/napalm_proxmox/utils.py +++ b/napalm_proxmox/utils.py @@ -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. diff --git a/napalm_proxmox/vm_mixin.py b/napalm_proxmox/vm_mixin.py index 6aaf0b5..b4379f9 100644 --- a/napalm_proxmox/vm_mixin.py +++ b/napalm_proxmox/vm_mixin.py @@ -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: diff --git a/napalm_proxmox/vm_provision_mixin.py b/napalm_proxmox/vm_provision_mixin.py index e29024c..6f339cc 100644 --- a/napalm_proxmox/vm_provision_mixin.py +++ b/napalm_proxmox/vm_provision_mixin.py @@ -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") diff --git a/pyproject.toml b/pyproject.toml index a4b7e63..875fafd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -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 diff --git a/tests/conftest.py b/tests/conftest.py index 3c9b320..b927a36 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -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, diff --git a/tests/quality_baseline.json b/tests/quality_baseline.json new file mode 100644 index 0000000..988e73a --- /dev/null +++ b/tests/quality_baseline.json @@ -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" + ] +} diff --git a/tests/test_connection.py b/tests/test_connection.py index 7687c3f..62dc64a 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -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): diff --git a/tests/test_environment.py b/tests/test_environment.py index 5dce8e9..c97bb42 100644 --- a/tests/test_environment.py +++ b/tests/test_environment.py @@ -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) diff --git a/tests/test_get_facts.py b/tests/test_get_facts.py index 6514472..3c54327 100644 --- a/tests/test_get_facts.py +++ b/tests/test_get_facts.py @@ -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 diff --git a/tests/test_interfaces.py b/tests/test_interfaces.py index c646847..7a02ca4 100644 --- a/tests/test_interfaces.py +++ b/tests/test_interfaces.py @@ -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}" diff --git a/tests/test_kernel_facts.py b/tests/test_kernel_facts.py index 645bbd7..1e932d4 100644 --- a/tests/test_kernel_facts.py +++ b/tests/test_kernel_facts.py @@ -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 = ( diff --git a/tests/test_licenses.py b/tests/test_licenses.py index a85e123..d057044 100644 --- a/tests/test_licenses.py +++ b/tests/test_licenses.py @@ -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") + ) diff --git a/tests/test_listening_sockets.py b/tests/test_listening_sockets.py index 58359b3..15aac9a 100644 --- a/tests/test_listening_sockets.py +++ b/tests/test_listening_sockets.py @@ -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 = ( diff --git a/tests/test_misc.py b/tests/test_misc.py index d005032..a8fd6c2 100644 --- a/tests/test_misc.py +++ b/tests/test_misc.py @@ -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() diff --git a/tests/test_ovs_and_arp.py b/tests/test_ovs_and_arp.py index 32f9053..464f84c 100644 --- a/tests/test_ovs_and_arp.py +++ b/tests/test_ovs_and_arp.py @@ -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): diff --git a/tests/test_sdn.py b/tests/test_sdn.py index b24bed4..50da8fe 100644 --- a/tests/test_sdn.py +++ b/tests/test_sdn.py @@ -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}) diff --git a/tests/test_service_packages.py b/tests/test_service_packages.py index 95bae3e..7768b71 100644 --- a/tests/test_service_packages.py +++ b/tests/test_service_packages.py @@ -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 = ( diff --git a/tests/test_system_identity.py b/tests/test_system_identity.py index d837c10..0f1b620 100644 --- a/tests/test_system_identity.py +++ b/tests/test_system_identity.py @@ -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. diff --git a/tests/test_utils.py b/tests/test_utils.py index 90d8430..091f577 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -2,7 +2,6 @@ from __future__ import annotations -import pytest from napalm_proxmox import utils diff --git a/tests/test_vm_provision_mixin.py b/tests/test_vm_provision_mixin.py index b1b5f97..d3d9f0d 100644 --- a/tests/test_vm_provision_mixin.py +++ b/tests/test_vm_provision_mixin.py @@ -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