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

fix_snmp wrote an application's snmpd.conf (community public, Managed
by netOrk, netork@localhost) and judged itself by a local snmpget.

The drivers implement napalm-device-types 4.13's SnmpAgentMixin steps
instead: net-snmp installed, snmpd.conf with one rocommunity line per
allowed source (staged without privilege, installed by root), snmpd
enabled and restarted. pf and ipfw rules stay the operator's. A
community or text that would break the file is refused. The caller
verifies with its own SNMP client.

Also get_management_peer() from SSH_CONNECTION, ensure_package from
the base package, and get_snmp_config no longer assumes "public".

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

Part of #12 (the OpenBSD new_version text is still open).

tests/test_drivers.py

@@ -462,10 +462,11 @@ class TestServices:
 
 class TestSnmp:
     """net-snmp from packages, as decided in NetOrk/netork#800: UCD-SNMP-MIB,
-    the same health metrics as on Linux."""
+    the same health metrics as on Linux. Set up through napalm-device-types'
+    ``configure_snmp_agent`` template; every value is the caller's (#12)."""
 
     STAGED = "/tmp/napalm-bsd-stage.Ab12Cd"
-    PROBE_OK = "STRING: FreeBSD host 15.1-RELEASE-p4 FreeBSD 15.1-RELEASE-p4 GENERIC amd64"
+    SETTINGS = {"community": "c0mm", "allowed_sources": ["10.7.224.20"], "contact": "ops"}
 
     @pytest.mark.parametrize(
         ("cls", "install", "conf", "start"),
@@ -484,8 +485,10 @@ class TestSnmp:
             ),
         ],
     )
-    def test_fix_snmp_installs_configures_and_starts_net_snmp(self, cls, install, conf, start):
-        driver = _channel(cls, {base._STAGE: self.STAGED, BsdDriver.SNMP_PROBE: self.PROBE_OK})
+    def test_installs_configures_and_starts_net_snmp_with_the_callers_values(
+        self, cls, install, conf, start
+    ):
+        driver = _channel(cls, {base._STAGE: self.STAGED})
         stdin_seen = []
         run = driver.run_command
 
@@ -495,24 +498,25 @@ class TestSnmp:
             return run(command, privileged=privileged, timeout=timeout, stdin=stdin)
 
         driver.run_command = recording  # type: ignore[method-assign]
-        result = driver.run_device_action("fix_snmp")
+        result = driver.configure_snmp_agent(self.SETTINGS)
 
         assert result["success"] is True
+        assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
         commands = [c for c, _ in driver.calls]
         assert commands[0] == install
         installed = next(c for c, p in driver.calls if p and "install -m 644" in c)
         assert conf in installed and self.STAGED in installed
-        assert start in commands
-        assert commands[-1] == BsdDriver.SNMP_PROBE
-        assert all(p for c, p in driver.calls if c not in (BsdDriver.SNMP_PROBE, base._STAGE))
+        assert commands[-1] == start
         # The configuration reaches the host only through the unprivileged stage (#8).
         assert [(c, p) for c, p, _ in stdin_seen] == [(base._STAGE, False)]
         written = stdin_seen[0][2].decode()
-        assert "agentAddress udp:161" in written and "rocommunity public" in written
+        assert "rocommunity c0mm 10.7.224.20/32" in written
+        assert "sysContact    ops" in written
+        assert "public" not in written and "netork" not in written.lower()
 
     def test_a_failed_stage_runs_nothing_as_root(self):
         driver = _channel(FreeBSDDriver, {base._STAGE: ("mktemp: permission denied", 1)})
-        result = driver.run_device_action("fix_snmp")
+        result = driver.configure_snmp_agent(self.SETTINGS)
         assert result["success"] is False
         assert not any(p and "install -m 644" in c for c, p in driver.calls)
 
@@ -526,20 +530,35 @@ class TestSnmp:
                 )
             },
         )
-        result = driver.run_device_action("fix_snmp")
+        result = driver.configure_snmp_agent(self.SETTINGS)
         assert result["success"] is False
         assert "No packages available" in result["output"]
         assert len(driver.calls) == 1
 
-    def test_no_answer_from_the_agent_is_a_failure(self):
-        driver = _channel(
-            OpenBSDDriver, {BsdDriver.SNMP_PROBE: "Timeout: No Response from 127.0.0.1"}
+    def test_the_host_firewall_is_left_to_the_operator(self):
+        result = _channel(FreeBSDDriver, {base._STAGE: self.STAGED}).configure_snmp_agent(
+            self.SETTINGS
         )
-        assert driver.run_device_action("fix_snmp")["success"] is False
+        firewall = next(s for s in result["steps"] if s["step"] == "firewall")
+        assert firewall["ok"] is True and "pf" in firewall["output"]
 
-    def test_unknown_action(self):
-        with pytest.raises(NotImplementedError):
-            _channel(FreeBSDDriver, {}).run_device_action("fix_apt_proxy")
+    @pytest.mark.parametrize("community", ["two words", 'q"uote', ""])
+    def test_a_community_that_would_break_the_file_is_refused(self, community):
+        driver = _channel(FreeBSDDriver, {base._STAGE: self.STAGED})
+        result = driver.configure_snmp_agent({"community": community, "allowed_sources": []})
+        assert result["success"] is False
+        assert not any(p and "install -m 644" in c for c, p in driver.calls)
+
+    def test_run_device_action_is_gone(self):
+        assert not hasattr(BsdDriver, "run_device_action")
+        assert not hasattr(BsdDriver, "_action_fix_snmp")
+
+    def test_the_management_peer_is_the_ssh_client(self):
+        driver = _channel(FreeBSDDriver, {base._PEER_COMMAND: "10.7.224.20 51234 10.0.0.9 22"})
+        assert driver.get_management_peer() == "10.7.224.20"
+
+    def test_no_ssh_connection_variable_no_peer(self):
+        assert _channel(FreeBSDDriver, {}).get_management_peer() is None
 
     def test_snmp_config_of_a_running_agent(self):
         driver = _channel(
@@ -556,6 +575,16 @@ class TestSnmp:
             "version": "2c",
         }
 
+    def test_a_community_that_is_not_in_the_file_is_not_assumed(self):
+        driver = _channel(
+            FreeBSDDriver,
+            {
+                "pgrep -f /usr/local/sbin/snmpd": "1234",
+                "cat /usr/local/etc/snmp/snmpd.conf": "agentAddress udp:161\n",
+            },
+        )
+        assert driver.get_snmp_config()["community"] == ""
+
     def test_no_agent_running_means_no_config(self):
         driver = _channel(OpenBSDDriver, {"pgrep -f /usr/local/sbin/snmpd": ("", 1)})
         assert driver.get_snmp_config() is None