From 0dcc3d03ef57df0d0e3f63255edcb438abc4d78b Mon Sep 17 00:00:00 2001 From: ajasnosz Date: Wed, 18 Dec 2024 15:49:52 +0100 Subject: [PATCH] test: add unit tests [run-int-tests] --- splunk_connect_for_snmp/snmp/tasks.py | 11 ++++++++--- test/snmp/test_tasks.py | 24 ++++++++++++++++++++++++ 2 files changed, 32 insertions(+), 3 deletions(-) diff --git a/splunk_connect_for_snmp/snmp/tasks.py b/splunk_connect_for_snmp/snmp/tasks.py index 65088e7a7..e30c843ab 100644 --- a/splunk_connect_for_snmp/snmp/tasks.py +++ b/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 diff --git a/test/snmp/test_tasks.py b/test/snmp/test_tasks.py index 321a11a89..a0c113bf3 100644 --- a/test/snmp/test_tasks.py +++ b/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")