diff --git a/napalm_tplink_jetstream/tplink_jetstream.py b/napalm_tplink_jetstream/tplink_jetstream.py index 30ba94e..7d0302b 100644 --- a/napalm_tplink_jetstream/tplink_jetstream.py +++ b/napalm_tplink_jetstream/tplink_jetstream.py @@ -17,15 +17,17 @@ Tested against: SG2210P, SG3210, T1500G, T1600G, T2600G, T3700G series. Netmiko device_type: ``tplink_jetstream`` """ +from __future__ import annotations + import re import socket -from typing import Dict, List, Optional, Union, Any +from typing import Any import netaddr from netmiko import ConnectHandler from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException -from napalm_device_types import SwitchDriver +from napalm_device_types import ConfigLifecycleMixin, SwitchDriver from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict from napalm.base import helpers as napalm_helpers from napalm.base.exceptions import ( @@ -39,7 +41,7 @@ from napalm.base.netmiko_helpers import netmiko_args import napalm.base.constants as C -class TPLinkJetstreamDriver(SwitchDriver): +class TPLinkJetstreamDriver(ConfigLifecycleMixin, SwitchDriver): """NAPALM driver for TP-Link Jetstream managed switches.""" VENDOR = "TP-Link" @@ -57,13 +59,13 @@ class TPLinkJetstreamDriver(SwitchDriver): username: str, password: str, timeout: int = 60, - optional_args: Optional[Dict] = None, + optional_args: dict | None = None, ) -> None: self.hostname = hostname self.username = username self.password = password self.timeout = timeout - self.device: Optional[ConnectHandler] = None + self.device: ConnectHandler | None = None if optional_args is None: optional_args = {} @@ -74,9 +76,9 @@ class TPLinkJetstreamDriver(SwitchDriver): self.netmiko_optional_args = netmiko_args(optional_args) # Config management state - self._candidate_config: Optional[str] = None - self._candidate_mode: Optional[str] = None # 'merge' or 'replace' - self._backup_config: Optional[str] = None + self._candidate_config: str | None = None + self._candidate_mode: str | None = None # 'merge' or 'replace' + self._backup_config: str | None = None # ------------------------------------------------------------------ # Connection management @@ -110,7 +112,7 @@ class TPLinkJetstreamDriver(SwitchDriver): self.device.disconnect() self.device = None - def is_alive(self) -> Dict[str, bool]: + def is_alive(self) -> dict[str, bool]: """Return connection liveness. Only checks the transport-level state – does NOT write anything to @@ -128,7 +130,7 @@ class TPLinkJetstreamDriver(SwitchDriver): # Internal helpers # ------------------------------------------------------------------ - def _send_command(self, command: Union[str, List[str]]) -> str: + def _send_command(self, command: str | list[str]) -> str: """Send a command (or list of candidate commands) to the device. When a list is supplied, commands are tried in order and the first @@ -194,14 +196,14 @@ class TPLinkJetstreamDriver(SwitchDriver): read_timeout=self.timeout, ) - def _apply_config_lines(self, config_text: str) -> List[str]: + def _apply_config_lines(self, config_text: str) -> list[str]: """Send config lines to the device while in config mode. Returns a list of error messages for any line that was rejected. Lines starting with ``!`` or ``#`` and blank lines are skipped. """ ep_any = self._any_prompt() - errors: List[str] = [] + errors: list[str] = [] for line in config_text.splitlines(): stripped = line.strip() if not stripped or stripped.startswith(("!", "#")): @@ -258,7 +260,7 @@ class TPLinkJetstreamDriver(SwitchDriver): # NAPALM getters # ------------------------------------------------------------------ - def get_facts(self) -> Dict: + def get_facts(self) -> dict: """Return a dictionary of general device facts. Runs ``show system-info`` and ``show interface`` to collect: @@ -312,7 +314,7 @@ class TPLinkJetstreamDriver(SwitchDriver): "interface_list": interface_list, } - def _get_interface_list(self) -> List[str]: + def _get_interface_list(self) -> list[str]: """Return a sorted list of interface names from ``show interface status``.""" output = self._send_command("show interface status") interfaces = [] @@ -326,7 +328,7 @@ class TPLinkJetstreamDriver(SwitchDriver): 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)]) - def get_interfaces(self) -> Dict[str, Dict]: + def get_interfaces(self) -> dict[str, dict]: """Return a dictionary of interface details. Each interface entry contains: @@ -357,7 +359,7 @@ class TPLinkJetstreamDriver(SwitchDriver): self._add_lag_info(interfaces) return interfaces - def _add_lag_info(self, interfaces: Dict[str, Dict]) -> None: + def _add_lag_info(self, interfaces: dict[str, dict]) -> None: """Enrich ``interfaces`` with LAG/trunk membership from ``show link-aggregation summary``. JetStream switches report one row per configured LAG, e.g.:: @@ -402,9 +404,9 @@ class TPLinkJetstreamDriver(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] = {} + interfaces: dict[str, dict] = {} # --- parse 'show interface status' --- # Columns: Port Status Speed Duplex FlowCtrl Active-Medium LAG Linkdown-Status Description @@ -465,7 +467,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return interfaces - def get_interfaces_ip(self) -> Dict[str, Dict]: + def get_interfaces_ip(self) -> dict[str, dict]: """Return all configured IP addresses grouped by interface. Runs ``show ip interface`` which produces a block per interface:: @@ -474,8 +476,8 @@ class TPLinkJetstreamDriver(SwitchDriver): Primary IP address is 172.22.8.15/24 """ output = self._send_command("show ip interface") - interfaces_ip: Dict[str, Dict] = {} - current_iface: Optional[str] = None + interfaces_ip: dict[str, dict] = {} + current_iface: str | None = None for line in output.splitlines(): # Detect interface header: "VLAN8 is up, ..." or "VLAN8 is down, ..." @@ -510,7 +512,7 @@ class TPLinkJetstreamDriver(SwitchDriver): full: bool = False, sanitized: bool = False, format: str = "text", - ) -> Dict[str, str]: + ) -> dict[str, str]: """Return running and/or startup configuration. TP-Link Jetstream does not support a candidate configuration; @@ -529,7 +531,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return configs - def get_arp_table(self, vrf: str = "") -> List[Dict]: + def get_arp_table(self, vrf: str = "") -> list[dict]: """Return the ARP table. Example ``show arp`` output:: @@ -585,7 +587,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return arp_table - def get_mac_address_table(self) -> List[Dict]: + def get_mac_address_table(self) -> list[dict]: """Return the MAC address table. Example ``show mac address-table`` output:: @@ -646,7 +648,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return mac_table - def get_lldp_neighbors(self) -> Dict[str, List[Dict]]: + def get_lldp_neighbors(self) -> dict[str, list[dict]]: """Return a dict of LLDP neighbors keyed by local port. Example ``show lldp neighbor-information`` output:: @@ -655,17 +657,17 @@ class TPLinkJetstreamDriver(SwitchDriver): ---- ------------ -------- ------------------ ---------------- ----------- Gi1/0/9 64:E8:81:E1:29:00 2 10.7.0.10 2 swt-l0-1-10 """ - neighbors: Dict[str, List[Dict]] = {} + neighbors: dict[str, list[dict]] = {} for row in self._get_lldp_table(): neighbors.setdefault(row["local_port"], []).append( {"hostname": row["system_name"], "port": row["port_id"]} ) return neighbors - def _get_lldp_table(self) -> List[Dict]: + def _get_lldp_table(self) -> list[dict]: """Parse ``show lldp neighbor-information`` into a list of row dicts.""" output = self._send_command("show lldp neighbor-information") - rows: List[Dict] = [] + rows: list[dict] = [] in_table = False for line in output.splitlines(): @@ -719,13 +721,13 @@ class TPLinkJetstreamDriver(SwitchDriver): return rows - def get_lldp_neighbors_detail(self, interface: str = "") -> Dict[str, List[Dict]]: + def get_lldp_neighbors_detail(self, interface: str = "") -> dict[str, list[dict]]: """Return detailed LLDP neighbor info. TP-Link does not support a per-port detail command; all available fields are extracted from ``show lldp neighbor-information``. """ - details: Dict[str, List[Dict]] = {} + details: dict[str, list[dict]] = {} for row in self._get_lldp_table(): if interface and row["local_port"] != interface: @@ -746,7 +748,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return details @staticmethod - def _parse_lldp_detail(output: str) -> Dict: + def _parse_lldp_detail(output: str) -> dict: """Parse a single ``show lldp neighbor-information interface`` block.""" defaults = { "parent_interface": "", @@ -784,7 +786,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return defaults - def get_vlans(self) -> Dict[str, Dict]: + def get_vlans(self) -> dict[str, dict]: """Return VLAN information. Example ``show vlan`` output:: @@ -797,8 +799,8 @@ class TPLinkJetstreamDriver(SwitchDriver): TG: Gi1/0/1, Gi1/0/9, Gi1/0/10 """ output = self._send_command("show vlan") - vlans: Dict[str, Dict] = {} - current_id: Optional[str] = None + vlans: dict[str, dict] = {} + current_id: str | None = None in_table = False for line in output.splitlines(): @@ -828,14 +830,14 @@ class TPLinkJetstreamDriver(SwitchDriver): return vlans @staticmethod - def _parse_vlan_ports(ports_raw: str) -> List[str]: + def _parse_vlan_ports(ports_raw: str) -> list[str]: """Parse TP-Link VLAN port string, stripping TG:/UT: prefixes and expanding ranges. Input examples:: "TG: Gi1/0/9, Gi1/0/10" "UT: Gi1/0/1-4, Gi1/0/7" """ - interfaces: List[str] = [] + interfaces: list[str] = [] # Split on "TG:" or "UT:" markers to get individual segments for segment in re.split(r"\b(?:TG|UT)\s*:", ports_raw, flags=re.I): segment = segment.strip() @@ -858,9 +860,9 @@ class TPLinkJetstreamDriver(SwitchDriver): return interfaces @staticmethod - def _expand_ports(ports_str: str) -> List[str]: + def _expand_ports(ports_str: str) -> list[str]: """Expand a comma-separated port string (without TG:/UT: markers).""" - result: List[str] = [] + result: list[str] = [] for token in ports_str.split(","): token = token.strip() if not token: @@ -886,8 +888,8 @@ class TPLinkJetstreamDriver(SwitchDriver): allows callers to pass the last seen marker so that wrap-around continuation lines are classified correctly. """ - tagged: List[str] = [] - untagged: List[str] = [] + tagged: list[str] = [] + untagged: list[str] = [] # re.split with a capturing group keeps the delimiters in the result list parts = re.split(r"\b(TG|UT)\s*:", ports_raw, flags=re.I) @@ -914,7 +916,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return tagged, untagged - def get_vlans_detail(self) -> Dict[str, Dict]: + def get_vlans_detail(self) -> dict[str, dict]: """Return VLAN information with tagged/untagged port separation. Returns:: @@ -925,8 +927,8 @@ class TPLinkJetstreamDriver(SwitchDriver): } """ output = self._send_command("show vlan") - vlans: Dict[str, Dict] = {} - current_id: Optional[str] = None + vlans: dict[str, dict] = {} + current_id: str | None = None in_table = False last_marker = "UT" # tracks last TG/UT seen; used for prefix-less continuation ports @@ -1047,93 +1049,10 @@ class TPLinkJetstreamDriver(SwitchDriver): # NAPALM configuration management # ------------------------------------------------------------------ - def load_merge_candidate( - self, filename: Optional[str] = None, config: Optional[str] = None - ) -> None: - """Stage a set of CLI configuration commands to be merged into the running config. + def _get_running_config(self) -> str: + return self._send_command("show running-config") - *config* is a plain-text string of CLI commands exactly as you would - type them in the switch's ``configure`` mode – one command per line, - including sub-mode entry (e.g. ``interface gigabitEthernet 1/0/1``) - and exit (``exit``). Blank lines and lines starting with ``!`` or - ``#`` are ignored. - - The configuration is **not** applied until :meth:`commit_config` is - called. - - :raises MergeConfigException: on invalid input. - """ - if filename is not None: - try: - with open(filename) as fh: - config = fh.read() - except OSError as exc: - raise MergeConfigException(str(exc)) from exc - if config is None: - raise MergeConfigException("Either 'filename' or 'config' must be provided.") - self._candidate_config = config - self._candidate_mode = "merge" - - def load_replace_candidate( - self, filename: Optional[str] = None, config: Optional[str] = None - ) -> None: - """Stage a full running-config replacement candidate. - - The candidate is a complete ``show running-config`` style text. - :meth:`compare_config` will show a unified diff against the current - running config. :meth:`commit_config` applies every non-comment line - from the candidate in config mode (additive / merge semantics – - TP-Link does not support atomic config-replace). Configuration that - exists on the device but is absent from the candidate will **not** be - removed automatically. - - :raises ReplaceConfigException: on invalid input. - """ - if filename is not None: - try: - with open(filename) as fh: - config = fh.read() - except OSError as exc: - raise ReplaceConfigException(str(exc)) from exc - if config is None: - raise ReplaceConfigException("Either 'filename' or 'config' must be provided.") - self._candidate_config = config - self._candidate_mode = "replace" - - def compare_config(self) -> str: - """Return a human-readable diff of the pending candidate vs running config. - - For a **merge** candidate: returns the staged commands prefixed with - ``+`` (they will all be added). - - For a **replace** candidate: returns a unified diff between the current - ``show running-config`` output and the candidate text. - - Returns an empty string when no candidate is staged. - """ - if self._candidate_config is None: - return "" - - if self._candidate_mode == "merge": - lines = [] - for line in self._candidate_config.splitlines(): - if line.strip() and not line.strip().startswith(("!", "#")): - lines.append(f"+{line}") - return "\n".join(lines) - - # replace mode – unified diff - import difflib - running = self._send_command("show running-config") - diff = difflib.unified_diff( - running.splitlines(), - self._candidate_config.splitlines(), - fromfile="running-config", - tofile="candidate-config", - lineterm="", - ) - return "\n".join(diff) - - def commit_config(self, message: str = "", revert_in: Optional[int] = None) -> None: + def commit_config(self, message: str = "", revert_in: int | None = None) -> None: """Apply the staged candidate configuration to the device and save it. 1. Saves the current running config as rollback backup. @@ -1153,7 +1072,7 @@ class TPLinkJetstreamDriver(SwitchDriver): # Save backup for potential rollback self._backup_config = self._send_command("show running-config") - errors: List[str] = [] + errors: list[str] = [] try: self._enter_config_mode() errors = self._apply_config_lines(self._candidate_config) @@ -1167,11 +1086,6 @@ class TPLinkJetstreamDriver(SwitchDriver): self._candidate_config = None self._candidate_mode = None - def discard_config(self) -> None: - """Discard the staged candidate configuration without applying it.""" - self._candidate_config = None - self._candidate_mode = None - def rollback(self) -> None: """Restore the running config to the state before the last :meth:`commit_config`. @@ -1200,7 +1114,7 @@ class TPLinkJetstreamDriver(SwitchDriver): self._backup_config = None @staticmethod - def _parse_config_blocks(config_text: str) -> Dict[str, List[str]]: + def _parse_config_blocks(config_text: str) -> dict[str, list[str]]: """Parse a running-config into a dict of context → [command lines]. The special key ``"__global__"`` holds top-level commands. @@ -1210,7 +1124,7 @@ class TPLinkJetstreamDriver(SwitchDriver): TP-Link uses bare ``#`` lines as block separators; lines starting with ``!`` are file-header comments. """ - blocks: Dict[str, List[str]] = {"__global__": []} + blocks: dict[str, list[str]] = {"__global__": []} ctx = "__global__" for line in config_text.splitlines(): @@ -1243,7 +1157,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return blocks @staticmethod - def _negate_command(cmd: str) -> Optional[str]: + def _negate_command(cmd: str) -> str | None: """Return the ``no`` form of *cmd*, or *None* if not known. Only commands whose entire effect is removed by ``no `` @@ -1273,12 +1187,12 @@ class TPLinkJetstreamDriver(SwitchDriver): return None - def _diff_to_commands(self, backup: str, current: str) -> List[str]: + def _diff_to_commands(self, backup: str, current: str) -> list[str]: """Generate the commands needed to revert *current* to *backup* state.""" backup_blocks = self._parse_config_blocks(backup) current_blocks = self._parse_config_blocks(current) - cmds: List[str] = [] + cmds: list[str] = [] all_ctxs = set(backup_blocks.keys()) | set(current_blocks.keys()) for ctx in sorted(all_ctxs): @@ -1311,11 +1225,7 @@ class TPLinkJetstreamDriver(SwitchDriver): # Additional NAPALM getters # ------------------------------------------------------------------ - def has_pending_commit(self) -> bool: - """Return True when a candidate configuration is staged but not yet committed.""" - return self._candidate_config is not None - - def get_environment(self) -> Dict: + def get_environment(self) -> dict: """Return device environment data (CPU, memory). TP-Link Jetstream switches do not expose fan, temperature, or power @@ -1355,7 +1265,7 @@ class TPLinkJetstreamDriver(SwitchDriver): }, } - def get_interfaces_counters(self) -> Dict[str, Dict]: + def get_interfaces_counters(self) -> dict[str, dict]: """Return per-interface packet and byte counters. Parses the block-format output of ``show interface counters``. @@ -1367,9 +1277,9 @@ class TPLinkJetstreamDriver(SwitchDriver): def _int(s: str) -> int: return int(s.replace(",", "")) if s.strip() else 0 - counters: Dict[str, Dict] = {} - current: Optional[Dict] = None - current_port: Optional[str] = None + counters: dict[str, dict] = {} + current: dict | None = None + current_port: str | None = None for line in output.splitlines(): # New port block @@ -1419,7 +1329,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return counters - def get_users(self) -> Dict[str, Dict]: + def get_users(self) -> dict[str, dict]: """Return local user accounts. Parses ``show user account`` output:: @@ -1428,7 +1338,7 @@ class TPLinkJetstreamDriver(SwitchDriver): 1 admin Admin """ output = self._send_command("show user account") - users: Dict[str, Dict] = {} + users: dict[str, dict] = {} in_table = False for line in output.splitlines(): @@ -1458,7 +1368,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return users - def get_snmp_information(self) -> Dict: + def get_snmp_information(self) -> dict: """Return SNMP configuration. Reads contact/location from ``show system-info`` and communities @@ -1471,7 +1381,7 @@ class TPLinkJetstreamDriver(SwitchDriver): mac = self._parse_key_value(sys_info, "Mac Address").replace("-", ":").upper() snmp_out = self._send_command("show snmp community") - communities: Dict[str, Dict] = {} + communities: dict[str, dict] = {} if "disabled" not in snmp_out.lower(): in_table = False @@ -1495,7 +1405,7 @@ class TPLinkJetstreamDriver(SwitchDriver): "chassis_id": mac, } - def get_ntp_servers(self) -> Dict[str, Dict]: + def get_ntp_servers(self) -> dict[str, dict]: """Return configured NTP servers. Extracts the server list from the running configuration line:: @@ -1503,7 +1413,7 @@ class TPLinkJetstreamDriver(SwitchDriver): system-time ntp [ ...] """ running = self._send_command("show running-config") - servers: Dict[str, Dict] = {} + servers: dict[str, dict] = {} for line in running.splitlines(): m = re.match(r"^system-time\s+ntp\s+\S+\s+(.*)", line.strip(), re.I) if m: @@ -1514,11 +1424,11 @@ class TPLinkJetstreamDriver(SwitchDriver): servers[srv] = {} return servers - def get_ntp_peers(self) -> Dict[str, Dict]: + def get_ntp_peers(self) -> dict[str, dict]: """Return NTP peers (same as servers on TP-Link Jetstream).""" return self.get_ntp_servers() - def get_ntp_stats(self) -> List[Dict]: + def get_ntp_stats(self) -> list[dict]: """Return NTP statistics. TP-Link CLI does not expose per-peer NTP sync statistics; returns @@ -1526,7 +1436,7 @@ class TPLinkJetstreamDriver(SwitchDriver): """ return [] - def get_optics(self) -> Dict: + def get_optics(self) -> dict: """Return optical transceiver data. TP-Link Jetstream CLI does not provide optical transceiver diagnostics @@ -1535,7 +1445,7 @@ class TPLinkJetstreamDriver(SwitchDriver): """ return {} - def get_ipv6_neighbors_table(self) -> List[Dict]: + def get_ipv6_neighbors_table(self) -> list[dict]: """Return the IPv6 neighbor table. TP-Link Jetstream CLI does not expose the IPv6 neighbor (ND) table. @@ -1549,7 +1459,7 @@ class TPLinkJetstreamDriver(SwitchDriver): destination: str = "", protocol: str = "", longer: bool = False, - ) -> Dict[str, List[Dict]]: + ) -> dict[str, list[dict]]: """Return routing table entries. Parses ``show ip route`` output:: @@ -1561,7 +1471,7 @@ class TPLinkJetstreamDriver(SwitchDriver): C 172.22.8.0/24 is directly connected, VLAN8 """ output = self._send_command("show ip route") - routes: Dict[str, List[Dict]] = {} + routes: dict[str, list[dict]] = {} proto_map = {"c": "connected", "s": "static", "r": "rip", "o": "ospf"} @@ -1639,9 +1549,23 @@ class TPLinkJetstreamDriver(SwitchDriver): """Configure a switch interface (mode, VLAN membership, admin state).""" mode = config.get("mode") iface = self._to_cli_interface(interface) + + # Read current VLAN state before entering config mode (show commands + # are not available in config mode on TP-Link Jetstream). + old_untagged_vlans: list[str] = [] + if mode == "access" and "access_vlan" in config: + try: + vlans = self.get_vlans_detail() + new_vid = str(config["access_vlan"]) + for vid_str, vlan_info in vlans.items(): + if interface in vlan_info.get("untagged", []) and vid_str != new_vid: + old_untagged_vlans.append(vid_str) + except Exception: + pass + self._enter_config_mode() try: - lines: List[str] = [f"interface {iface}"] + lines: list[str] = [f"interface {iface}"] if "enabled" in config: lines.append(" no shutdown" if config["enabled"] else " shutdown") if "description" in config: @@ -1653,10 +1577,14 @@ class TPLinkJetstreamDriver(SwitchDriver): if mode == "trunk": for vid in config.get("trunk_vlans", []): lines.append(f" switchport general allowed vlan {vid} tagged") + if "native_vlan" in config: + lines.append(f" switchport pvid {config['native_vlan']}") elif mode == "access": 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 pvid {config['access_vlan']}") + lines.append(f" switchport pvid {config['access_vlan']}") lines.append("exit") errors = self._apply_config_lines("\n".join(lines)) if errors: @@ -1671,7 +1599,7 @@ class TPLinkJetstreamDriver(SwitchDriver): # LAG / trunk membership # ------------------------------------------------------------------ - def set_lag_members(self, lag_name: str, members: List[str]) -> None: + def set_lag_members(self, lag_name: str, members: list[str]) -> None: """Set the full member-port list of a LAG (e.g. ``"Lag1"``). Diffs ``members`` against the LAG's current members (as reported by @@ -1697,7 +1625,7 @@ class TPLinkJetstreamDriver(SwitchDriver): mode = current.get("lag_mode") or "lacp" channel_mode = "active" if mode == "lacp" else "on" - lines: List[str] = [] + lines: list[str] = [] for port in to_remove: cli_iface = self._to_cli_interface(port) lines.append(f"interface {cli_iface}") @@ -1722,7 +1650,7 @@ class TPLinkJetstreamDriver(SwitchDriver): # PoE # ------------------------------------------------------------------ - def get_poe_status(self) -> Dict[str, Dict]: + def get_poe_status(self) -> dict[str, dict]: """Return PoE configuration per port from ``show power inline``. Expected ``show power inline`` columns (JetStream T-series):: @@ -1732,7 +1660,7 @@ class TPLinkJetstreamDriver(SwitchDriver): """ output = self._send_command("show power inline") - result: Dict[str, Dict] = {} + result: dict[str, dict] = {} in_table = False for line in output.splitlines(): line_s = line.strip() @@ -1768,7 +1696,7 @@ class TPLinkJetstreamDriver(SwitchDriver): return result - def set_poe(self, interface: str, config: Dict) -> None: + def set_poe(self, interface: str, config: dict) -> None: """Update PoE configuration for a single port. Uses the ``power inline`` interface sub-commands (JetStream CLI). @@ -1806,7 +1734,7 @@ class TPLinkJetstreamDriver(SwitchDriver): count: int = 5, vrf: str = "", source_interface: str = "", - ) -> Dict: + ) -> dict: """Ping *destination* from the device. Uses ``ping -n ``. TP-Link does not support source IP, @@ -1866,7 +1794,7 @@ class TPLinkJetstreamDriver(SwitchDriver): ttl: int = 255, timeout: int = 2, vrf: str = "", - ) -> Dict: + ) -> dict: """Traceroute to *destination* from the device. Uses ``tracert ``. TP-Link caps the hop count at 4 and does @@ -1879,7 +1807,7 @@ class TPLinkJetstreamDriver(SwitchDriver): if "Error" in output or "Invalid" in output or "Bad command" in output: return {"error": output.strip()} - hops: Dict[str, Dict] = {} + hops: dict[str, dict] = {} for line in output.splitlines(): # "1 20 ms 1 ms 1 ms 172.22.8.1" @@ -1917,15 +1845,15 @@ class TPLinkJetstreamDriver(SwitchDriver): def cli( self, - commands: List[str], + commands: list[str], encoding: str = "text", - ) -> Dict[str, Union[str, Dict[str, Any]]]: + ) -> 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." ) - result: Dict[str, Union[str, Dict[str, Any]]] = {} + result: dict[str, str | dict[str, Any]] = {} for cmd in commands: result[cmd] = self._send_command(cmd) return result @@ -1969,13 +1897,13 @@ class TPLinkJetstreamDriver(SwitchDriver): """Return device warnings. Currently delegates to SNMP detection only.""" return [] - def run_device_action(self, action: str) -> Dict: + def run_device_action(self, action: str) -> dict: """Execute a named action on the switch.""" if action == "fix_snmp": return self._action_fix_snmp() raise NotImplementedError(f"Unknown action: {action!r}") - def _detect_management_vlan(self) -> Optional[int]: + def _detect_management_vlan(self) -> int | None: """Detect which VLAN interface carries the management IP (self.hostname).""" try: out = self._send_command("show ip interface") @@ -1994,7 +1922,7 @@ class TPLinkJetstreamDriver(SwitchDriver): except Exception: return None - def _action_fix_snmp(self) -> Dict: + def _action_fix_snmp(self) -> dict: """Enable SNMP and configure community 'public' on the switch. TP-Link JetStream uses CLI commands — SNMP is built-in, no install needed. @@ -2100,5 +2028,3 @@ class TPLinkJetstreamDriver(SwitchDriver): build_if_metrics(metrics, descr, speed, in_oct, out_oct, in_err, out_err) return metrics - - return {"success": success, "output": "\n".join(lines)}