fix: change way of checking if profile is an OID to use regex (#112)

* fix: change way of checking if profile is an OID to use regex, add unit tests

Author: omrozowicz-splunk

splunk_connect_for_snmp_poller/manager/task_utilities.py

@@ -15,6 +15,7 @@
 #
 import json
 import os
+import re
 from collections import namedtuple
 
 from celery.utils.log import get_task_logger
@@ -778,3 +779,14 @@ def build_contextData(version, community, server_config):
         return ContextData(contextEngineId, contextName)
     except Exception as e:
         logger.error(f"Error happend while building ContextData: {e}")
+
+
+def is_oid(profile: str) -> bool:
+    """
+    This function checks if profile is an OID. OID is defined as a string of format:
+        - ex. 1.3.6.1.2.1.2
+        - ex. 1.3.6.2.1.*
+    @param profile: variable from inventory file, can be a name of profile or an OID
+    @return: if the profile is of OID structure
+    """
+    return bool(re.match(r"^\d(\.\d*)*(\.\*)?$", profile))

splunk_connect_for_snmp_poller/manager/tasks.py

@@ -24,6 +24,7 @@
     VarbindCollection,
     build_authData,
     build_contextData,
+    is_oid,
     mib_string_handler,
     parse_port,
     snmp_bulk_handler,
@@ -150,7 +151,7 @@ def snmp_polling(
 
     try:
         # Perform SNNP Polling for string profile in inventory.csv
-        if "." not in profile:
+        if not is_oid(profile):
             logger.info(
                 f"Executing SNMP Polling for Varbinds in config.yaml for {host} profile={profile}"
             )

tests/test_task_utilities.py

@@ -18,6 +18,7 @@
 from splunk_connect_for_snmp_poller.manager.task_utilities import (
     _sort_walk_data,
     is_metric_data,
+    is_oid,
     parse_port,
 )
 
@@ -110,3 +111,23 @@ def test__sort_walk_data_non_metric(self):
         self.assertEqual(merged_result_metric, [])
         self.assertEqual(merged_result, [varbind_metric_dict])
         self.assertEqual(merged_result_non_metric, [varbind])
+
+    def test_is_oid_asterisk(self):
+        oid = "1.3.6.1.2.1.2.2.1.1.*"
+        self.assertTrue(is_oid(oid))
+
+    def test_is_oid(self):
+        oid = "1.3.6.1.2.1"
+        self.assertTrue(is_oid(oid))
+
+    def test_is_oid_multinumber(self):
+        oid = "1.3.6.1.2.195.218.254.105.56134.205.188.8.43.5190"
+        self.assertTrue(is_oid(oid))
+
+    def test_is_oid_profile(self):
+        oid = "router"
+        self.assertFalse(is_oid(oid))
+
+    def test_is_oid_profile_with_dot(self):
+        oid = "router.12"
+        self.assertFalse(is_oid(oid))