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

fix_snmp hard-coded community "public". The driver implements
napalm-device-types 4.13's configure_snmp_agent as a whole, because
over REST a temporary SSH session has to run the config-mode CLI:
community (read-only), contact and location from the caller, saved
and verified in show snmp-server. The agent is built in and the switch
filters nothing for its management plane; the steps say so instead of
pretending. A value that would break the CLI is refused.

The unused _detect_management_vlan goes with it.

BREAKING CHANGE: run_device_action is removed.

Part of #12 and #6 (the dead helper).

napalm_procurve/procurve.py

@@ -32,6 +32,7 @@ from napalm_device_types import (
     FingerprintRule,
     SwitchDriver,
     add_lag_interfaces,
+    run_fix_steps,
 )
 from napalm_device_types.models import InterfaceConfigDict, VlanConfigDict
 from napalm.base import helpers as napalm_helpers
@@ -76,6 +77,10 @@ _SSH_DISABLED_LEGACY = {
 }
 
 
+#: What may stand in a quoted CLI value: no whitespace, quote or #.
+_CLI_WORD = re.compile(r'^[^\s"#]+$')
+
+
 class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
     """NAPALM driver for HPE/Aruba ProCurve switches.
 
@@ -1220,93 +1225,79 @@ class ProcurveDriver(ConfigLifecycleMixin, SwitchDriver):
 
         return SNMPConfigDict(running=True, community=community, port=161, version="2c")
 
-    def run_device_action(self, action: str) -> Dict:
+    # ------------------------------------------------------------------
-        """Execute a named action on the switch."""
+    # SNMP agent (napalm-device-types SnmpAgentMixin): the caller's values
-        if action == "fix_snmp":
+    # ------------------------------------------------------------------
-            return self._action_fix_snmp()
-        raise NotImplementedError(f"Unknown action: {action!r}")
 
-    def _detect_management_vlan(self) -> Optional[int]:
+    def configure_snmp_agent(self, settings: Dict[str, Any]) -> Dict[str, Any]:
-        """Find the VLAN that carries this switch's management IP."""
+        """The agent is built in; configure community, contact and location, then save.
-        try:
-            out = self._send_command("show ip")
-            for line in out.splitlines():
-                # "  VLAN8     | 172.22.8.x  ..."  or similar
-                m = re.search(r"VLAN\s*(\d+)[^|]*\|\s*" + re.escape(self.hostname), line, re.I)
-                if m:
-                    return int(m.group(1))
-                # Alternative: "VLAN  8   172.22.8.x"
-                m2 = re.search(r"\bVLAN\s+(\d+)\b.*\b" + re.escape(self.hostname) + r"\b", line, re.I)
-                if m2:
-                    return int(m2.group(1))
-        except Exception:
-            pass
-        return None
 
-    def _action_fix_snmp(self) -> Dict:
+        The REST API has no writable SNMP endpoint, so over REST a temporary SSH
-        """Enable SNMP with community 'public' (read-only) on the ProCurve switch.
+        session runs the config-mode CLI.
-
-        ProCurve/Aruba syntax:
-          snmp-server community "public" manager restricted
-
-        When connected via REST API, opens a temporary SSH session for the
-        config-mode CLI commands (REST API has no writable SNMP endpoint).
         """
-        if self._transport == "api":
+        steps = [
-            # REST API cannot run config-mode CLI.  Temporarily open SSH.
+            ("install", lambda: {"ok": True, "output": "the SNMP agent is built in"}),
-            old_api, old_device, old_transport = self._api, self._device, self._transport
+            ("configure", lambda: self._snmp_configure_cli(settings)),
-            self._api = None
+            ("firewall", lambda: self._snmp_sources_note(settings)),
-            self._device = None
+            ("start", lambda: self._snmp_save_and_verify(settings)),
-            self._transport = None
+        ]
-
+        if self._transport != "api":
-            ssh_error: str = ""
+            return run_fix_steps(steps)
-            for legacy in (False, True):
+        old_api, old_device, old_transport = self._api, self._device, self._transport
-                try:
+        self._api = self._device = self._transport = None
-                    conn = ConnectHandler(**self._netmiko_kwargs(legacy))
+        ssh_error = ""
-                    self._device = conn
+        for legacy in (False, True):
-                    self._transport = "ssh_legacy" if legacy else "ssh"
-                    break
-                except Exception as exc:
-                    ssh_error = str(exc)
-
-            if self._device is None:
-                self._api, self._device, self._transport = old_api, old_device, old_transport
-                return {
-                    "success": False,
-                    "output": f"REST transport active; SSH also failed: {ssh_error}",
-                }
             try:
-                return self._action_fix_snmp_cli()
+                self._device = ConnectHandler(**self._netmiko_kwargs(legacy))
-            finally:
+                self._transport = "ssh_legacy" if legacy else "ssh"
+                break
+            except Exception as exc:  # noqa: BLE001 - the next form, then report
+                ssh_error = str(exc)
+        try:
+            if self._device is None:
+                failed = {"ok": False, "output": f"no SSH for the CLI: {ssh_error}"}
+                return run_fix_steps([("configure", lambda: failed)])
+            return run_fix_steps(steps)
+        finally:
+            if self._device is not None:
                 try:
                     self._device.disconnect()
-                except Exception:
+                except Exception:  # noqa: BLE001 - closing a temporary session
                     pass
-                self._api, self._device, self._transport = old_api, old_device, old_transport
+            self._api, self._device, self._transport = old_api, old_device, old_transport
 
-        return self._action_fix_snmp_cli()
+    def _snmp_configure_cli(self, settings: Dict[str, Any]) -> Dict[str, Any]:
-
+        community = settings["community"]
-    def _action_fix_snmp_cli(self) -> Dict:
+        if not _CLI_WORD.match(community):
-        """Run the SNMP fix via CLI (SSH / Telnet, self._device must be open)."""
+            raise ValueError("the community must be one word without quotes or #")
-        lines: list = []
+        lines = [f'snmp-server community "{community}" manager restricted']
+        for key in ("contact", "location"):
+            value = settings.get(key, "")
+            if value:
+                if '"' in value or "\n" in value:
+                    raise ValueError(f"{key} must be one line without double quotes")
+                lines.append(f'snmp-server {key} "{value}"')
         self._enter_config_mode()
         try:
-            errors = self._apply_config_lines('snmp-server community "public" manager restricted')
+            errors = self._apply_config_lines("\n".join(lines))
-            if errors:
-                lines.append(f"[warn] Config errors: {errors}")
-                return {"success": False, "output": "\n".join(lines)}
-            lines.append("[config] SNMP community 'public' (manager restricted / read-only) configured.")
         finally:
             self._exit_config_mode()
+        if errors:
+            return {"ok": False, "output": f"the switch refused: {errors}"}
+        return {"ok": True, "output": "community (read-only), contact and location set"}
 
+    @staticmethod
+    def _snmp_sources_note(settings: Dict[str, Any]) -> Dict[str, Any]:
+        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}
+
+    def _snmp_save_and_verify(self, settings: Dict[str, Any]) -> Dict[str, Any]:
         self._save_config()
-        lines.append("[config] Configuration saved.")
+        shown = self._send_command("show snmp-server")
-
+        if re.search(rf"^\s*{re.escape(settings['community'])}\s", shown, re.M):
-        out = self._send_command("show snmp-server")
+            return {"ok": True, "output": "configuration saved; the community is active"}
-        success = "public" in out
+        return {"ok": False, "output": f"saved, but the community is not shown: {shown[:200]}"}
-        lines.append("[ok] SNMP is active with community 'public'." if success
-                     else f"[warn] Verification — community not found: {out[:200]}")
-        return {"success": success, "output": "\n".join(lines)}
 
 
 # ---------------------------------------------------------------------------

tests/unit/test_snmp_agent.py

@@ -0,0 +1,93 @@
+"""The SNMP agent through napalm-device-types' configure_snmp_agent (netOrk #1105).
+
+fix_snmp hard-coded community "public". The switch's agent is built in and the
+switch filters nothing for its own management plane, so of the template's steps
+only "configure" (community, contact, location) and "start" (save, verify) do
+work here; the caller's values are used as given.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from napalm_procurve.procurve import ProcurveDriver
+
+SETTINGS = {
+    "community": "c0mm",
+    "allowed_sources": ["10.7.224.20"],
+    "contact": "ops",
+    "location": "rack 1",
+}
+
+
+@pytest.fixture
+def driver():
+    d = ProcurveDriver("10.0.0.40", "admin", "pw")
+    d._transport = "ssh"
+    d._device = MagicMock()
+    d.applied: list[str] = []
+    d._enter_config_mode = MagicMock()
+    d._exit_config_mode = MagicMock()
+    d._save_config = MagicMock()
+    d._apply_config_lines = lambda text: d.applied.append(text) or []
+    d._send_command = MagicMock(
+        return_value=" Community Name   MIB View Write Access\n c0mm   Manager Restricted\n"
+    )
+    return d
+
+
+def test_community_contact_and_location_are_the_callers(driver):
+    result = driver.configure_snmp_agent(SETTINGS)
+
+    assert result["success"] is True
+    config = "\n".join(driver.applied)
+    assert 'snmp-server community "c0mm" manager restricted' in config
+    assert 'snmp-server contact "ops"' in config
+    assert 'snmp-server location "rack 1"' in config
+    assert "public" not in config
+    driver._save_config.assert_called_once()
+
+
+def test_the_steps_say_what_the_switch_does_not_do(driver):
+    result = driver.configure_snmp_agent(SETTINGS)
+
+    steps = {s["step"]: s for s in result["steps"]}
+    assert list(steps) == ["install", "configure", "firewall", "start"]
+    assert "built in" in steps["install"]["output"]
+    assert "10.7.224.20" in steps["firewall"]["output"]  # not enforced, and said so
+
+
+def test_a_community_the_switch_does_not_show_afterwards_fails(driver):
+    driver._send_command.return_value = " Community Name   MIB View\n public  Manager\n"
+
+    assert driver.configure_snmp_agent(SETTINGS)["success"] is False
+
+
+def test_a_rejected_line_fails_and_nothing_is_saved(driver):
+    driver._apply_config_lines = lambda text: ["Invalid input: community"]
+
+    result = driver.configure_snmp_agent(SETTINGS)
+
+    assert result["success"] is False
+    driver._save_config.assert_not_called()
+
+
+@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.applied == []
+
+
+def test_the_old_action_and_its_dead_helper_are_gone():
+    gone = (
+        "run_device_action",
+        "_action_fix_snmp",
+        "_action_fix_snmp_cli",
+        "_detect_management_vlan",
+    )
+    for name in gone:
+        assert not hasattr(ProcurveDriver, name), name