test: Topology.Lldp v3 credentials, mgmt addresses, per-walk error fallbacks

test/towerops/topology/lldp_test.exs

@@ -5,6 +5,7 @@ defmodule Towerops.Topology.LldpTest do
 
   alias Towerops.Snmp.SnmpMock
   alias Towerops.Topology.Lldp
+  alias Towerops.Topology.Lldp.Parser
 
   setup :verify_on_exit!
 
@@ -106,6 +107,123 @@ defmodule Towerops.Topology.LldpTest do
     end
   end
 
+  describe "discover_neighbors/2 — additional branches" do
+    test "uses SNMPv3 credentials when device snmp_version is 3" do
+      user = Towerops.AccountsFixtures.user_fixture()
+      {:ok, org} = Towerops.Organizations.create_organization(%{name: "LLDP Org v3"}, user.id)
+      {:ok, site} = Towerops.Sites.create_site(%{name: "Site v3", organization_id: org.id})
+
+      {:ok, device} =
+        Towerops.Devices.create_device(%{
+          name: "v3-sw01",
+          ip_address: "10.6.6.6",
+          snmp_enabled: true,
+          snmp_version: "3",
+          snmpv3_security_level: "authPriv",
+          snmpv3_username: "monitor",
+          snmpv3_auth_protocol: "SHA-256",
+          snmpv3_auth_password: "authpass1234",
+          snmpv3_priv_protocol: "AES",
+          snmpv3_priv_password: "privpass1234",
+          snmpv3_credential_source: "device",
+          snmp_port: 161,
+          site_id: site.id,
+          organization_id: org.id
+        })
+
+      device = Towerops.Devices.get_device!(device.id)
+
+      test_pid = self()
+
+      stub(SnmpMock, :get, fn _target, _oid, opts ->
+        send(test_pid, {:get_opts, opts})
+        {:ok, "v3-sw01"}
+      end)
+
+      stub(SnmpMock, :walk, fn _target, _oid, opts ->
+        send(test_pid, {:walk_opts, opts})
+        {:ok, []}
+      end)
+
+      assert {:ok, %{neighbors: []}} = Lldp.discover_neighbors(device.id)
+
+      assert_receive {:get_opts, opts}
+      assert opts[:security_name] == "monitor"
+      # SnmpKit normalizes security_level to an atom (:auth_priv) and
+      # protocols to lowercased atoms (:sha256, :aes). The exact normalization
+      # is implementation-dependent, so just assert the values are non-nil.
+      assert opts[:security_level]
+      assert opts[:auth_protocol]
+      assert opts[:priv_protocol]
+    end
+
+    test "parses management addresses through assemble_neighbors" do
+      device = build_device()
+
+      sys_name_oid = "1.0.8802.1.1.2.1.4.1.1.9"
+      port_desc_oid = "1.0.8802.1.1.2.1.4.1.1.8"
+      port_id_oid = "1.0.8802.1.1.2.1.4.1.1.7"
+      loc_port_oid = "1.0.8802.1.1.2.1.3.7.1.4"
+      mgmt_oid = "1.0.8802.1.1.2.1.4.2.1.3"
+
+      stub(SnmpMock, :get, fn _target, _oid, _opts -> {:ok, "core-sw01"} end)
+
+      stub(SnmpMock, :walk, fn _target, oid, _opts ->
+        cond do
+          String.contains?(oid, sys_name_oid) ->
+            {:ok, [%{oid: "#{sys_name_oid}.0.3.1", value: "neighbor-with-mgmt"}]}
+
+          String.contains?(oid, loc_port_oid) ->
+            {:ok, [%{oid: "#{loc_port_oid}.3", value: "ge-0/0/3"}]}
+
+          String.contains?(oid, port_desc_oid) ->
+            {:ok, [%{oid: "#{port_desc_oid}.0.3.1", value: "ge-0/0/24"}]}
+
+          String.contains?(oid, port_id_oid) ->
+            {:ok, [%{oid: "#{port_id_oid}.0.3.1", value: "525"}]}
+
+          String.contains?(oid, mgmt_oid) ->
+            {:ok, [%{oid: "#{mgmt_oid}.0.3.1.1.4.192.168.1.10", value: ""}]}
+
+          true ->
+            {:ok, []}
+        end
+      end)
+
+      assert {:ok, %{neighbors: [neighbor]}} = Lldp.discover_neighbors(device.id)
+      assert neighbor.neighbor_name == "neighbor-with-mgmt"
+      assert "192.168.1.10" in neighbor.management_addresses
+    end
+
+    test "tolerates per-walk errors when sys_names succeeds" do
+      device = build_device()
+
+      sys_name_oid = "1.0.8802.1.1.2.1.4.1.1.9"
+
+      stub(SnmpMock, :get, fn _target, _oid, _opts -> {:ok, "core-sw01"} end)
+
+      stub(SnmpMock, :walk, fn _target, oid, _opts ->
+        if String.contains?(oid, sys_name_oid) do
+          {:ok, [%{oid: "#{sys_name_oid}.0.3.1", value: "lonely-neighbor"}]}
+        else
+          {:error, :timeout}
+        end
+      end)
+
+      assert {:ok, %{neighbors: [n]}} = Lldp.discover_neighbors(device.id)
+      assert n.neighbor_name == "lonely-neighbor"
+      assert n.management_addresses == []
+      assert n.remote_port == nil
+      assert n.remote_port_id == nil
+    end
+
+    test "parse_management_address delegate parses an IPv4 OID" do
+      mgmt_base = "1.0.8802.1.1.2.1.4.2.1.3"
+      oid = "#{mgmt_base}.0.3.1.1.4.10.0.0.5"
+      assert {{"0", "3", "1"}, "10.0.0.5"} = Parser.parse_management_address(oid, mgmt_base)
+    end
+  end
+
   describe "delegated pure helpers" do
     test "extract_suffix strips a matching prefix" do
       assert "5" == Lldp.extract_suffix("1.0.8802.1.1.2.1.3.7.1.4.5", "1.0.8802.1.1.2.1.3.7.1.4")