refactor: use conf-parser in eventgen.conf parser

Author: artemrys

pytest_splunk_addon/standard_lib/index_tests/test_generator.py

@@ -49,11 +49,12 @@ def generate_tests(self, store_events, app_path, config_path, test_type):
         store_sample = sample_generator.get_samples(store_events)
         tokenized_events = store_sample.get("tokenized_events")
         if not store_sample.get("conf_name") == "psa-data-gen":
-            LOGGER.warning(
-                "Index Time tests cannot be executed using eventgen.conf, pytest-splunk-addon-data.conf is required."
+            msg = (
+                "Index time tests cannot be executed without "
+                "pytest-splunk-addon-data.conf"
             )
-            return " Index Time tests cannot be executed using eventgen.conf,\
-                 pytest-splunk-addon-data.conf is required."
+            LOGGER.warning(msg)
+            return msg
 
         if test_type == "line_breaker":
             LOGGER.info("Generating line breaker test")

pytest_splunk_addon/standard_lib/sample_generation/__init__.py

@@ -16,7 +16,7 @@
 from .sample_event import SampleEvent
 from .rule import Rule, raise_warning
 from .sample_stanza import SampleStanza
-from .eventgen_parser import EventgenParser
+from .pytest_splunk_addon_data_parser import PytestSplunkAddonDataParser
 from .sample_event import SampleEvent
 from .sample_generator import SampleGenerator
 from .sample_xdist_generator import SampleXdistGenerator

pytest_splunk_addon/standard_lib/sample_generation/eventgen_parser.py renamed to pytest_splunk_addon/standard_lib/sample_generation/pytest_splunk_addon_data_parser.py

@@ -16,28 +16,31 @@
 import os
 import re
 import logging
-from splunk_appinspect import App
-from .rule import Rule, raise_warning
+from .rule import raise_warning
 from . import SampleStanza
 
+import addonfactory_splunk_conf_parser_lib as conf_parser
+
 LOGGER = logging.getLogger("pytest-splunk-addon")
-import warnings
 
 
-class EventgenParser:
+PSA_DATA_CONFIG_FILE = "pytest-splunk-addon-data.conf"
+
+
+class PytestSplunkAddonDataParser:
     """
-    This class represents the entire eventgen.conf file and handles parsing mechanism of eventgen and the rules.
+    This class parses pytest-splunk-addon-data.conf file.
 
     Args:
-        addon_path (str): Path to the Splunk App
+        addon_path: Path to the Splunk App
     """
 
     conf_name = " "
 
-    def __init__(self, addon_path, config_path=None):
-        self._app = App(addon_path, python_analyzer_enable=False)
+    def __init__(self, addon_path: str, config_path: str):
+        self._conf_parser = conf_parser.TABConfigParser()
         self.config_path = config_path
-        self._eventgen = None
+        self._psa_data = None
         self.addon_path = addon_path
         self.match_stanzas = set()
 
@@ -71,64 +74,36 @@ def path_to_samples(self):
             return os.path.join(self.addon_path, "samples")
 
     @property
-    def eventgen(self):
-        try:
-            relative_path = os.path.relpath(self.config_path, self.addon_path)
-            if os.path.exists(
-                os.path.join(self.config_path, "pytest-splunk-addon-data.conf")
-            ):
-                self._eventgen = self._app.get_config(
-                    "pytest-splunk-addon-data.conf", dir=relative_path
-                )
-                self.conf_name = "psa-data-gen"
-                path = self._app.get_filename(
-                    relative_path, "pytest-splunk-addon-data.conf"
-                )
-
-            elif os.path.exists(os.path.join(self.config_path, "eventgen.conf")):
-
-                self._eventgen = self._app.get_config(
-                    "eventgen.conf", dir=relative_path
-                )
-                self.conf_name = "eventgen"
-                path = self._app.get_filename(relative_path, "eventgen.conf")
-
-            else:
-                self._eventgen = self._app.get_config("eventgen.conf")
-                self.conf_name = "eventgen"
-                path = self._app.get_filename("default", "eventgen.conf")
-            LOGGER.info(
-                "Using Eventgen path: {e}\nUsing Conf file name: {c}".format(
-                    e=path, c=self.conf_name
-                )
-            )
-            return self._eventgen
-
-        except OSError:
-            LOGGER.warning("pytest-splunk-addon-data.conf/eventgen.conf not Found")
-            raise FileNotFoundError(
-                "pytest-splunk-addon-data.conf/eventgen.conf not Found"
-            )
+    def psa_data(self):
+        path = os.path.join(self.config_path, PSA_DATA_CONFIG_FILE)
+        if os.path.exists(path):
+            self._conf_parser.read(path)
+            self.conf_name = "psa-data-gen"
+            self._psa_data = self._conf_parser.item_dict()
+            return self._psa_data
+        else:
+            LOGGER.warning(f"{PSA_DATA_CONFIG_FILE} not found")
+            raise FileNotFoundError(f"{PSA_DATA_CONFIG_FILE} not found")
 
     def get_sample_stanzas(self):
         """
-        Converts a stanza in eventgen.conf to an object of SampleStanza.
+        Converts a stanza in pytest-splunk-addon-data.conf to an object of SampleStanza.
 
         Yields:
             SampleStanza Object
         """
-        eventgen_dict = self.get_eventgen_stanzas()
+        _psa_data = self.get_psa_data_stanzas()
         self.check_samples()
-        for sample_name, stanza_params in sorted(eventgen_dict.items()):
+        for sample_name, stanza_params in sorted(_psa_data.items()):
             sample_path = os.path.join(self.path_to_samples, sample_name)
             yield SampleStanza(
                 sample_path,
                 stanza_params,
             )
 
-    def get_eventgen_stanzas(self):
+    def get_psa_data_stanzas(self):
         """
-        Parses the eventgen.conf file and converts it into a dictionary.
+        Parses the pytest-splunk-addon-data.conf file and converts it into a dictionary.
 
         Format::
 
@@ -149,44 +124,39 @@ def get_eventgen_stanzas(self):
             }
 
         Return:
-            Dictionary representing eventgen.conf in the above format.
+            Dictionary representing pytest-splunk-addon-data.conf in the above format.
         """
-        eventgen_dict = {}
+        psa_data_dict = {}
         if os.path.exists(self.path_to_samples):
             for sample_file in os.listdir(self.path_to_samples):
-                for stanza in sorted(self.eventgen.sects):
+                for stanza, fields in sorted(self.psa_data.items()):
                     stanza_match_obj = re.search(stanza, sample_file)
                     if stanza_match_obj and stanza_match_obj.group(0) == sample_file:
                         self.match_stanzas.add(stanza)
-                        eventgen_sections = self.eventgen.sects[stanza]
-                        eventgen_dict.setdefault((sample_file), {"tokens": {}})
-                        for stanza_param in eventgen_sections.options:
-                            eventgen_property = eventgen_sections.options[stanza_param]
-                            if eventgen_property.name.startswith("token"):
-                                _, token_id, token_param = eventgen_property.name.split(
-                                    "."
-                                )
-                                token_key = "{}_{}".format(stanza, token_id)
+                        psa_data_dict.setdefault(sample_file, {"tokens": {}})
+                        for key, value in fields.items():
+                            if key.startswith("token"):
+                                _, token_id, token_param = key.split(".")
+                                token_key = f"{stanza}_{token_id}"
                                 if (
                                     not token_key
-                                    in eventgen_dict[sample_file]["tokens"].keys()
+                                    in psa_data_dict[sample_file]["tokens"].keys()
                                 ):
-                                    eventgen_dict[sample_file]["tokens"][token_key] = {}
-                                eventgen_dict[sample_file]["tokens"][token_key][
+                                    psa_data_dict[sample_file]["tokens"][token_key] = {}
+                                psa_data_dict[sample_file]["tokens"][token_key][
                                     token_param
-                                ] = eventgen_property.value
+                                ] = value
                             else:
-                                eventgen_dict[sample_file][
-                                    eventgen_property.name
-                                ] = eventgen_property.value
-        return eventgen_dict
+                                psa_data_dict[sample_file][key] = value
+        return psa_data_dict
 
     def check_samples(self):
         """
-        Gives a user warning when sample file is not found for the stanza peresent in the configuration file.
+        Gives a user warning when sample file is not found for the stanza
+        present in the configuration file.
         """
         if os.path.exists(self.path_to_samples):
-            for stanza in self.eventgen.sects:
+            for stanza in self.psa_data.keys():
                 if stanza not in self.match_stanzas:
                     raise_warning("No sample file found for stanza : {}".format(stanza))
                 LOGGER.info("Sample file found for stanza : {}".format(stanza))

pytest_splunk_addon/standard_lib/sample_generation/sample_generator.py

@@ -13,10 +13,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-import time
-from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor
+from concurrent.futures import ThreadPoolExecutor
 
-from . import EventgenParser
+from . import PytestSplunkAddonDataParser
 from . import SampleStanza
 from itertools import cycle
 
@@ -47,14 +46,13 @@ def get_samples(self):
         Generate SampleEvent object
         """
         if not SampleGenerator.sample_stanzas:
-            eventgen_parser = EventgenParser(
+            eventgen_parser = PytestSplunkAddonDataParser(
                 self.addon_path, config_path=self.config_path
             )
             sample_stanzas = list(eventgen_parser.get_sample_stanzas())
             SampleGenerator.conf_name = eventgen_parser.conf_name
             with ThreadPoolExecutor(min(20, max(len(sample_stanzas), 1))) as t:
                 t.map(SampleStanza.get_raw_events, sample_stanzas)
-            # with ProcessPoolExecutor(self.process_count) as p:
             _ = list(
                 map(
                     SampleStanza.tokenize,

tests/unit/tests_standard_lib/test_index_tests/test_test_generator.py

@@ -26,7 +26,7 @@ def test_generate_tests_without_conf_file(mock_object, caplog):
         )
     )
     assert caplog.messages == [
-        "Index Time tests cannot be executed using eventgen.conf, pytest-splunk-addon-data.conf is required."
+        "Index time tests cannot be executed without pytest-splunk-addon-data.conf"
     ]
 
 

tests/unit/tests_standard_lib/test_utilities/test_create_new_eventgen.py

@@ -1,9 +1,7 @@
 import pytest
 from unittest.mock import MagicMock, patch, PropertyMock, call
-from collections import namedtuple
 from pytest_splunk_addon.standard_lib.utilities.create_new_eventgen import (
     UpdateEventgen,
-    main,
 )
 
 output_to_build = {
@@ -58,16 +56,6 @@ def os_listdir_mock(monkeypatch):
     return listdir
 
 
-@pytest.fixture()
-def update_eventgen_mock(monkeypatch):
-    ue = MagicMock()
-    monkeypatch.setattr(
-        "pytest_splunk_addon.standard_lib.utilities.create_new_eventgen.UpdateEventgen",
-        ue,
-    )
-    return ue
-
-
 def test_update_eventgen_instantiation(app_mock):
     app_mock.return_value = "APP_RETURN_VALUE"
     event = UpdateEventgen("fake_path")
@@ -345,22 +333,3 @@ def test_create_new_eventgen(
             call.__exit__(None, None, None),
         ]
     )
-
-
-def test_main(argparse_mock, update_eventgen_mock):
-    update_eventgen_mock.return_value = update_eventgen_mock
-    update_eventgen_mock.get_eventgen_stanzas.return_value = {"fake": "init"}
-    update_eventgen_mock.update_eventgen_stanzas.return_value = {"fake": "updated"}
-    args = namedtuple("Namespace", ["addon_path", "new_conf_path"])
-    argparse_mock.parse_args.return_value = args(
-        "fake_addon_path", "fake_new_conf_path"
-    )
-    main()
-    update_eventgen_mock.assert_has_calls(
-        [
-            call("fake_addon_path"),
-            call.get_eventgen_stanzas(),
-            call.update_eventgen_stanzas({"fake": "init"}),
-            call.create_new_eventgen({"fake": "updated"}, "fake_new_conf_path"),
-        ]
-    )

tests/unit/tests_standard_lib/tests_sample_generation/test_data/with_samples/pytest-splunk-addon-data.conf

@@ -0,0 +1,44 @@
+[test1.samples]
+sourcetype = test:sourcetype
+source = source://test
+input_type = modinput
+host_type = plugin
+sourcetype_to_search = test:sourcetype
+timestamp_type = event
+sample_count = 1
+
+token.0.token = ##Timestamp##
+token.0.replacementType = timestamp
+token.0.replacement = %Y-%m-%dT%H:%M:%S
+token.0.field = _time
+
+token.1.token = ##user##
+token.1.replacementType = random
+token.1.replacement = list["[email protected]","[email protected]"]
+
+token.2.token = ##ip##
+token.2.replacementType = random
+token.2.replacement = src["ipv4"]
+token.2.field = src
+
+token.3.token = ##number##
+token.3.replacementType = random
+token.3.replacement = integer[100000000000000000000:999999999999999999999]
+
+[test2.samples]
+sourcetype = test:sourcetype
+source = source://test:text
+input_type = modinput
+host_type = plugin
+sourcetype_to_search = test:sourcetype
+timestamp_type = event
+sample_count = 1
+
+token.1.token = ##user##
+token.1.replacementType = random
+token.1.replacement = list["[email protected]","[email protected]"]
+
+token.2.token = ##ip##
+token.2.replacementType = random
+token.2.replacement = src["ipv4"]
+token.2.field = src

tests/unit/tests_standard_lib/tests_sample_generation/test_data/with_samples/samples/test1.samples

@@ -0,0 +1 @@
+{"number": "##number##", "ip": "##ip##", "user": "##user##", "time": "##Timestamp##"}

tests/unit/tests_standard_lib/tests_sample_generation/test_data/with_samples/samples/test2.samples

@@ -0,0 +1 @@
+Src ip address ##ip## for user ##user##

tests/unit/tests_standard_lib/tests_sample_generation/test_data/without_samples/pytest-splunk-addon-data.conf

@@ -0,0 +1,44 @@
+[test1.samples]
+sourcetype = test:sourcetype
+source = source://test
+input_type = modinput
+host_type = plugin
+sourcetype_to_search = test:sourcetype
+timestamp_type = event
+sample_count = 1
+
+token.0.token = ##Timestamp##
+token.0.replacementType = timestamp
+token.0.replacement = %Y-%m-%dT%H:%M:%S
+token.0.field = _time
+
+token.1.token = ##user##
+token.1.replacementType = random
+token.1.replacement = list["[email protected]","[email protected]"]
+
+token.2.token = ##ip##
+token.2.replacementType = random
+token.2.replacement = src["ipv4"]
+token.2.field = src
+
+token.3.token = ##number##
+token.3.replacementType = random
+token.3.replacement = integer[100000000000000000000:999999999999999999999]
+
+[test2.samples]
+sourcetype = test:sourcetype
+source = source://test:text
+input_type = modinput
+host_type = plugin
+sourcetype_to_search = test:sourcetype
+timestamp_type = event
+sample_count = 1
+
+token.1.token = ##user##
+token.1.replacementType = random
+token.1.replacement = list["[email protected]","[email protected]"]
+
+token.2.token = ##ip##
+token.2.replacementType = random
+token.2.replacement = src["ipv4"]
+token.2.field = src