diff --git a/napalm_tplink_jetstream/tplink_jetstream.py b/napalm_tplink_jetstream/tplink_jetstream.py index 76921fd..1336b2b 100644 --- a/napalm_tplink_jetstream/tplink_jetstream.py +++ b/napalm_tplink_jetstream/tplink_jetstream.py @@ -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 [] diff --git a/pyproject.toml b/pyproject.toml index 7f85cc4..9cda733 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,4 +50,11 @@ include = ["napalm_tplink_jetstream*"] [tool.ruff] line-length = 100 -target-version = "py38" +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/unit/test_driver.py b/tests/unit/test_driver.py index 5ea81c2..b794051 100644 --- a/tests/unit/test_driver.py +++ b/tests/unit/test_driver.py @@ -1,15 +1,16 @@ """Unit tests for TPLinkJetstreamDriver — no real device required.""" -import pytest from unittest.mock import MagicMock, patch -from napalm_tplink_jetstream.tplink_jetstream import TPLinkJetstreamDriver +import pytest +from napalm_tplink_jetstream.tplink_jetstream import TPLinkJetstreamDriver # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- + @pytest.fixture def driver(): """Return a driver instance with a mocked Netmiko connection.""" @@ -83,12 +84,11 @@ Gi1/0/1 00-AA-BB-CC-DD-EE Gi0/1 10.0.0.1 uplink """ - - # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- + class TestGetFacts: def test_returns_required_keys(self, driver): driver.device.send_command.return_value = SHOW_SYSTEM_INFO @@ -100,8 +100,16 @@ class TestGetFacts: ) facts = driver.get_facts() - for key in ("vendor", "model", "hostname", "os_version", "serial_number", - "uptime", "interface_list", "fqdn"): + for key in ( + "vendor", + "model", + "hostname", + "os_version", + "serial_number", + "uptime", + "interface_list", + "fqdn", + ): assert key in facts def test_vendor(self, driver): @@ -257,11 +265,14 @@ class TestIsAlive: class TestUptimeParser: - @pytest.mark.parametrize("uptime_str,expected", [ - ("0 day(s) 0 hour(s) 0 min(s) 0 sec(s)", 0), - ("1 day(s) 0 hour(s) 0 min(s) 0 sec(s)", 86400), - ("5 day(s) 2 hour(s) 35 min(s) 16 sec(s)", 5 * 86400 + 2 * 3600 + 35 * 60 + 16), - ]) + @pytest.mark.parametrize( + "uptime_str,expected", + [ + ("0 day(s) 0 hour(s) 0 min(s) 0 sec(s)", 0), + ("1 day(s) 0 hour(s) 0 min(s) 0 sec(s)", 86400), + ("5 day(s) 2 hour(s) 35 min(s) 16 sec(s)", 5 * 86400 + 2 * 3600 + 35 * 60 + 16), + ], + ) def test_parse_uptime_seconds(self, uptime_str, expected): result = TPLinkJetstreamDriver._parse_uptime_seconds(uptime_str) assert result == float(expected) diff --git a/tests/unit/test_snmp_agent.py b/tests/unit/test_snmp_agent.py index c1e8fce..48d5f23 100644 --- a/tests/unit/test_snmp_agent.py +++ b/tests/unit/test_snmp_agent.py @@ -35,9 +35,11 @@ def driver(): d._exit_config_mode = MagicMock() d._save_config = MagicMock() d._send_command = MagicMock( - side_effect=lambda cmd: SHOW_IP_INTERFACE - if cmd == "show ip interface" - else "Index Community-Name Access\n1 c0mm read-only\n" + side_effect=lambda cmd: ( + SHOW_IP_INTERFACE + if cmd == "show ip interface" + else "Index Community-Name Access\n1 c0mm read-only\n" + ) ) return d diff --git a/tests/unit/test_ssh_host_key.py b/tests/unit/test_ssh_host_key.py index c6a1afb..5298e3f 100644 --- a/tests/unit/test_ssh_host_key.py +++ b/tests/unit/test_ssh_host_key.py @@ -24,7 +24,9 @@ HANDLER = "napalm_tplink_jetstream.tplink_jetstream.ConnectHandler" def _device() -> MagicMock: device = MagicMock() - device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = SERVER_KEY + device.remote_conn_pre.get_transport.return_value.get_remote_server_key.return_value = ( + SERVER_KEY + ) return device