NAPALM/napalm-tplink-jetstream
1
0 Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases 1 Wiki Activity

style: ruff format and the lint rules of the driver standard

Browse Source 
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.
This commit is contained in:
Christian Manivong committed 2026-10-23 08:45:32 +02:00
1 parent 8b94818269
commit e53a4b4cc6
5 files changed
+119 -81
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

napalm_tplink_jetstream/tplink_jetstream.py
+81 -65
View File
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
@@ -20,26 +19,24 @@ Netmiko device_type: ``tplink_jetstream``
from __future__ import annotations
import re
import socket
from typing import Any, ClassVar
import napalm.base.constants as C
import netaddr
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SnmpAgentMixin, SwitchDriver
from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict
from napalm.base import helpers as napalm_helpers
from napalm.base.exceptions import (
ConnectionException,
ConnectionClosedException,
CommandErrorException,
ConnectionClosedException,
ConnectionException,
MergeConfigException,
ReplaceConfigException,
)
from napalm.base.netmiko_helpers import netmiko_args
from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SnmpAgentMixin, SwitchDriver
from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict
from napalm_device_types.ssh_host_keys import HostKeyCheck, SshHostKeyMismatch
import napalm.base.constants as C
from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
@@ -105,7 +102,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# Config management state
self._candidate_config: str | None = None
self._candidate_mode: str | None = None # 'merge' or 'replace'
self._candidate_mode: str | None = None # 'merge' or 'replace'
self._backup_config: str | None = None
# ------------------------------------------------------------------
@@ -130,13 +127,9 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
except SshHostKeyMismatch:
raise # refused before the login; nothing was sent
except NetmikoTimeoutException as exc:
raise ConnectionException(
f"Cannot connect to {self.hostname}: {exc}"
) from exc
raise ConnectionException(f"Cannot connect to {self.hostname}: {exc}") from exc
except NetmikoAuthenticationException as exc:
raise ConnectionException(
f"Authentication failed for {self.hostname}: {exc}"
) from exc
raise ConnectionException(f"Authentication failed for {self.hostname}: {exc}") from exc
def close(self) -> None:
"""Close the SSH connection."""
@@ -155,7 +148,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return {"is_alive": False}
try:
return {"is_alive": self.device.remote_conn.transport.is_active()}
except (socket.error, EOFError, AttributeError):
except (OSError, EOFError, AttributeError):
return {"is_alive": False}
# ------------------------------------------------------------------
@@ -190,7 +183,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return output
return output
return _do_send(command)
except (socket.error, EOFError) as exc:
except (OSError, EOFError) as exc:
raise ConnectionClosedException(str(exc)) from exc
# ------------------------------------------------------------------
@@ -320,14 +313,12 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
model = hw_version.split()[0] if hw_version else ""
# "Software Version" (Omada) or "Firmware Version" (classic)
os_version = (
self._parse_key_value(sys_info, "Software Version")
or self._parse_key_value(sys_info, "Firmware Version")
os_version = self._parse_key_value(sys_info, "Software Version") or self._parse_key_value(
sys_info, "Firmware Version"
)
# "Running Time" (Omada) or "Run Time" (classic)
uptime_str = (
self._parse_key_value(sys_info, "Running Time")
or self._parse_key_value(sys_info, "Run Time")
uptime_str = self._parse_key_value(sys_info, "Running Time") or self._parse_key_value(
sys_info, "Run Time"
)
uptime = self._parse_uptime_seconds(uptime_str)
@@ -358,7 +349,10 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
)
if match:
interfaces.append(match.group(1) + match.group(2))
return sorted(set(interfaces), key=lambda s: [int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)])
return sorted(
set(interfaces),
key=lambda s: [int(x) if x.isdigit() else x for x in re.split(r"(\d+)", s)],
)
def get_interfaces(self) -> dict[str, dict]:
"""Return a dictionary of interface details.
@@ -436,7 +430,9 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"lag_mode": lag_mode,
}
def _parse_interfaces(self, status_output: str, config_output: str = "", mtu: int = 1518) -> dict[str, dict]:
def _parse_interfaces(
self, status_output: str, config_output: str = "", mtu: int = 1518
) -> dict[str, dict]:
"""Parse tabular output of 'show interface status' and 'show interface configuration'."""
interfaces: dict[str, dict] = {}
@@ -451,8 +447,14 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
if not match:
continue
port, status, speed, duplex, _, _, _, _, description = (
match.group(1), match.group(2), match.group(3), match.group(4),
match.group(5), match.group(6), match.group(7), match.group(8),
match.group(1),
match.group(2),
match.group(3),
match.group(4),
match.group(5),
match.group(6),
match.group(7),
match.group(8),
match.group(9).strip(),
)
speed_val = 0.0
@@ -751,7 +753,9 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# Port Description (column 4+shift) may also be "gigabitEthernet X/X/X"
pd_idx = 4 + shift
desc_shift = 1 if (len(parts) > pd_idx and parts[pd_idx].lower() in PORT_TYPE_MAP) else 0
desc_shift = (
1 if (len(parts) > pd_idx and parts[pd_idx].lower() in PORT_TYPE_MAP) else 0
)
rows.append(
{
@@ -760,7 +764,9 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
"port_id": port_id,
"mgmt_address": parts[3 + shift] if len(parts) > 3 + shift else "",
"port_description": parts[pd_idx] if len(parts) > pd_idx else "",
"system_name": parts[5 + shift + desc_shift] if len(parts) > 5 + shift + desc_shift else "",
"system_name": parts[5 + shift + desc_shift]
if len(parts) > 5 + shift + desc_shift
else "",
}
)
@@ -1112,7 +1118,9 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
if self._candidate_config is None:
raise MergeConfigException("No candidate configuration is staged.")
ex_cls = ReplaceConfigException if self._candidate_mode == "replace" else MergeConfigException
ex_cls = (
ReplaceConfigException if self._candidate_mode == "replace" else MergeConfigException
)
# Save backup for potential rollback
self._backup_config = self._send_command("show running-config")
@@ -1334,12 +1342,18 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
counters[current_port] = current
current_port = m.group(1)
current = {
"tx_errors": 0, "rx_errors": 0,
"tx_discards": 0, "rx_discards": 0,
"tx_octets": 0, "rx_octets": 0,
"tx_unicast_packets": 0, "rx_unicast_packets": 0,
"tx_multicast_packets": 0, "rx_multicast_packets": 0,
"tx_broadcast_packets": 0, "rx_broadcast_packets": 0,
"tx_errors": 0,
"rx_errors": 0,
"tx_discards": 0,
"rx_discards": 0,
"tx_octets": 0,
"rx_octets": 0,
"tx_unicast_packets": 0,
"rx_unicast_packets": 0,
"tx_multicast_packets": 0,
"rx_multicast_packets": 0,
"tx_broadcast_packets": 0,
"rx_broadcast_packets": 0,
}
continue
@@ -1353,18 +1367,18 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
val = val.strip()
mapping = {
"Tx Errors": "tx_errors",
"Rx Errors": "rx_errors",
"Tx Discards": "tx_discards",
"Rx Discards": "rx_discards",
"Tx Bytes": "tx_octets",
"Rx Bytes": "rx_octets",
"Tx Ucast": "tx_unicast_packets",
"Rx Ucast": "rx_unicast_packets",
"Tx Mcast": "tx_multicast_packets",
"Rx Mcast": "rx_multicast_packets",
"Tx Bcast": "tx_broadcast_packets",
"Rx Bcast": "rx_broadcast_packets",
"Tx Errors": "tx_errors",
"Rx Errors": "rx_errors",
"Tx Discards": "tx_discards",
"Rx Discards": "rx_discards",
"Tx Bytes": "tx_octets",
"Rx Bytes": "rx_octets",
"Tx Ucast": "tx_unicast_packets",
"Rx Ucast": "rx_unicast_packets",
"Tx Mcast": "tx_multicast_packets",
"Rx Mcast": "rx_multicast_packets",
"Tx Bcast": "tx_broadcast_packets",
"Rx Bcast": "rx_broadcast_packets",
}
if key in mapping:
current[mapping[key]] = _int(val)
@@ -1400,7 +1414,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
if len(parts) < 3:
continue
try:
int(parts[0]) # first column is index
int(parts[0]) # first column is index
except ValueError:
continue
@@ -1583,9 +1597,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
lines.append("exit")
errors = self._apply_config_lines("\n".join(lines))
if errors:
raise CommandErrorException(
f"set_vlan({vlan_id}) errors: {errors}"
)
raise CommandErrorException(f"set_vlan({vlan_id}) errors: {errors}")
finally:
self._exit_config_mode()
self._save_config()
@@ -1628,14 +1640,14 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
if "access_vlan" in config:
for old_vid in old_untagged_vlans:
lines.append(f" no switchport general allowed vlan {old_vid}")
lines.append(f" switchport general allowed vlan {config['access_vlan']} untagged")
lines.append(
f" switchport general allowed vlan {config['access_vlan']} untagged"
)
lines.append(f" switchport pvid {config['access_vlan']}")
lines.append("exit")
errors = self._apply_config_lines("\n".join(lines))
if errors:
raise CommandErrorException(
f"set_interface({interface}) errors: {errors}"
)
raise CommandErrorException(f"set_interface({interface}) errors: {errors}")
finally:
self._exit_config_mode()
self._save_config()
@@ -1749,11 +1761,16 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
iface = self._to_cli_interface(interface)
lines = [f"interface {iface}"]
if "is_poe_enabled" in config:
lines.append("power inline supply" if config["is_poe_enabled"] else "no power inline supply")
lines.append(
"power inline supply" if config["is_poe_enabled"] else "no power inline supply"
)
if "poe_priority" in config:
priority = self._POE_PRIORITY_MAP_REV.get(config["poe_priority"], "low")
lines.append(f"power inline priority {priority}")
if "allocated_power_in_watts" in config and config.get("poe_allocation_method") == "PPAM_VALUE":
if (
"allocated_power_in_watts" in config
and config.get("poe_allocation_method") == "PPAM_VALUE"
):
lines.append(f"power inline power-limit {config['allocated_power_in_watts']}")
lines.append("exit")
@@ -1803,7 +1820,9 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
# Parse RTT: "Minimum = 0ms , Maximum = 10ms , Average = 2ms"
rtt_min = rtt_max = rtt_avg = 0.0
m = re.search(r"Minimum\s*=\s*(\d+)ms.*?Maximum\s*=\s*(\d+)ms.*?Average\s*=\s*(\d+)ms", output)
m = re.search(
r"Minimum\s*=\s*(\d+)ms.*?Maximum\s*=\s*(\d+)ms.*?Average\s*=\s*(\d+)ms", output
)
if m:
rtt_min = float(m.group(1))
rtt_max = float(m.group(2))
@@ -1895,9 +1914,7 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
) -> dict[str, str | dict[str, Any]]:
"""Execute a list of CLI commands and return their output."""
if encoding != "text":
raise NotImplementedError(
f"Encoding '{encoding}' is not supported by this driver."
)
raise NotImplementedError(f"Encoding '{encoding}' is not supported by this driver.")
result: dict[str, str | dict[str, Any]] = {}
for cmd in commands:
result[cmd] = self._send_command(cmd)
@@ -1937,7 +1954,6 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
return SNMPConfigDict(running=True, community=community, port=161, version="2c")
def get_device_warnings(self) -> list:
"""Return device warnings. Currently delegates to SNMP detection only."""
return []