Merge pull request 'feat!: configure_snmp_agent with the caller's values replaces fix_snmp (0.3.0)' (#6) from feat/snmp-agent-template into main

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