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

fix: set_interface access VLAN uses correct TP-Link CLI commands

Browse Source 
- Replace 'switchport general pvid {vlan}' (invalid) with 'switchport pvid {vlan}'
- For access mode: read current untagged VLAN memberships before entering
  config mode, then issue 'no switchport general allowed vlan {old_vid}'
  for each to avoid leaving the port with two untagged VLANs
- Add native_vlan support for trunk mode via 'switchport pvid {native_vlan}'
- Remove invalid 'switchport mode access/general' commands (not supported
  on this switch family)

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