Feature/mib unit tests (#28)

* code cleaning

* feat: add unit-test to MIB server
- added initial trap_translator_test using mongomock (for mongodb)
- removed unused imports/variables
- fixed Python warnings (like case-sensitiveness of variables' names)

* feat: add unit-test to MIB server (#2)
- added negative tests for format_trap

* feat: add unit-test to MIB server (#3)
- handling missing input parameters for format_trap (we now check
  that each var_bind has at least the following keys: [ "oid", "val",
  "oid_type", "val_type" ]
- added unit-test for convering it

* feat: add unit-test to MIB server (#4)
- added positive unit-tests for metrics
- added negative unit-tests for metrics

* bug(translation): exception when translating IP-MIB::ipNetToMediaPhysAddress
- we have a weird error when translating this OID: 1.3.6.1.2.1.4.22.1.2

* bug(translation): exception when translating IP-MIB::ipNetToMediaPhysAddress (#2)
- replace spaces with colon for Hex-STRING types

Co-authored-by: weliasz <[email protected]>

Author: lstoppa

pyproject.toml

@@ -15,6 +15,7 @@ Flask = "^1.1.2"
 Flask-AutoIndex = "^0.6.6"
 
 [tool.poetry.dev-dependencies]
+mongomock = "^3.22.1"
 
 [tool.poetry.scripts]
 sc4snmp-mib-server = "splunk_connect_for_snmp_mib_server.snmp_mib_server:main"

splunk_connect_for_snmp_mib_server/mib_server.py

@@ -1,8 +1,5 @@
-from flask import Flask, request, abort, jsonify, send_from_directory
+from flask import Flask, request
 from flask_autoindex import AutoIndex
-import os
-import yaml
-from splunk_connect_for_snmp_mib_server.mongo import MibsRepository, OidsRepository
 from splunk_connect_for_snmp_mib_server.translator import Translator
 import logging
 
@@ -18,9 +15,8 @@ def __init__(self, args, server_config):
 
     def build_flask_app(self):
         app = Flask(__name__)
-        # ppath = os.environ['MIBS_SERVER_URL']
-        ppath = self._server_config["snmp"]["mibs"]["mibs_path"]
-        files_index = AutoIndex(app, ppath, add_url_rules=False)
+        mibs_path = self._server_config["snmp"]["mibs"]["mibs_path"]
+        files_index = AutoIndex(app, mibs_path, add_url_rules=False)
 
         @app.route("/")
         def hello():
@@ -43,8 +39,7 @@ def translator():
             logger.debug(f"var_binds: {var_binds}")
             logger.debug(f"type of metric: {str(metric)} -- metric: {metric}")
             if metric == "True":
-                # TODO
-                # If metric is true, var_binds just has one
+                # if metric is true, var_binds has just one element
                 var_bind = var_binds[0]
                 result = self._translator.format_metric_data(var_bind)
             else:
@@ -54,5 +49,5 @@ def translator():
         return app
 
     def run_mib_server(self):
-        # TODO poetry run fails when debug=True
+        # poetry run fails when debug=True
         self._flask_app.run(host="0.0.0.0", port=self._args.port)

splunk_connect_for_snmp_mib_server/mongo.py

@@ -1,4 +1,4 @@
-from pymongo import MongoClient
+import pymongo
 import os
 import logging
 
@@ -7,26 +7,16 @@
 
 class MibsRepository:
     def __init__(self, mongo_config):
-        """
-        Create a collection in mongodb to store mib files
-        """
-        self._client = MongoClient(
+        self._client = pymongo.MongoClient(
             os.environ["MONGO_SERVICE_SERVICE_HOST"],
             int(os.environ["MONGO_SERVICE_SERVICE_PORT"]),
         )
         self._mibs = self._client[mongo_config["database"]][mongo_config["collection"]]
 
     def upload_files(self, mib_files_dir):
-        """
-        Upload mib files from dir to mongo
-        @param mib_files_dir: the path of the mib files directory
-        """
-        # TODO check duplicate before insert, using filename as PK
         for filename in os.listdir(mib_files_dir):
             file_path = mib_files_dir + "/" + filename
-            # print(file_path)
             with open(file_path, "r") as mib_file:
-                # TODO add try catch, insert only if PK doesn't exit
                 try:
                     self._mibs.insert_one(
                         dict(content=mib_file.read(), filename=filename, _id=filename)
@@ -37,41 +27,29 @@ def upload_files(self, mib_files_dir):
                     )
 
     def search_oid(self, oid):
-        """
-        Search mib file based on oid
-        @param oid: oid, format: "1, 3, 6, 1, 4, 1, 2356, 11, 0, 0, 10000"
-        @return mib filename
-        """
         data = self._mibs.find_one({"content": {"$regex": oid}})
         if data:
             return data["filename"]
         else:
             return None
 
     def delete_mib(self, filename):
-        """
-        Delete mib files based on filename
-        @param filename: mib filename
-        """
         self._mibs.delete_many({"filename": {"$regex": filename}})
 
     def clear(self):
-        """
-        Clear the collection
-        """
         self._mibs.remove()
 
 
 class OidsRepository:
     def __init__(self, mongo_config):
-        self._client = MongoClient(
+        self._client = pymongo.MongoClient(
             os.environ["MONGO_SERVICE_SERVICE_HOST"],
             int(os.environ["MONGO_SERVICE_SERVICE_PORT"]),
         )
         self._oids = self._client[mongo_config["database"]][mongo_config["collection"]]
 
     def contains_oid(self, oid):
-        return self._oids.find({"oid": oid}).count()
+        return self._oids.count_documents({"oid": oid})
 
     def add_oid(self, oid):
         self._oids.insert_one({"oid": oid})

splunk_connect_for_snmp_mib_server/snmp_mib_server.py

@@ -1,7 +1,6 @@
 import argparse
-import os
 import yaml
-from splunk_connect_for_snmp_mib_server.mongo import MibsRepository, OidsRepository
+from splunk_connect_for_snmp_mib_server.mongo import MibsRepository
 from splunk_connect_for_snmp_mib_server.mib_server import MibServer
 import logging
 
@@ -36,25 +35,19 @@ def main():
         "--hec_threads", default=10, help="Max http worker thread count", type=int
     )
     parser.add_argument("-c", "--config", default="config.yaml", help="Config File")
-
     args = parser.parse_args()
-
     log_level = args.loglevel.upper()
     config_file = args.config
 
     logging.getLogger().setLevel(log_level)
     logger.info(f"Log Level is {log_level}")
     logger.info(f"Config file is {config_file}")
-
     logger.info("Completed Argument parsing")
 
     with open(config_file, "r") as yamlfile:
         server_config = yaml.load(yamlfile, Loader=yaml.FullLoader)
 
-    # Upload mib fiels into mongo
     upload_mibs(server_config)
-
-    # Init MibServer
     mib_server = MibServer(args, server_config)
     mib_server.run_mib_server()