chore: adopt the NAPALM driver standard #9

Merged
christianmanivong merged 4 commits from chore/driver-standard into main 2026-10-23 09:57:17 +00:00
3 changed files with 16 additions and 9 deletions
Showing only changes of commit bef2f23ab2 - Show all commits

No files matched your search

+14 -7
View File
@@ -18,6 +18,7 @@ Netmiko device_type: ``tplink_jetstream``
from __future__ import annotations from __future__ import annotations
import logging
import re import re
from typing import Any, ClassVar from typing import Any, ClassVar
@@ -38,6 +39,11 @@ from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
from netmiko import ConnectHandler from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
logger = logging.getLogger(__name__)
#: get_users' password field: the switch never shows a password, so it stays empty.
_PASSWORD_NOT_SHOWN = "" # nosec B105
class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver): class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"""NAPALM driver for TP-Link Jetstream managed switches.""" """NAPALM driver for TP-Link Jetstream managed switches."""
@@ -372,7 +378,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
Port State Speed Duplex FlowCtrl Description Port State Speed Duplex FlowCtrl Description
Gi1/0/1 Enable Auto Auto Disable uplink Gi1/0/1 Enable Auto Auto Disable uplink
""" """ # noqa: E501 - recorded device output keeps its lines
jumbo_out = self._send_command("show jumbo") jumbo_out = self._send_command("show jumbo")
global_mtu = 1518 global_mtu = 1518
m = re.search(r"(\d+)", jumbo_out) m = re.search(r"(\d+)", jumbo_out)
@@ -446,7 +452,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
) )
if not match: if not match:
continue continue
port, status, speed, duplex, _, _, _, _, description = ( port, status, speed, _duplex, _, _, _, _, description = (
match.group(1), match.group(1),
match.group(2), match.group(2),
match.group(3), match.group(3),
@@ -703,7 +709,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
Port Device ID Port ID management address Port Description System Name Port Device ID Port ID management address Port Description System Name
---- ------------ -------- ------------------ ---------------- ----------- ---- ------------ -------- ------------------ ---------------- -----------
Gi1/0/9 64:E8:81:E1:29:00 2 10.7.0.10 2 swt-l0-1-10 Gi1/0/9 64:E8:81:E1:29:00 2 10.7.0.10 2 swt-l0-1-10
""" """ # noqa: E501 - recorded device output keeps its lines
neighbors: dict[str, list[dict]] = {} neighbors: dict[str, list[dict]] = {}
for row in self._get_lldp_table(): for row in self._get_lldp_table():
neighbors.setdefault(row["local_port"], []).append( neighbors.setdefault(row["local_port"], []).append(
@@ -1026,7 +1032,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
@staticmethod @staticmethod
def _to_cli_interface(port: str) -> str: def _to_cli_interface(port: str) -> str:
"""Convert abbreviated port name (e.g. ``Gi1/0/1``) to CLI form (``gigabitEthernet 1/0/1``).""" """Convert a short port name (``Gi1/0/1``) to CLI form (``gigabitEthernet 1/0/1``)."""
_PREFIX_MAP = { _PREFIX_MAP = {
"gi": "gigabitEthernet", "gi": "gigabitEthernet",
"te": "ten-gigabitEthernet", "te": "ten-gigabitEthernet",
@@ -1150,7 +1156,8 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
""" """
if self._backup_config is None: if self._backup_config is None:
raise CommandErrorException( raise CommandErrorException(
"No backup configuration available – commit_config has not been called in this session." "No backup configuration available – "
"commit_config has not been called in this session."
) )
current = self._send_command("show running-config") current = self._send_command("show running-config")
@@ -1423,7 +1430,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# Map TP-Link roles to NAPALM privilege levels (1-15) # Map TP-Link roles to NAPALM privilege levels (1-15)
level = 15 if role == "admin" else 1 level = 15 if role == "admin" else 1
users[username] = {"level": level, "password": "", "sshkeys": []} users[username] = {"level": level, "password": _PASSWORD_NOT_SHOWN, "sshkeys": []}
return users return users
@@ -1618,7 +1625,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
if interface in vlan_info.get("untagged", []) and vid_str != new_vid: if interface in vlan_info.get("untagged", []) and vid_str != new_vid:
old_untagged_vlans.append(vid_str) old_untagged_vlans.append(vid_str)
except Exception: except Exception:
pass logger.debug("reading the old untagged VLANs: ignored", exc_info=True)
self._enter_config_mode() self._enter_config_mode()
try: try:
+1 -1
View File
@@ -96,7 +96,7 @@ class TestGetFacts:
# _get_interface_list needs show interface # _get_interface_list needs show interface
driver.device.send_command.side_effect = lambda cmd, **kw: ( driver.device.send_command.side_effect = lambda cmd, **kw: (
SHOW_SYSTEM_INFO if "system-info" in cmd else SHOW_INTERFACE SHOW_SYSTEM_INFO if "system-info" in cmd else SHOW_INTERFACE_STATUS
) )
facts = driver.get_facts() facts = driver.get_facts()
+1 -1
View File
@@ -76,7 +76,7 @@ def test_a_changed_key_is_refused():
try: try:
raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY) raise paramiko.BadHostKeyException("10.0.0.8", SERVER_KEY, OTHER_KEY)
except paramiko.SSHException as exc: # what netmiko does with it except paramiko.SSHException as exc: # what netmiko does with it
raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") raise NetmikoTimeoutException(f"A paramiko SSHException occurred: {exc}") from exc
driver = _driver(seen, SshHostKey.of(OTHER_KEY)) driver = _driver(seen, SshHostKey.of(OTHER_KEY))
with patch(HANDLER, side_effect=refuse): with patch(HANDLER, side_effect=refuse):