chore: ADDON-80802 fixing linting failures

Author: spanchal-crest

pytest_splunk_addon/cim_tests/test_generator.py

@@ -53,7 +53,6 @@ def __init__(
         test_field_type=["required", "conditional"],
         common_fields_path=None,
     ):
-
         self.data_model_handler = DataModelHandler(data_model_path)
         self.addon_parser = AddonParser(addon_path)
         self.tokenized_events = tokenized_events

pytest_splunk_addon/event_ingestors/hec_event_ingestor.py

@@ -87,14 +87,13 @@ def ingest(self, events, thread_count):
         """
         data = list()
         for event in events:
-
             event_dict = {
                 "sourcetype": event.metadata.get("sourcetype", "pytest_splunk_addon"),
                 "source": event.metadata.get("source", "pytest_splunk_addon:hec:event"),
                 "event": event.event,
                 "index": event.metadata.get("index", "main"),
             }
-            if event.metadata["ingest_with_uuid"] == "true":
+            if event.metadata.get("ingest_with_uuid") == "true":
                 event_dict["fields"] = {"unique_identifier": event.unique_identifier}
 
             if event.metadata.get("host_type") in ("plugin", None):

pytest_splunk_addon/event_ingestors/ingestor_helper.py

@@ -95,7 +95,9 @@ def ingest_events(
             thread_count (int): number of threads to use for ingestion
             store_events (bool): Boolean param for generating json files with tokenised events
         """
-        sample_generator = SampleXdistGenerator(addon_path, ingest_meta_data["ingest_with_uuid"], config_path)
+        sample_generator = SampleXdistGenerator(
+            addon_path, ingest_meta_data["ingest_with_uuid"], config_path
+        )
         store_sample = sample_generator.get_samples(store_events)
         tokenized_events = store_sample.get("tokenized_events")
         ingestor_dict = cls.get_consolidated_events(tokenized_events)

pytest_splunk_addon/fields_tests/test_generator.py

@@ -191,11 +191,11 @@ def generate_requirements_datamodels_tests(self):
                 for datamodel in datamodels
             ]
             sample_event = {
-                    "datamodels": datamodels,
-                    "stanza": escaped_event,
-                }
-            if event.metadata["ingest_with_uuid"] == "true":
-                    sample_event["unique_identifier"] = event.unique_identifier
+                "datamodels": datamodels,
+                "stanza": escaped_event,
+            }
+            if event.metadata.get("ingest_with_uuid") == "true":
+                sample_event["unique_identifier"] = event.unique_identifier
             yield pytest.param(
                 sample_event,
                 id=f"{'-'.join(datamodels)}::sample_name::{event.sample_name}::host::{event.metadata.get('host')}",
@@ -265,11 +265,11 @@ def generate_requirements_tests(self):
                     if field not in exceptions
                 }
                 sample_event = {
-                        "escaped_event": escaped_event,
-                        "fields": requirement_fields,
-                        "modinput_params": modinput_params,
-                    }
-                if metadata["ingest_with_uuid"] == "true":
+                    "escaped_event": escaped_event,
+                    "fields": requirement_fields,
+                    "modinput_params": modinput_params,
+                }
+                if metadata.get("ingest_with_uuid") == "true":
                     sample_event["unique_identifier"] = event.unique_identifier
                 yield pytest.param(
                     sample_event,

pytest_splunk_addon/fields_tests/test_templates.py

@@ -163,7 +163,7 @@ def test_requirements_fields(
 
         # Search Query
         record_property(
-            "stanza_name", splunk_searchtime_fields_requirements["escaped_event"]
+            "Event_with", splunk_searchtime_fields_requirements["escaped_event"]
         )
         record_property("fields", splunk_searchtime_fields_requirements["fields"])
         record_property(
@@ -187,11 +187,13 @@ def test_requirements_fields(
 
         if splunk_searchtime_fields_requirements.get("unique_identifier"):
             record_property(
-            "stanza_name", splunk_searchtime_fields_requirements["unique_identifier"]
-        )
-            unique_identifier = splunk_searchtime_fields_requirements["unique_identifier"]
-            
-            search = f"search {index_list} {basic_search} unique_identifier=\"{unique_identifier}\" | fields *"
+                "Event_with", splunk_searchtime_fields_requirements["unique_identifier"]
+            )
+            unique_identifier = splunk_searchtime_fields_requirements[
+                "unique_identifier"
+            ]
+
+            search = f'search {index_list} {basic_search} unique_identifier="{unique_identifier}" | fields *'
         else:
             search = f"search {index_list} {basic_search} {escaped_event} | fields *"
 
@@ -230,12 +232,16 @@ def test_requirements_fields(
         if not wrong_value_fields == {}:
             self.logger.error("Wrong field values:\n" + wrong_values_table)
 
-        assert wrong_value_fields == {}, (
+        error_message = (
             f"\nNot all required fields have correct values or some fields are missing in Splunk. Wrong field values:\n{wrong_values_table}"
             f"{format_search_query_log(search)}"
-            f"Test failed for event: {escaped_event}\n"
         )
 
+        if splunk_searchtime_fields_requirements.get("unique_identifier"):
+            error_message += f"Test failed for event: {escaped_event}\n"
+
+        assert wrong_value_fields == {}, error_message
+
     @pytest.mark.splunk_searchtime_fields
     @pytest.mark.splunk_searchtime_fields_negative
     def test_props_fields_no_dash_not_empty(
@@ -415,19 +421,19 @@ def test_datamodels(
 
         if splunk_searchtime_fields_datamodels.get("unique_identifier"):
             record_property(
-            "stanza_name", splunk_searchtime_fields_datamodels["unique_identifier"]
-        )
+                "Event_with", splunk_searchtime_fields_datamodels["unique_identifier"]
+            )
             unique_identifier = splunk_searchtime_fields_datamodels["unique_identifier"]
 
             self.logger.info(
-            f"Testing for tag {datamodels} with unique_identifier=\"{unique_identifier}\""
-        )
-            
-            search = f"search {index_list} unique_identifier=\"{unique_identifier}\" | fields *"
+                f'Testing for tag {datamodels} with unique_identifier="{unique_identifier}"'
+            )
+
+            search = f'search {index_list} unique_identifier="{unique_identifier}" | fields *'
         else:
             self.logger.info(
-            f"Testing for tag {datamodels} with tag_query {escaped_event}"
-        )
+                f"Testing for tag {datamodels} with tag_query {escaped_event}"
+            )
             search = f"search {index_list} {escaped_event} | fields *"
 
         self.logger.info(f"Search: {search}")
@@ -487,9 +493,12 @@ def test_datamodels(
             ],
         )
 
-        assert (
-            missing_datamodels == [] and wrong_datamodels == []
-        ), f"Incorrect datamodels found:\n{exc_message}"
+        error_message = f"Incorrect datamodels found:\n{exc_message}"
+
+        if splunk_searchtime_fields_datamodels.get("unique_identifier"):
+            error_message += f"\nTest failed for event: {escaped_event}\n"
+
+        assert missing_datamodels == [] and wrong_datamodels == [], error_message
 
     @pytest.mark.splunk_searchtime_fields
     @pytest.mark.splunk_searchtime_fields_eventtypes

pytest_splunk_addon/sample_generation/pytest_splunk_addon_data_parser.py

@@ -107,7 +107,9 @@ def get_sample_stanzas(self):
         results = []
         for sample_name, stanza_params in sorted(_psa_data.items()):
             sample_path = os.path.join(self._path_to_samples, sample_name)
-            results.append(SampleStanza(sample_path, stanza_params, self.ingest_with_uuid))
+            results.append(
+                SampleStanza(sample_path, stanza_params, self.ingest_with_uuid)
+            )
         return results
 
     def _get_psa_data_stanzas(self):

pytest_splunk_addon/sample_generation/sample_event.py

@@ -68,7 +68,7 @@ def __init__(self, event_string, metadata, sample_name, requirement_test_data=No
         self.time_values = list()
         self.metadata = metadata
         self.sample_name = sample_name
-        if metadata["ingest_with_uuid"] == "true":
+        if metadata.get("ingest_with_uuid") == "true":
             self.unique_identifier = str(uuid.uuid4())
         self.host_count = 0
         self.requirement_test_data = requirement_test_data

pytest_splunk_addon/sample_generation/sample_generator.py

@@ -45,7 +45,9 @@ def get_samples(self):
         """
         if not SampleGenerator.sample_stanzas:
             psa_data_parser = PytestSplunkAddonDataParser(
-                self.addon_path, config_path=self.config_path, ingest_with_uuid=self.ingest_with_uuid
+                self.addon_path,
+                config_path=self.config_path,
+                ingest_with_uuid=self.ingest_with_uuid,
             )
             sample_stanzas = psa_data_parser.get_sample_stanzas()
             SampleGenerator.conf_name = psa_data_parser.conf_name

pytest_splunk_addon/sample_generation/sample_xdist_generator.py

@@ -80,7 +80,9 @@ def get_samples(self, store_events):
                     with open(file_path, "wb") as file_obj:
                         pickle.dump(store_sample, file_obj)
         else:
-            sample_generator = SampleGenerator(self.addon_path, self.ingest_with_uuid, self.config_path)
+            sample_generator = SampleGenerator(
+                self.addon_path, self.ingest_with_uuid, self.config_path
+            )
             tokenized_events = list(sample_generator.get_samples())
             store_sample = {
                 "conf_name": SampleGenerator.conf_name,
@@ -140,14 +142,16 @@ def store_events(self, tokenized_events):
                     ],
                 }
                 if self.ingest_with_uuid == "true":
-                    tokenized_samples_dict[each_event.sample_name]["events"][0]["unique_identifier"] = each_event.unique_identifier
+                    tokenized_samples_dict[each_event.sample_name]["events"][0][
+                        "unique_identifier"
+                    ] = each_event.unique_identifier
             else:
                 sample_event = {
-                        "event": each_event.event,
-                        "key_fields": each_event.key_fields,
-                        "time_values": each_event.time_values,
-                        "requirement_test_data": each_event.requirement_test_data,
-                    }
+                    "event": each_event.event,
+                    "key_fields": each_event.key_fields,
+                    "time_values": each_event.time_values,
+                    "requirement_test_data": each_event.requirement_test_data,
+                }
                 if self.ingest_with_uuid == "true":
                     sample_event["unique_identifier"] = each_event.unique_identifier
                 tokenized_samples_dict[each_event.sample_name]["events"].append(

pytest_splunk_addon/splunk.py

@@ -52,10 +52,9 @@ def pytest_addoption(parser):
         "--ingest-with-uuid",
         action="store",
         dest="ingest_with_uuid",
-        default="False",
+        default="false",
         help=(
-            "Type of ingesting and searching the events into Splunk "
-            "with uuid or without uuid."
+            '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".'
         ),
     )