chore: minor refactor

Author: mkolasinski-splunk

pytest_splunk_addon/app_test_generator.py

@@ -45,14 +45,16 @@ class AppTestGenerator(object):
     def __init__(self, pytest_config):
         self.pytest_config = pytest_config
         self.seen_tests = set()
+        self.ingest_with_uuid = self.pytest_config.getoption("ingest_with_uuid")
+        self.config_path = self.pytest_config.getoption("splunk_data_generator")
+        self.store_events = self.pytest_config.getoption("store_events")
 
-        store_events = self.pytest_config.getoption("store_events")
-        config_path = self.pytest_config.getoption("splunk_data_generator")
-        ingest_with_uuid = self.pytest_config.getoption("ingest_with_uuid")
         sample_generator = SampleXdistGenerator(
-            self.pytest_config.getoption("splunk_app"), ingest_with_uuid, config_path
+            self.pytest_config.getoption("splunk_app"),
+            self.ingest_with_uuid,
+            self.config_path,
         )
-        store_sample = sample_generator.get_samples(store_events)
+        store_sample = sample_generator.get_samples(self.store_events)
         self.tokenized_events = store_sample.get("tokenized_events")
         LOGGER.debug("Initializing FieldTestGenerator to generate the test cases")
         self.fieldtest_generator = FieldTestGenerator(
@@ -72,6 +74,38 @@ def __init__(self, pytest_config):
         )
         self.indextime_test_generator = IndexTimeTestGenerator()
 
+    def _generate_indextime_tests(self, fixture):
+        """
+        Generate index time tests based on the fixture type.
+
+        Args:
+            fixture (str): The fixture name containing the test type
+
+        Returns:
+            list: List of pytest parameters for the specified test type
+        """
+        app_path = self.pytest_config.getoption("splunk_app")
+        config_path = self.pytest_config.getoption("splunk_data_generator")
+
+        if "key_fields" in fixture:
+            test_type = "key_fields"
+        elif "_time" in fixture:
+            test_type = "_time"
+        elif "line_breaker" in fixture:
+            test_type = "line_breaker"
+        else:
+            return []
+
+        return list(
+            self.indextime_test_generator.generate_tests(
+                self.store_events,
+                app_path=app_path,
+                config_path=config_path,
+                test_type=test_type,
+                ingest_with_uuid=self.ingest_with_uuid,
+            )
+        )
+
     def generate_tests(self, fixture):
         """
         Generate the test cases based on the fixture provided
@@ -101,51 +135,8 @@ def generate_tests(self, fixture):
                 self.cim_test_generator.generate_tests(fixture),
                 fixture,
             )
-
         elif fixture.startswith("splunk_indextime"):
-            # TODO: What should be the id of the test case?
-            # Sourcetype + Host + Key field + _count
-
-            pytest_params = None
-
-            store_events = self.pytest_config.getoption("store_events")
-            app_path = self.pytest_config.getoption("splunk_app")
-            config_path = self.pytest_config.getoption("splunk_data_generator")
-            ingest_with_uuid = self.pytest_config.getoption("ingest_with_uuid")
-
-            if "key_fields" in fixture:
-                pytest_params = list(
-                    self.indextime_test_generator.generate_tests(
-                        store_events,
-                        app_path=app_path,
-                        config_path=config_path,
-                        test_type="key_fields",
-                        ingest_with_uuid=ingest_with_uuid,
-                    )
-                )
-
-            elif "_time" in fixture:
-                pytest_params = list(
-                    self.indextime_test_generator.generate_tests(
-                        store_events,
-                        app_path=app_path,
-                        config_path=config_path,
-                        test_type="_time",
-                        ingest_with_uuid=ingest_with_uuid,
-                    )
-                )
-
-            elif "line_breaker" in fixture:
-                pytest_params = list(
-                    self.indextime_test_generator.generate_tests(
-                        store_events,
-                        app_path=app_path,
-                        config_path=config_path,
-                        test_type="line_breaker",
-                        ingest_with_uuid=ingest_with_uuid,
-                    )
-                )
-
+            pytest_params = self._generate_indextime_tests(fixture)
             yield from sorted(pytest_params, key=lambda param: param.id)
 
     def dedup_tests(self, test_list, fixture):

pytest_splunk_addon/event_ingestors/hec_event_ingestor.py

@@ -91,7 +91,7 @@ def ingest(self, events, thread_count):
                 "event": event.event,
                 "index": event.metadata.get("index", "main"),
             }
-            if event.metadata.get("ingest_with_uuid") == "true":
+            if event.metadata.get("ingest_with_uuid"):
                 event_dict["fields"] = {"unique_identifier": event.unique_identifier}
 
             if event.metadata.get("host_type") in ("plugin", None):

pytest_splunk_addon/fields_tests/test_generator.py

@@ -194,7 +194,7 @@ def generate_requirements_datamodels_tests(self):
                 "datamodels": datamodels,
                 "stanza": escaped_event,
             }
-            if event.metadata.get("ingest_with_uuid") == "true":
+            if event.metadata.get("ingest_with_uuid"):
                 sample_event["unique_identifier"] = event.unique_identifier
             yield pytest.param(
                 sample_event,
@@ -270,8 +270,7 @@ def generate_requirements_tests(self):
                     "modinput_params": modinput_params,
                 }
 
-                # Add unique_identifier if UUID ingestion is enabled and event has UUID
-                if metadata.get("ingest_with_uuid") == "true":
+                if metadata.get("ingest_with_uuid"):
                     unique_identifier = getattr(event, "unique_identifier", None)
                     if unique_identifier is not None:
                         sample_event["unique_identifier"] = unique_identifier

pytest_splunk_addon/index_tests/test_generator.py

@@ -33,7 +33,7 @@ class IndexTimeTestGenerator(object):
     """
 
     def generate_tests(
-        self, store_events, app_path, config_path, test_type, ingest_with_uuid="false"
+        self, store_events, app_path, config_path, test_type, ingest_with_uuid=False
     ):
         """
         Generates the test cases based on test_type
@@ -43,7 +43,7 @@ def generate_tests(
             app_path (str): Path of the app package
             config_path (str): Path of package which contains pytest-splunk-addon-data.conf
             test_type (str): Type of test case
-            ingest_with_uuid (str): Whether to ingest events with UUID
+            ingest_with_uuid (bool): Whether to ingest events with UUID
 
         Yields:
             pytest.params for the test templates

pytest_splunk_addon/sample_generation/sample_event.py

@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-import uuid
 import re
 import logging
 from ..index_tests import key_fields
@@ -68,7 +67,6 @@ def __init__(self, event_string, metadata, sample_name, requirement_test_data=No
         self.time_values = list()
         self.metadata = metadata
         self.sample_name = sample_name
-        # UUID will be assigned post-tokenization per final event to avoid duplicates
         self.host_count = 0
         self.requirement_test_data = requirement_test_data
 

pytest_splunk_addon/sample_generation/sample_stanza.py

@@ -48,7 +48,7 @@ class SampleStanza(object):
         psa_data_params (dict): Dictionary representing pytest-splunk-addon-data.conf
     """
 
-    def __init__(self, sample_path, psa_data_params, ingest_with_uuid):
+    def __init__(self, sample_path, psa_data_params, ingest_with_uuid: bool):
         self.sample_path = sample_path
         self.sample_name = os.path.basename(sample_path)
         self.metadata = self._parse_meta(psa_data_params)
@@ -105,7 +105,7 @@ def tokenize(self, conf_name):
                     raw_event[event_counter] = each_rule.apply(raw_event[event_counter])
             for event in raw_event[event_counter]:
                 # Assign UUID per finalized event to ensure uniqueness across variants
-                if event.metadata.get("ingest_with_uuid") == "true":
+                if event.metadata.get("ingest_with_uuid"):
                     event.unique_identifier = str(uuid.uuid4())
 
                 host_value = event.metadata.get("host")

pytest_splunk_addon/sample_generation/sample_xdist_generator.py

@@ -33,7 +33,7 @@ class SampleXdistGenerator:
         process_count (num): generate {no} process for execution
     """
 
-    def __init__(self, addon_path, ingest_with_uuid, config_path=None, process_count=4):
+    def __init__(self, addon_path, ingest_with_uuid: bool, config_path=None, process_count=4):
         self.addon_path = addon_path
         self.process_count = process_count
         self.config_path = config_path
@@ -141,7 +141,7 @@ def store_events(self, tokenized_events):
                         }
                     ],
                 }
-                if self.ingest_with_uuid == "true":
+                if self.ingest_with_uuid:
                     tokenized_samples_dict[each_event.sample_name]["events"][0][
                         "unique_identifier"
                     ] = each_event.unique_identifier
@@ -152,7 +152,7 @@ def store_events(self, tokenized_events):
                     "time_values": each_event.time_values,
                     "requirement_test_data": each_event.requirement_test_data,
                 }
-                if self.ingest_with_uuid == "true":
+                if self.ingest_with_uuid:
                     sample_event["unique_identifier"] = each_event.unique_identifier
                 tokenized_samples_dict[each_event.sample_name]["events"].append(
                     sample_event

pytest_splunk_addon/splunk.py

@@ -49,13 +49,11 @@ def pytest_addoption(parser):
     """
     group = parser.getgroup("splunk-addon")
     group.addoption(
-        "--ingest-with-uuid",
-        action="store",
+        "--use-uuid",
+        action="store_true",
         dest="ingest_with_uuid",
-        default="false",
-        help=(
-            'Use generated UUID for ingesting and searching events. Setting this parameter to "true" will lead to matching events in search by the ID and not by escaped _raw. Default is "false".'
-        ),
+        default=False,
+        help="Use generated UUID for ingesting and searching events. Setting this flag will lead to matching events in search by the ID and not by escaped _raw.",
     )
 
     group.addoption(

tests/e2e/test_splunk_addon.py

@@ -845,7 +845,7 @@ def empty_method():
         "--search-interval=4",
         "--search-retry=4",
         "--search-index=*,_internal",
-        "--ingest-with-uuid=true",
+        "--use-uuid=true",
     )
 
     logger.info(result.outlines)
@@ -899,7 +899,7 @@ def empty_method():
         "--search-retry=4",
         "--search-index=*",
         "--splunk-data-generator=tests/addons/TA_transition_from_req/default",
-        "--ingest-with-uuid=true",
+        "--use-uuid=true",
     )
     logger.info(result.outlines)
 

tests/unit/tests_standard_lib/test_app_test_generator.py

@@ -11,7 +11,7 @@
     "store_events": True,
     "splunk_data_generator": "psa.conf",
     "requirement_test": "fake_requirement_path",
-    "ingest_with_uuid": "false",
+    "ingest_with_uuid": False,
 }
 pytest_config = namedtuple("Config", ["getoption"])
 test_config = pytest_config(getoption=lambda x, *y: config[x])
@@ -97,7 +97,7 @@ def test_app_test_generator_instantiation(
                 "app_path": "fake_app",
                 "config_path": "psa.conf",
                 "test_type": "key_fields",
-                "ingest_with_uuid": "false",
+                "ingest_with_uuid": False,
             },
             [
                 params(values=f"splunk_indextime_key_fields_test_1", id=1),
@@ -118,7 +118,7 @@ def test_app_test_generator_instantiation(
                 "app_path": "fake_app",
                 "config_path": "psa.conf",
                 "test_type": "_time",
-                "ingest_with_uuid": "false",
+                "ingest_with_uuid": False,
             },
             [
                 params(values=f"splunk_indextime__time_test_1", id=1),
@@ -139,7 +139,7 @@ def test_app_test_generator_instantiation(
                 "app_path": "fake_app",
                 "config_path": "psa.conf",
                 "test_type": "line_breaker",
-                "ingest_with_uuid": "false",
+                "ingest_with_uuid": False,
             },
             [
                 params(values=f"splunk_indextime_line_breaker_test_1", id=1),