feat!: fix templates with the caller's values replace run_device_action (0.2.0)

fix_snmp set no community at all and opened the management zone's
udp/161 to every source, with a rule named after an application; a
second fix_snmp in the system mixin was shadowed and never ran (#5).
install_lldpd, install_auc and install_coreutils_base64 were actions of
their own.

The driver implements napalm-device-types 4.13's templates instead:
- configure_snmp_agent(settings): snmpd installed when missing;
  community, contact and location in UCI with one com2sec (com2sec6)
  per allowed source; the zone-resolving firewall step of the old fix,
  now limited to the sources (src_ip) and named Allow-SNMP; snmpd
  enabled, restarted and listening;
- get_management_peer() from SSH_CONNECTION;
- ensure_package(name) from the base package, over install_package.
A community, contact or location that would break UCI is refused.

BREAKING CHANGE: run_device_action is removed. Requires
napalm-device-types 4.13.

Closes #5. Part of #13.

tests/unit/test_driver.py

@@ -1488,6 +1488,12 @@ class _FakeShell:
 
     def __call__(self, cmd, **kw):
         self.issued.append(cmd)
+        if "/usr/sbin/snmpd" in cmd:
+            return "present"
+        if "__OWRT_RC" in cmd:
+            return "ok\n__OWRT_RC=0"
+        if cmd.startswith("uci show snmpd"):
+            return ""
         if cmd.startswith("uci show firewall"):
             return self.firewall
         if cmd.startswith("uci show network"):
@@ -1507,8 +1513,11 @@ class _FakeShell:
         return ""
 
 
+_SETTINGS = {"community": "c0mm", "allowed_sources": []}
+
+
 class TestFixSnmpZoneDetection:
-    """_action_fix_snmp() must resolve the real zone that owns the mgmt address."""
+    """The firewall step must resolve the real zone that owns the mgmt address."""
 
     def test_static_mgmt_address_selects_owning_zone(self, driver):
         """10.10.0.5 lives on network 'mgmt' → zone 'mgmt', not the 'lan' fallback."""
@@ -1518,7 +1527,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="10.10.0.1 51234 10.10.0.5 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "Management zone: 'mgmt'" in result["output"]
         assert any("firewall.allow_snmp_from_mgmt.src='mgmt'" in c for c in shell.issued)
 
@@ -1530,7 +1539,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "Management zone: 'lan'" in result["output"]
 
     def test_dhcp_mgmt_address_resolved_via_l3_device(self, driver):
@@ -1542,7 +1551,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="10.10.0.1 51234 10.10.0.5 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "Management zone: 'lan'" in result["output"]
 
     def test_nameless_zone_is_reported_and_action_fails(self, driver):
@@ -1553,7 +1562,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
         assert "@zone[0]" in result["output"]
         assert "name" in result["output"]
@@ -1566,7 +1575,7 @@ class TestFixSnmpZoneDetection:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         assert not any("allow_snmp" in c for c in shell.issued)
 
 
@@ -1586,7 +1595,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         assert any("firewall.allow_snmp_from_lan.src='lan'" in c for c in shell.issued)
 
     def test_stale_rules_are_deleted_highest_index_first(self, driver):
@@ -1605,7 +1614,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         deletes = [c for c in shell.issued if "delete" in c and "@rule" in c]
         joined = " ".join(deletes)
         assert joined.index("@rule[2]") < joined.index("@rule[1]")
@@ -1623,7 +1632,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         delete_idx = next(i for i, c in enumerate(shell.issued) if "snmp_netork" in c and "delete" in c)
         assert any("uci commit firewall" in c for c in shell.issued[delete_idx:])
 
@@ -1641,7 +1650,7 @@ class TestFixSnmpRuleWriting:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        driver._action_fix_snmp()
+        driver.configure_snmp_agent(_SETTINGS)
         assert any("snmp_netork" in c and "delete" in c for c in shell.issued)
 
 
@@ -1663,7 +1672,7 @@ class TestFixSnmpReloadVerification:
             reload_out=self.FW4_ZONE_ERROR,
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
 
     def test_reload_output_is_not_truncated(self, driver):
@@ -1675,7 +1684,7 @@ class TestFixSnmpReloadVerification:
             reload_out=self.FW4_ZONE_ERROR,
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert "references unknown zone" in result["output"]
 
     def test_missing_live_rule_fails_the_action(self, driver):
@@ -1687,7 +1696,7 @@ class TestFixSnmpReloadVerification:
             nft_hits="0",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
 
     def test_snmpd_not_listening_fails_the_action(self, driver):
@@ -1698,7 +1707,7 @@ class TestFixSnmpReloadVerification:
             listen_hits="0",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is False
 
     def test_fully_healthy_run_succeeds(self, driver):
@@ -1708,7 +1717,7 @@ class TestFixSnmpReloadVerification:
             ssh_connection="192.168.1.50 5000 192.168.1.1 22",
         )
         driver._send_command = shell
-        result = driver._action_fix_snmp()
+        result = driver.configure_snmp_agent(_SETTINGS)
         assert result["success"] is True
 
 

tests/unit/test_snmp_agent.py

@@ -0,0 +1,96 @@
+"""configure_snmp_agent's values: community, sources, contact, location (netOrk #1105).
+
+The old fix_snmp set no community at all and opened the management zone to every
+source; the rule was named after an application. Every value is the caller's now.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from napalm_openwrt import OpenWrtDriver
+from tests.unit.test_driver import UCI_FIREWALL_HEALTHY, UCI_NETWORK_STATIC, _FakeShell
+
+SETTINGS = {
+    "community": "c0mm",
+    "allowed_sources": ["10.7.224.20", "fd00::20"],
+    "contact": "ops",
+    "location": "rack 1",
+}
+
+
+@pytest.fixture
+def shell():
+    return _FakeShell(
+        firewall=UCI_FIREWALL_HEALTHY,
+        network=UCI_NETWORK_STATIC,
+        ssh_connection="10.7.224.20 51234 192.168.1.1 22",
+    )
+
+
+@pytest.fixture
+def owrt(shell):
+    d = OpenWrtDriver("ap", "root", "pw")
+    d._send_command = shell
+    return d
+
+
+def _joined(shell):
+    return " ".join(shell.issued)
+
+
+def test_the_steps_run_in_order_and_succeed(owrt):
+    result = owrt.configure_snmp_agent(SETTINGS)
+
+    assert result["success"] is True
+    assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
+
+
+def test_community_per_source_in_uci(owrt, shell):
+    owrt.configure_snmp_agent(SETTINGS)
+
+    uci = _joined(shell)
+    assert "uci set snmpd.public.source='10.7.224.20/32'" in uci
+    assert "uci set snmpd.public.community='c0mm'" in uci
+    assert "uci set snmpd.public6.source='fd00::20/128'" in uci
+    assert "snmpd.@system[0].sysLocation='rack 1'" in uci
+    assert "uci commit snmpd" in uci
+
+
+def test_without_sources_every_address_may_ask(owrt, shell):
+    owrt.configure_snmp_agent({"community": "c0mm", "allowed_sources": []})
+
+    assert "uci set snmpd.public.source='default'" in _joined(shell)
+
+
+def test_ipv4_sources_only_keep_ipv6_to_localhost(owrt, shell):
+    owrt.configure_snmp_agent({"community": "c0mm", "allowed_sources": ["10.7.224.20"]})
+
+    assert "uci set snmpd.public6.source='::1/128'" in _joined(shell)
+
+
+def test_the_firewall_rule_is_limited_to_the_sources_and_neutrally_named(owrt, shell):
+    owrt.configure_snmp_agent(SETTINGS)
+
+    uci = _joined(shell)
+    assert "uci add_list firewall.allow_snmp_from_lan.src_ip='10.7.224.20/32'" in uci
+    assert "firewall.allow_snmp_from_lan.name='Allow-SNMP'" in uci
+    assert "netork" not in uci.lower()
+
+
+def test_the_management_peer_is_the_ssh_client(owrt):
+    assert owrt.get_management_peer() == "10.7.224.20"
+
+
+@pytest.mark.parametrize("community", ["two words", "it's", ""])
+def test_a_community_that_would_break_uci_is_refused(owrt, shell, community):
+    result = owrt.configure_snmp_agent({"community": community, "allowed_sources": []})
+
+    assert result["success"] is False
+    assert "uci commit snmpd" not in _joined(shell)
+
+
+def test_the_old_actions_are_gone():
+    gone = ("run_device_action", "_action_fix_snmp", "_action_install_lldpd", "_action_install_auc")
+    for name in gone:
+        assert not hasattr(OpenWrtDriver, name), name