feat: add VLAN/interface write methods and SG3210 support (v0.1.2)

- Implement `delete_vlan()`: removes VLAN from switch, automatically
  moves untagged (PVID) ports to VLAN 1 before deletion to satisfy
  TP-Link's constraint
- Implement `set_vlan()`: creates or updates a VLAN with optional name
- Implement `set_interface()`: configures trunk/access mode and VLAN
  membership on a switch interface
- Add `_to_cli_interface()` helper to expand abbreviated port names
  (e.g. Gi1/0/1 → gigabitEthernet 1/0/1)
- Import `InterfaceConfigDict` and `VlanConfigDict` from
  `napalm_device_types.models`
- Bump `napalm_device_types` dependency to >=0.2.0
- Add TL-SG3210 to tested-device table in README
- Bump version to 0.1.2
This commit is contained in:
Christian Manivong
2026-05-29 08:53:13 +02:00
parent ec70aad7ca
commit 1874622543
3 changed files with 127 additions and 3 deletions
+1
View File
@@ -7,6 +7,7 @@ NAPALM community driver for **TP-Link Jetstream** managed switches (T1500G, T160
| Model | Series | Tested |
|---|---|---|
| TL-SG2210P | Jetstream Smart | ✅ |
| TL-SG3210 | Jetstream Smart | ✅ |
> More devices from the T1500G, T1600G, T2600G and T3700G series should work as well — contributions welcome.
+124 -1
View File
@@ -13,7 +13,7 @@
"""NAPALM driver for TP-Link Jetstream managed switches.
Tested against: SG2210P, T1500G, T1600G, T2600G, T3700G series.
Tested against: SG2210P, SG3210, T1500G, T1600G, T2600G, T3700G series.
Netmiko device_type: ``tplink_jetstream``
"""
@@ -26,6 +26,7 @@ from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
from napalm_device_types import SwitchDriver
from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict
from napalm.base import helpers as napalm_helpers
from napalm.base.exceptions import (
ConnectionException,
@@ -914,6 +915,82 @@ class TPLinkJetstreamDriver(SwitchDriver):
return vlans
# ------------------------------------------------------------------
# VLAN write operations
# ------------------------------------------------------------------
@staticmethod
def _to_cli_interface(port: str) -> str:
"""Convert abbreviated port name (e.g. ``Gi1/0/1``) to CLI form (``gigabitEthernet 1/0/1``)."""
_PREFIX_MAP = {
"gi": "gigabitEthernet",
"te": "ten-gigabitEthernet",
"fa": "fastEthernet",
"lag": "port-channel",
}
m = re.match(r"^([A-Za-z]+)(\d.*)", port)
if not m:
return port
full = _PREFIX_MAP.get(m.group(1).lower(), m.group(1))
return f"{full} {m.group(2)}"
def delete_vlan(self, vlan_id: int) -> None:
"""Remove a VLAN from the switch.
Before deleting, any port whose PVID is this VLAN (untagged member
with no other native VLAN) is moved to VLAN 1 via ``switchport pvid 1``.
TP-Link refuses to execute ``no vlan <id>`` while the VLAN is still
the PVID of at least one port.
:param vlan_id: VLAN ID (1–4094) to delete.
:raises ValueError: If *vlan_id* is out of the valid range.
:raises CommandErrorException: If the device reports an error.
"""
if not 1 <= vlan_id <= 4094:
raise ValueError(f"VLAN ID {vlan_id} is out of range (1–4094)")
# Move untagged (PVID) ports off this VLAN before deleting it.
vlan_detail = self.get_vlans_detail()
untagged_ports = vlan_detail.get(str(vlan_id), {}).get("untagged", [])
if untagged_ports:
self._enter_config_mode()
try:
ep_any = self._any_prompt()
for port in untagged_ports:
cli_iface = self._to_cli_interface(port)
self.device.send_command(
f"interface {cli_iface}",
expect_string=ep_any,
read_timeout=self.timeout,
)
self.device.send_command(
"switchport pvid 1",
expect_string=ep_any,
read_timeout=self.timeout,
)
self.device.send_command(
"exit",
expect_string=ep_any,
read_timeout=self.timeout,
)
finally:
self._exit_config_mode()
self._enter_config_mode()
try:
out = self.device.send_command(
f"no vlan {vlan_id}",
expect_string=self._any_prompt(),
read_timeout=self.timeout,
).strip()
if out and any(kw in out for kw in ("Error", "% Invalid", "% Unknown", "not exist")):
raise CommandErrorException(f"delete_vlan({vlan_id}): {out}")
finally:
self._exit_config_mode()
self._save_config()
# ------------------------------------------------------------------
# NAPALM configuration management
# ------------------------------------------------------------------
@@ -1484,6 +1561,52 @@ class TPLinkJetstreamDriver(SwitchDriver):
return routes
# ------------------------------------------------------------------
# SwitchDriver: write methods
# ------------------------------------------------------------------
def set_vlan(self, vlan_id: int, config: VlanConfigDict) -> None:
"""Create or update a VLAN on the switch."""
self._enter_config_mode()
try:
lines = [f"vlan {vlan_id}"]
if "name" in config:
# Quote the name to handle spaces
lines.append(f' name "{config["name"]}"')
lines.append("exit")
errors = self._apply_config_lines("\n".join(lines))
if errors:
raise CommandErrorException(
f"set_vlan({vlan_id}) errors: {errors}"
)
finally:
self._exit_config_mode()
self._save_config()
def set_interface(self, interface: str, config: InterfaceConfigDict) -> None:
"""Configure a switch interface (mode, VLAN membership)."""
mode = config.get("mode")
iface = self._to_cli_interface(interface)
self._enter_config_mode()
try:
lines: List[str] = [f"interface {iface}"]
if mode == "trunk":
for vid in config.get("trunk_vlans", []):
lines.append(f" switchport general allowed vlan {vid} tagged")
elif mode == "access":
if "access_vlan" in config:
lines.append(f" switchport general allowed vlan {config['access_vlan']} untagged")
lines.append(f" switchport general 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}"
)
finally:
self._exit_config_mode()
self._save_config()
def ping(
self,
destination: str,
+2 -2
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "napalm-tplink-jetstream"
version = "0.1.1"
version = "0.1.2"
description = "NAPALM driver for TP-Link Jetstream managed switches"
readme = "README.md"
license = { text = "Apache-2.0" }
@@ -26,7 +26,7 @@ classifiers = [
]
dependencies = [
"napalm>=4.0.0",
"napalm_device_types>=0.1.0",
"napalm_device_types>=0.2.0",
"netmiko>=4.0.0",
"netaddr",
]