diff --git a/splunk_connect_for_snmp_mib_server/mib_server.py b/splunk_connect_for_snmp_mib_server/mib_server.py index 9fbbd8be..fdcaf089 100644 --- a/splunk_connect_for_snmp_mib_server/mib_server.py +++ b/splunk_connect_for_snmp_mib_server/mib_server.py @@ -13,7 +13,6 @@ # See the License for the specific language governing permissions and # limitations under the License. # ######################################################################## -import json from flask import Flask, request from flask_autoindex import AutoIndex @@ -24,7 +23,6 @@ class MibServer: - def __init__(self, args, server_config): self._args = args self._server_config = server_config @@ -33,7 +31,7 @@ def __init__(self, args, server_config): self.data_format = { "METRIC": self._translator.format_metric_data, "TEXT": self._translator.format_text_event, - "MULTIMETRIC": self._translator.format_multimetric_data + "MULTIMETRIC": self._translator.format_multimetric_data, } def build_flask_app(self): diff --git a/splunk_connect_for_snmp_mib_server/translator.py b/splunk_connect_for_snmp_mib_server/translator.py index 1d9f0f76..f9c2088f 100644 --- a/splunk_connect_for_snmp_mib_server/translator.py +++ b/splunk_connect_for_snmp_mib_server/translator.py @@ -349,7 +349,9 @@ def format_text_event(self, var_binds: list): def format_metric_data(self, var_bind: list, index=0): """ format one varBind object into metric format - @param var_bind: single varBind object + :param var_bind: varBind list, usually it contains only one element + :param index: index of the element we want to process + :return: translated data dictionary """ var_bind = var_bind[index] metric_data = {} @@ -376,7 +378,7 @@ def format_metric_data(self, var_bind: list, index=0): # .. if the metric name contains a . replace with _ metric_data[ "metric_name" - ] = f'sc4snmp.{translated_oid.replace(".","_").replace("::", ".")}' + ] = f'sc4snmp.{translated_oid.replace(".", "_").replace("::", ".")}' metric_data["_value"] = translated_val metric_data["metric_type"] = val_type if custom_translated_oid: @@ -399,6 +401,9 @@ def format_multimetric_data(self, varbinds: list) -> dict: """ result_dict = self.format_metric_data(varbinds) result_string = self.format_text_event(varbinds) - result = {'metric_name': result_dict['metric_name'], 'metric': json.dumps(result_dict), - 'non_metric': result_string} + result = { + "metric_name": result_dict["metric_name"], + "metric": json.dumps(result_dict), + "non_metric": result_string, + } return result diff --git a/tests/test_translator.py b/tests/test_translator.py index 0e58b457..accc00eb 100644 --- a/tests/test_translator.py +++ b/tests/test_translator.py @@ -16,12 +16,10 @@ from unittest import TestCase, mock import mongomock -import yaml from splunk_connect_for_snmp_mib_server.translator import Translator import os import logging -import json from pysnmp.hlapi import * logger = logging.getLogger(__name__) @@ -91,7 +89,7 @@ def setUp(self): "mib": {"database": "files", "collection": "mib_files"}, }, } - #server_config["snmp"]["mibs"]["dir"] = "../mibs/pysnmp" + # server_config["snmp"]["mibs"]["dir"] = "../mibs/pysnmp" self.my_translator = Translator(server_config) @mongomock.patch() @@ -193,11 +191,13 @@ def test_format_trap_invalid_input(self): @mongomock.patch() def test_format_metric(self): - input_var_binds = [{ - "oid": "1.3.6.1.2.1.1.3.0", - "val": "123", - "val_type": "TimeTicks", - }] + input_var_binds = [ + { + "oid": "1.3.6.1.2.1.1.3.0", + "val": "123", + "val_type": "TimeTicks", + } + ] translated_dict = self.my_translator.format_metric_data(input_var_binds) for required_key in ["metric_name", "_value", "metric_type"]: assert required_key in translated_dict @@ -208,11 +208,13 @@ def test_format_metric(self): @mongomock.patch() def test_format_non_existing_metric(self): - input_var_binds = [{ - "oid": "1.3.6666.16666.26666.16666.1.3.0", - "val": "123", - "val_type": "TimeTicks", - }] + input_var_binds = [ + { + "oid": "1.3.6666.16666.26666.16666.1.3.0", + "val": "123", + "val_type": "TimeTicks", + } + ] translated_dict = self.my_translator.format_metric_data(input_var_binds) for required_key in ["metric_name", "_value", "metric_type"]: assert required_key in translated_dict @@ -279,11 +281,13 @@ def test_translate_all_snmp_simulator_data_types(self): and len(value_types) == len(expected_translations) ) for index in range(0, len(oids)): - input_var_binds_colons = [{ - "oid": oids[index], - "val": str_values[index], - "val_type": value_types[index], - }] + input_var_binds_colons = [ + { + "oid": oids[index], + "val": str_values[index], + "val_type": value_types[index], + } + ] translated_metrics = self.my_translator.format_metric_data( input_var_binds_colons ) @@ -309,7 +313,5 @@ def test_more_mib_files(self): ] for i in range(0, len(input_var_binds)): - translated_dict = self.my_translator.format_metric_data( - input_var_binds, i - ) + translated_dict = self.my_translator.format_metric_data(input_var_binds, i) assert translated_dict["metric_name"] == expected_values[i]