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

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<name>[^\s:]+): flags=[0-9a-f]+<(?P<flags>[^>]*)>(?:.*?\bmtu (?P<mtu>\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 = []

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"]

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 {}