From de00041b469cbafd9401956623812a768e81da92 Mon Sep 17 00:00:00 2001 From: Christian Manivong Date: Fri, 23 Oct 2026 08:32:35 +0200 Subject: [PATCH] style: ruff format and the lint rules of the driver standard Mechanical: ruff format, and ruff check --fix with the rule set every NAPALM driver uses (E, F, I, UP, B at line length 100, Python 3.10). What ruff cannot fix comes in the next commit. --- napalm_bsd/base.py | 41 ++++++++++++++++++++++++----------------- napalm_bsd/parse.py | 16 ++++++++-------- pyproject.toml | 7 +++++++ tests/test_drivers.py | 6 +++--- 4 files changed, 42 insertions(+), 28 deletions(-) diff --git a/napalm_bsd/base.py b/napalm_bsd/base.py index 15ee69c..e9e9e8c 100644 --- a/napalm_bsd/base.py +++ b/napalm_bsd/base.py @@ -11,22 +11,21 @@ from __future__ import annotations import ipaddress import logging import re -import socket from shlex import quote -from typing import Any, Optional +from typing import Any import paramiko from napalm.base.exceptions import ConnectionClosedException, ConnectionException from napalm_device_types import EnsurePackageMixin, OSDriver, SnmpAgentMixin -from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict -from napalm_device_types.host_status import HostStatusMixin -from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch from napalm_device_types.channel import ( ByteStream, CommandResult, open_stream_on_transport, run_on_transport, ) +from napalm_device_types.host_status import HostStatusMixin +from napalm_device_types.models import FixStepOutcomeDict, SnmpAgentSettingsDict +from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch from napalm_bsd import parse @@ -63,6 +62,7 @@ _STAGE = ( '{ cat > "$t" && printf %s "$t" || { s=$?; rm -f -- "$t"; exit $s; }; }' ) + def _source(value: str) -> str: """An allowed source as an address or network, or ValueError.""" return str(ipaddress.ip_network(value.strip(), strict=False)) @@ -113,7 +113,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): username: str, password: str, timeout: int = 60, - optional_args: Optional[dict] = None, + optional_args: dict | None = None, ) -> None: self.hostname = hostname self.username = username @@ -121,14 +121,14 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): self.timeout = timeout args = optional_args or {} self.port: int = int(args.get("port", 22)) - self._key_file: Optional[str] = args.get("key_file") - self._sudo_password: Optional[str] = args.get("sudo_password") + self._key_file: str | None = args.get("key_file") + self._sudo_password: str | None = args.get("sudo_password") self._allow_agent: bool = bool(args.get("allow_agent", False)) self._look_for_keys: bool = bool(args.get("look_for_keys", False)) # The host key to expect, and where the one seen is reported (netork#1167). self._host_key = HostKeyCheck(args, hostname, self.port) - self._client: Optional[paramiko.SSHClient] = None - self._root: Optional[bool] = None + self._client: paramiko.SSHClient | None = None + self._root: bool | None = None # -- connection -------------------------------------------------------------- @@ -147,7 +147,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): except SshHostKeyMismatch: client.close() # refused before any login; nothing was sent raise - except (paramiko.SSHException, socket.error) as exc: + except (OSError, paramiko.SSHException) as exc: client.close() raise ConnectionException(f"SSH to {self.hostname} failed: {exc}") from exc self._client = client @@ -175,7 +175,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): self._root = run_on_transport(self._transport(), "id -u").stdout.strip() == "0" return self._root - def _privileged(self, command: str, privileged: bool) -> tuple[str, Optional[bytes]]: + def _privileged(self, command: str, privileged: bool) -> tuple[str, bytes | None]: """The command line that runs *command* as asked, and what goes to stdin first. With a sudo password, ``sudo -S`` reads it from stdin, so it never @@ -194,7 +194,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): *, privileged: bool = False, timeout: float = 60, - stdin: Optional[bytes] = None, + stdin: bytes | None = None, ) -> CommandResult: """Run *command* on an exec channel: no PTY, stderr apart, a real exit code. @@ -217,7 +217,11 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): staged = self.run_command(_STAGE, timeout=30, stdin=content.encode()) lines = staged.stdout.strip().splitlines() tmp = lines[-1] if lines else "" - if staged.exit_code != 0 or not tmp.startswith(_STAGE_PREFIX) or "/" in tmp[len(_STAGE_PREFIX):]: + if ( + staged.exit_code != 0 + or not tmp.startswith(_STAGE_PREFIX) + or "/" in tmp[len(_STAGE_PREFIX) :] + ): reason = (staged.stderr or staged.stdout).strip() or f"could not stage {path}" return CommandResult(reason, "", staged.exit_code or 1) tmp_q, part_q, path_q = (quote(p) for p in (tmp, f"{path}.napalm-tmp", path)) @@ -474,7 +478,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): #: net-snmp's daemon; OpenBSD's base snmpd is /usr/sbin/snmpd. SNMPD_DAEMON = "/usr/local/sbin/snmpd" - def get_management_peer(self) -> Optional[str]: + def get_management_peer(self) -> str | None: """The address this host sees the SSH connection come from (``SSH_CONNECTION``).""" out = self.run_command(_PEER_COMMAND, timeout=15).stdout.split() try: @@ -488,7 +492,10 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): def _step(self, command: str, done: str) -> FixStepOutcomeDict: result = self.run_command(command, privileged=True, timeout=600) output = "\n".join(filter(None, (result.stdout.strip(), result.stderr.strip()))) - return {"ok": result.exit_code == 0, "output": done if result.exit_code == 0 else output[-300:]} + return { + "ok": result.exit_code == 0, + "output": done if result.exit_code == 0 else output[-300:], + } def _install_snmp_agent(self) -> FixStepOutcomeDict: return self._step(self.INSTALL_COMMAND.format(name="net-snmp"), "net-snmp installed") @@ -507,7 +514,7 @@ class BsdDriver(SnmpAgentMixin, EnsurePackageMixin, HostStatusMixin, OSDriver): def _start_snmp_agent(self) -> FixStepOutcomeDict: return self._step(self.SNMPD_START, "snmpd enabled and running") - def get_snmp_config(self) -> Optional[dict[str, Any]]: + def get_snmp_config(self) -> dict[str, Any] | None: """net-snmp's community and port, if its daemon runs; None otherwise.""" if not self._out(f"pgrep -f {self.SNMPD_DAEMON}"): return None diff --git a/napalm_bsd/parse.py b/napalm_bsd/parse.py index 24a38b3..a8968a3 100644 --- a/napalm_bsd/parse.py +++ b/napalm_bsd/parse.py @@ -9,7 +9,7 @@ from __future__ import annotations import ipaddress import re -from typing import Any, Optional +from typing import Any _IFACE_HEADER = re.compile( r"^(?P[^\s:]+): flags=[0-9a-f]+<(?P[^>]*)>(?:.*?\bmtu (?P\d+))?" @@ -26,7 +26,7 @@ def _prefix_from_netmask(mask: str) -> int: def ifconfig(text: str) -> dict[str, dict[str, Any]]: """``ifconfig -a`` -> per interface: flags, MTU, MAC, state, speed, addresses.""" interfaces: dict[str, dict[str, Any]] = {} - current: Optional[dict[str, Any]] = None + current: dict[str, Any] | None = None for line in text.splitlines(): header = _IFACE_HEADER.match(line) if header: @@ -69,7 +69,7 @@ def ifconfig(text: str) -> dict[str, dict[str, Any]]: return interfaces -def _network(destination: str, family: str) -> Optional[str]: +def _network(destination: str, family: str) -> str | None: """A netstat destination as a network: ``default``, ``10.0.2/24``, ``fe80::%em0/64``.""" if destination == "default": return "0.0.0.0/0" if family == "ipv4" else "::/0" @@ -93,7 +93,7 @@ def routes(text: str) -> list[dict[str, str]]: """ result: list[dict[str, str]] = [] family = "" - iface_column: Optional[int] = None + iface_column: int | None = None for line in text.splitlines(): words = line.split() if not words: @@ -253,15 +253,15 @@ def crontab(text: str, *, system: bool = False, user: str = "") -> list[dict[str return jobs -def boottime(text: str) -> Optional[int]: +def boottime(text: str) -> int | None: """``sysctl -n kern.boottime``: FreeBSD ``{ sec = N, … }``, OpenBSD ``N``.""" match = re.search(r"sec = (\d+)", text) or re.fullmatch(r"\s*(\d+)\s*", text) return int(match[1]) if match else None def _socket( - proto: str, local: str, family: str, process: Optional[str], pid: Optional[int] -) -> Optional[dict[str, Any]]: + proto: str, local: str, family: str, process: str | None, pid: int | None +) -> dict[str, Any] | None: """A ListeningSocketDict from a local endpoint: ``*:22``, ``[::1]:25``, ``127.0.0.1:25``, ``[fe80::1%lo0]:25`` (sockstat, fstat) or ``*.22``, ``::1.25``, ``fe80::1%lo0.25`` (netstat).""" @@ -288,7 +288,7 @@ def _socket( } -def _unique(sockets: list[Optional[dict[str, Any]]]) -> list[dict[str, Any]]: +def _unique(sockets: list[dict[str, Any] | None]) -> list[dict[str, Any]]: """One entry per socket, the first process holding it, as ss reports it.""" seen: set[tuple] = set() result = [] diff --git a/pyproject.toml b/pyproject.toml index 75f71aa..48be403 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,3 +47,10 @@ include = ["napalm_bsd*"] [tool.ruff] line-length = 100 target-version = "py310" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP", "B"] + +[tool.ruff.lint.per-file-ignores] +# Recorded device output keeps its lines. +"tests/**" = ["E501"] diff --git a/tests/test_drivers.py b/tests/test_drivers.py index ef5559b..cb9c003 100644 --- a/tests/test_drivers.py +++ b/tests/test_drivers.py @@ -8,8 +8,7 @@ import pytest from napalm.base import NetworkDriver from napalm_device_types import OSDriver -from napalm_bsd import FreeBSDDriver, OpenBSDDriver -from napalm_bsd import base +from napalm_bsd import FreeBSDDriver, OpenBSDDriver, base from napalm_bsd.base import BsdDriver FIXTURES = Path(__file__).parent / "fixtures" @@ -600,7 +599,8 @@ class TestPrivilegedStdin: def _driver(self, monkeypatch, *, sudo_password, root=False): sent = [] monkeypatch.setattr( - base, "run_on_transport", + base, + "run_on_transport", lambda transport, line, stdin=None, timeout=60: sent.append((line, stdin)) or _Result(), ) args = {"sudo_password": sudo_password} if sudo_password else {}