chore: ADDON-80802 Updated the working of test_datamodels test using uuid

Author: spanchal-crest

pytest_splunk_addon/event_ingestors/hec_event_ingestor.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
+import json
 from .base_event_ingestor import EventIngestor
 import requests
 from time import time, mktime
@@ -117,20 +118,21 @@ def ingest(self, events, thread_count):
 
     def __ingest(self, data):
         try:
+            batch_data = "\n".join(json.dumps(obj) for obj in data)
             LOGGER.info(
                 "Making a HEC event request with the following params:\nhec_uri:{}\nheaders:{}".format(
                     str(self.hec_uri), str(self.session_headers)
                 )
             )
             LOGGER.debug(
                 "Creating the following sample event to be ingested via HEC event endoipnt:{}".format(
-                    str(data)
+                    str(batch_data)
                 )
             )
             response = requests.post(  # nosemgrep: splunk.disabled-cert-validation
                 "{}/{}".format(self.hec_uri, "event"),
                 auth=None,
-                json=data,
+                data=batch_data,
                 headers=self.session_headers,
                 verify=False,
             )

pytest_splunk_addon/fields_tests/test_generator.py

@@ -190,11 +190,14 @@ def generate_requirements_datamodels_tests(self):
                 datamodel.replace(" ", "_").replace(":", "_")
                 for datamodel in datamodels
             ]
-            yield pytest.param(
-                {
+            sample_event = {
                     "datamodels": datamodels,
                     "stanza": escaped_event,
-                },
+                }
+            if event.metadata["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')}",
             )
 

pytest_splunk_addon/fields_tests/test_templates.py

@@ -401,21 +401,34 @@ def test_datamodels(
             record_property (fixture): Document facts of test cases to provide more info in the test failure reports.
             caplog (fixture): fixture to capture logs.
         """
-        esacaped_event = splunk_searchtime_fields_datamodels["stanza"]
+        escaped_event = splunk_searchtime_fields_datamodels["stanza"]
         datamodels = splunk_searchtime_fields_datamodels["datamodels"]
-        self.logger.info(
-            f"Testing for tag {datamodels} with tag_query {esacaped_event}"
-        )
 
-        record_property("Event_with", esacaped_event)
+        record_property("Event_with", escaped_event)
         record_property("datamodels", datamodels)
 
         index_list = (
             "(index="
             + " OR index=".join(splunk_search_util.search_index.split(","))
             + ")"
         )
-        search = f"search {index_list} {esacaped_event} | fields *"
+
+        if splunk_searchtime_fields_datamodels.get("unique_identifier"):
+            record_property(
+            "stanza_name", 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 *"
+        else:
+            self.logger.info(
+            f"Testing for tag {datamodels} with tag_query {escaped_event}"
+        )
+            search = f"search {index_list} {escaped_event} | fields *"
 
         self.logger.info(f"Search: {search}")