feat!: configure_snmp_agent with the caller's values replaces fix_snmp (0.3.0)

fix_snmp hard-coded community "public" and found the management VLAN
for the source interface with a substring match, so 10.0.0.1 matched
10.0.0.10 and a DNS name never matched (#3).

The driver implements napalm-device-types 4.13's configure_snmp_agent
steps: the caller's community (read-only, viewDefault), the source
interface set to the VLAN whose address is exactly the listen address
or the management address, configuration saved, and the community
checked in show snmp community. Contact and location are reported as
not set; there is no verified CLI for them on this switch. A value
that would break the CLI is refused.

The requirements now name napalm-device-types 4.13 in both files.

BREAKING CHANGE: run_device_action is removed.

Part of #3 and #4.

napalm_tplink_jetstream/tplink_jetstream.py

@@ -27,7 +27,7 @@ import netaddr
 from netmiko import ConnectHandler
 from netmiko.exceptions import NetmikoTimeoutException, NetmikoAuthenticationException
 
-from napalm_device_types import ConfigLifecycleMixin, FingerprintRule, SwitchDriver
+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 (
@@ -41,7 +41,7 @@ from napalm.base.netmiko_helpers import netmiko_args
 import napalm.base.constants as C
 
 
-class TPLinkJetstreamDriver(ConfigLifecycleMixin, SwitchDriver):
+class TPLinkJetstreamDriver(ConfigLifecycleMixin, SnmpAgentMixin, SwitchDriver):
     """NAPALM driver for TP-Link Jetstream managed switches."""
 
     VENDOR = "TP-Link"
@@ -1935,81 +1935,65 @@ class TPLinkJetstreamDriver(ConfigLifecycleMixin, SwitchDriver):
         """Return device warnings. Currently delegates to SNMP detection only."""
         return []
 
-    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}")
+    # ------------------------------------------------------------------
+    # SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values
+    # ------------------------------------------------------------------
 
-    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")
-            import re as _re
-            # Match "VLAN<N> is up" followed by IP line containing self.hostname
-            vlan = None
-            current_vlan = None
-            for line in out.splitlines():
-                m = _re.match(r"VLAN(\d+)\s+is", line, _re.IGNORECASE)
-                if m:
-                    current_vlan = int(m.group(1))
-                if current_vlan and self.hostname in line and "IP address" in line:
-                    vlan = current_vlan
-                    break
-            return vlan
-        except Exception:
-            return None
+    def _management_vlan(self, address: str) -> int | None:
+        """The VLAN interface whose IP address is exactly *address*."""
+        current = None
+        pattern = re.compile(rf"IP address[:\s]+{re.escape(address)}(?![\d.])", re.I)
+        for line in self._send_command("show ip interface").splitlines():
+            m = re.match(r"VLAN(\d+)\s+is", line, re.I)
+            if m:
+                current = int(m.group(1))
+            elif current is not None and pattern.search(line):
+                return current
+        return None
 
-    def _action_fix_snmp(self) -> dict:
-        """Enable SNMP and configure community 'public' on the switch.
+    def _install_snmp_agent(self) -> dict:
+        return {"ok": True, "output": "the SNMP agent is built in"}
 
-        TP-Link JetStream uses CLI commands — SNMP is built-in, no install needed.
-        Also sets snmp-server source-interface to the management VLAN so that
-        SNMP responses are sent from the correct interface.
-        """
-        lines: list = []
-
-        # Detect management VLAN (the VLAN interface carrying this device's IP)
-        mgmt_vlan = self._detect_management_vlan()
-        if mgmt_vlan:
-            lines.append(f"[detect] Management VLAN: {mgmt_vlan}")
-        else:
-            lines.append("[detect] Could not auto-detect management VLAN.")
-
-        # Configure SNMP in config mode
+    def _write_snmp_agent_config(self, settings: dict) -> dict:
+        community = settings["community"]
+        if not re.match(r'^[^\s"#]+$', community):
+            raise ValueError("the community must be one word without quotes or #")
+        vlan = self._management_vlan(settings.get("listen") or self.hostname)
         conf_prompt = self._conf_prompt()
         self._enter_config_mode()
-        self.device.send_command("snmp-server", expect_string=conf_prompt, read_timeout=self.timeout)
-        lines.append("[config] SNMP enabled.")
+        try:
+            for command in (
+                "snmp-server",
+                f'snmp-server community "{community}" read-only "viewDefault"',
+                *([f"snmp-server source-interface vlan {vlan}"] if vlan else []),
+            ):
+                self.device.send_command(
+                    command, expect_string=conf_prompt, read_timeout=self.timeout
+                )
+        finally:
+            self._exit_config_mode()
+        notes = [f"source interface vlan {vlan}" if vlan else "management VLAN not found"]
+        if settings.get("contact") or settings.get("location"):
+            notes.append("contact and location not set (no verified CLI on this switch)")
+        return {"ok": True, "output": "community set (read-only); " + "; ".join(notes)}
 
-        # Community: TP-Link syntax is snmp-server community "<name>" read-only "<view>"
-        self.device.send_command(
-            'snmp-server community "public" read-only "viewDefault"',
-            expect_string=conf_prompt,
-            read_timeout=self.timeout,
-        )
-        lines.append("[config] Community 'public' (read-only, viewDefault) configured.")
+    @staticmethod
+    def _allow_snmp_sources(settings: dict) -> dict:
+        sources = ", ".join(settings.get("allowed_sources", [])) or "none given"
+        note = f"no host firewall on the switch; sources not enforced ({sources})"
+        return {"ok": True, "output": note}
 
-        # Set source-interface so responses come from the management VLAN
-        if mgmt_vlan:
-            self.device.send_command(
-                f"snmp-server source-interface vlan {mgmt_vlan}",
-                expect_string=conf_prompt,
-                read_timeout=self.timeout,
-            )
-            lines.append(f"[config] Source-interface set to vlan {mgmt_vlan}.")
-
-        # Exit config mode and save
-        self._exit_config_mode()
+    def _start_snmp_agent(self) -> dict:
         self._save_config()
-        lines.append("[config] Configuration saved.")
+        return {"ok": True, "output": "configuration saved"}
 
-        # Verify
-        out = self._send_command("show snmp community")
-        success = "public" in out and "disabled" not in out.lower()
-        if success:
-            lines.append("[ok] SNMP is active with community 'public'.")
-        else:
-            lines.append(f"[warn] Verification failed: {out[:100]}")
-
-        return {"success": success, "output": "\n".join(lines)}
+    def configure_snmp_agent(self, settings: dict) -> dict:
+        """The template's steps, then a check that the switch lists the community."""
+        result = SnmpAgentMixin.configure_snmp_agent(self, settings)
+        if not result["success"]:
+            return result
+        listed = self._send_command("show snmp community")
+        if re.search(rf"(?<!\S){re.escape(settings['community'])}(?!\S)", listed):
+            return result
+        missing = f"[verify] the switch does not list the community: {listed[:200]}"
+        return {**result, "success": False, "output": f"{result['output']}\n{missing}"}

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "napalm-tplink-jetstream"
-version = "0.2.0"
+version = "0.3.0"
 description = "NAPALM driver for TP-Link Jetstream managed switches"
 readme = "README.md"
 license = { text = "Apache-2.0" }
@@ -24,7 +24,7 @@ classifiers = [
 ]
 dependencies = [
     "napalm>=4.0.0",
-    "napalm_device_types>=0.2.0",
+    "napalm_device_types>=4.13.0",
     "netmiko>=4.0.0",
     "paramiko>=5.0.0",  # CVE-2026-44405
     "netaddr",

requirements.txt

@@ -1,4 +1,4 @@
 napalm>=4.0.0
-napalm_device_types>=0.1.0
+napalm_device_types>=4.13.0
 netmiko>=4.0.0
 netaddr

tests/unit/test_snmp_agent.py

@@ -0,0 +1,84 @@
+"""The SNMP agent through napalm-device-types' configure_snmp_agent (netOrk #1105).
+
+fix_snmp hard-coded community "public" and found the management VLAN by a
+substring match (10.0.0.1 matched 10.0.0.10, #3). The community is the caller's
+now, and the source interface is the VLAN that carries exactly the management
+address.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from napalm_tplink_jetstream.tplink_jetstream import TPLinkJetstreamDriver
+
+SETTINGS = {"community": "c0mm", "allowed_sources": ["10.7.224.20"], "contact": "ops"}
+
+SHOW_IP_INTERFACE = """\
+VLAN1 is up, line protocol is up
+  Primary IP address: 10.0.0.10/24
+VLAN8 is up, line protocol is up
+  Primary IP address: 10.0.0.1/24
+"""
+
+
+@pytest.fixture
+def driver():
+    d = TPLinkJetstreamDriver("10.0.0.1", "admin", "pw")
+    d.device = MagicMock()
+    d.sent: list[str] = []
+    d.device.send_command.side_effect = lambda cmd, **kw: d.sent.append(cmd) or ""
+    d._conf_prompt = lambda: r"\(config\)#"
+    d._enter_config_mode = MagicMock()
+    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"
+    )
+    return d
+
+
+def test_the_callers_community(driver):
+    result = driver.configure_snmp_agent(SETTINGS)
+
+    assert result["success"] is True
+    assert 'snmp-server community "c0mm" read-only "viewDefault"' in driver.sent
+    assert not any("public" in c for c in driver.sent)
+    driver._save_config.assert_called_once()
+
+
+def test_the_source_interface_is_the_vlan_of_exactly_the_management_address(driver):
+    driver.configure_snmp_agent(SETTINGS)
+
+    assert "snmp-server source-interface vlan 8" in driver.sent
+
+
+def test_contact_and_location_are_reported_as_not_set(driver):
+    steps = {s["step"]: s for s in driver.configure_snmp_agent(SETTINGS)["steps"]}
+
+    assert "contact" in steps["configure"]["output"]
+
+
+def test_a_community_the_switch_does_not_list_fails(driver):
+    driver._send_command.side_effect = lambda cmd: (
+        SHOW_IP_INTERFACE if cmd == "show ip interface" else "1  public  read-only\n"
+    )
+
+    assert driver.configure_snmp_agent(SETTINGS)["success"] is False
+
+
+@pytest.mark.parametrize("community", ["two words", 'q"uote', ""])
+def test_a_community_that_would_break_the_cli_is_refused(driver, community):
+    result = driver.configure_snmp_agent({"community": community, "allowed_sources": []})
+
+    assert result["success"] is False
+    assert driver.sent == []
+
+
+def test_the_old_action_is_gone():
+    for name in ("run_device_action", "_action_fix_snmp", "_detect_management_vlan"):
+        assert not hasattr(TPLinkJetstreamDriver, name), name