test: add unit tests [run-int-tests]

splunk_connect_for_snmp/snmp/tasks.py

@@ -154,9 +154,7 @@ def trap(self, work):
     remotemibs = set()
     metrics = {}
 
-    # IPv4 addresses from IPv6 socket have added ::ffff: prefix, which is removed
-    if IPv6_ENABLED and "." in work["host"]:
-        work["host"] = work["host"].split(":")[-1]
+    work["host"] = format_ipv4_address(work["host"])
 
     for w in work["data"]:
 
@@ -207,3 +205,10 @@ def trap(self, work):
         "detectchange": False,
         "sourcetype": SPLUNK_SOURCETYPE_TRAPS,
     }
+
+
+def format_ipv4_address(host):
+    # IPv4 addresses from IPv6 socket have added ::ffff: prefix, which is removed
+    if IPv6_ENABLED and "." in host:
+        return host.split(":")[-1]
+    return host

test/snmp/test_tasks.py

@@ -289,3 +289,27 @@ def test_trap_reverse_dns_lookup(
             },
             result,
         )
+
+    @patch("splunk_connect_for_snmp.snmp.tasks.IPv6_ENABLED", True)
+    def test_format_ipv4_address(self, ipv6_enabled):
+        from splunk_connect_for_snmp.snmp.tasks import format_ipv4_address
+
+        ip_address = "::ffff:172.31.20.76"
+        host = format_ipv4_address(ip_address)
+        self.assertEqual(host, "172.31.20.76")
+
+    @patch("splunk_connect_for_snmp.snmp.tasks.IPv6_ENABLED", False)
+    def test_format_ipv4_address_disabled(self, ipv6_enabled):
+        from splunk_connect_for_snmp.snmp.tasks import format_ipv4_address
+
+        ip_address = "::ffff:172.31.20.76"
+        host = format_ipv4_address(ip_address)
+        self.assertEqual(host, "::ffff:172.31.20.76")
+
+    @patch("splunk_connect_for_snmp.snmp.tasks.IPv6_ENABLED", True)
+    def test_format_ipv4_address_ipv6(self, ipv6_enabled):
+        from splunk_connect_for_snmp.snmp.tasks import format_ipv4_address
+
+        ip_address = " fd02::b24a:409e:a35e:b580"
+        host = format_ipv4_address(ip_address)
+        self.assertEqual(host, " fd02::b24a:409e:a35e:b580")