chore: modify uuid approach

Author: mkolasinski-splunk

pytest_splunk_addon/fields_tests/test_generator.py

@@ -264,27 +264,20 @@ def generate_requirements_tests(self):
                     for field, value in requirement_fields.items()
                     if field not in exceptions
                 }
-                unique_identifier = (
-                    getattr(event, "unique_identifier", None)
-                    if metadata.get("ingest_with_uuid") == "true"
-                    else None
-                )
-                variant_id = metadata.get("variant_id")
-                search_selector = (
-                    {
-                        "unique_identifier": unique_identifier,
-                        "escaped_event": escaped_event,
-                        "variant_id": variant_id,
-                    }
-                    if unique_identifier is not None
-                    else {"escaped_event": escaped_event, "variant_id": variant_id}
-                )
+                sample_event = {
+                    "escaped_event": escaped_event,
+                    "fields": requirement_fields,
+                    "modinput_params": modinput_params,
+                }
+
+                # Add unique_identifier if UUID ingestion is enabled and event has UUID
+                if metadata.get("ingest_with_uuid") == "true":
+                    unique_identifier = getattr(event, "unique_identifier", None)
+                    if unique_identifier is not None:
+                        sample_event["unique_identifier"] = unique_identifier
+
                 yield pytest.param(
-                    {
-                        **search_selector,
-                        "fields": requirement_fields,
-                        "modinput_params": modinput_params,
-                    },
+                    sample_event,
                     id=f"sample_name::{event.sample_name}::host::{event.metadata.get('host')}",
                 )
 

pytest_splunk_addon/fields_tests/test_templates.py

@@ -162,19 +162,20 @@ def test_requirements_fields(
         """
 
         # Search Query
-        record_property(
-            "Event_with", splunk_searchtime_fields_requirements["escaped_event"]
+        unique_identifier = splunk_searchtime_fields_requirements.get(
+            "unique_identifier"
         )
+        escaped_event = splunk_searchtime_fields_requirements.get("escaped_event")
+        event_identifier = (
+            unique_identifier if unique_identifier is not None else escaped_event
+        )
+
+        record_property("Event_with", event_identifier)
         record_property("fields", splunk_searchtime_fields_requirements["fields"])
         record_property(
             "modinput_params", splunk_searchtime_fields_requirements["modinput_params"]
         )
 
-        escaped_event = splunk_searchtime_fields_requirements.get("escaped_event")
-        unique_identifier = splunk_searchtime_fields_requirements.get(
-            "unique_identifier"
-        )
-        variant_id = splunk_searchtime_fields_requirements.get("variant_id")
         fields = splunk_searchtime_fields_requirements["fields"]
         modinput_params = splunk_searchtime_fields_requirements["modinput_params"]
 
@@ -190,16 +191,13 @@ def test_requirements_fields(
                 basic_search += f" {param}={param_value}"
 
         if unique_identifier is not None:
-            selector = f'fields.unique_identifier="{unique_identifier}"'
+            selector = f'unique_identifier="{unique_identifier}"'
         elif escaped_event is not None:
             selector = escaped_event
         else:
             selector = ""
-        variant_clause = f" variant_id={variant_id}" if variant_id is not None else ""
 
-        search = (
-            f"search {index_list} {basic_search} {selector}{variant_clause} | fields *"
-        )
+        search = f"search {index_list} {basic_search} {selector} | fields *"
 
         self.logger.info(f"Executing the search query: {search}")
 
@@ -241,8 +239,7 @@ def test_requirements_fields(
             f"{format_search_query_log(search)}"
         )
 
-        if splunk_searchtime_fields_requirements.get("unique_identifier"):
-            error_message += f"Test failed for event: {escaped_event}\n"
+        error_message += f"Test failed for event: {event_identifier}\n"
 
         assert wrong_value_fields == {}, error_message
 

pytest_splunk_addon/sample_generation/sample_event.py

@@ -68,8 +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.get("ingest_with_uuid") == "true":
-            self.unique_identifier = str(uuid.uuid4())
+        # 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

@@ -16,6 +16,7 @@
 import os
 import re
 import copy
+import uuid
 from . import Rule
 from . import raise_warning
 from . import SampleEvent
@@ -103,6 +104,10 @@ def tokenize(self, conf_name):
                 if each_rule:
                     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":
+                    event.unique_identifier = str(uuid.uuid4())
+
                 host_value = event.metadata.get("host")
                 host = token_value(key=host_value, value=host_value)
                 event.update_requirement_test_field("host", "##host##", host)
@@ -267,9 +272,6 @@ def get_eventmetadata(self):
         self.host_count += 1
         event_host = self.metadata.get("host") + "_" + str(self.host_count)
         event_metadata = copy.deepcopy(self.metadata)
-        # Add variant_id only when UUID ingestion is enabled
-        if event_metadata.get("ingest_with_uuid") == "true":
-            event_metadata.update(variant_id=self.host_count)
         event_metadata.update(host=event_host)
         LOGGER.info("event metadata: {}".format(event_metadata))
         return event_metadata

tests/unit/tests_standard_lib/test_fields_tests/test_test_generator.py

@@ -486,7 +486,6 @@ def test_generate_field_tests(
                 (
                     {
                         "escaped_event": "escaped_event",
-                        "variant_id": None,
                         "fields": {
                             "severity": "low",
                             "signature_id": "405001",
@@ -502,7 +501,6 @@ def test_generate_field_tests(
                 (
                     {
                         "escaped_event": "escaped_event",
-                        "variant_id": None,
                         "fields": {
                             "src": "192.168.0.1",
                             "type": "event",
@@ -537,34 +535,35 @@ def test_generate_requirement_tests(tokenised_events, expected_output):
 
 
 def test_generate_requirement_tests_with_uuid(mock_uuid4):
-    tokenised_events = [
-        SampleEvent(
-            event_string="escaped_event",
-            metadata={
-                "input_type": "modinput",
-                "sourcetype_to_search": "dummy_sourcetype",
-                "host": "dummy_host",
-                "ingest_with_uuid": "true",
-                "unique_identifier": "uuid",
-            },
-            sample_name="file1.xml",
-            requirement_test_data={
-                "cim_fields": {
-                    "severity": "low",
-                    "signature_id": "405001",
-                    "src": "192.168.0.1",
-                    "type": "event",
-                },
+    event = SampleEvent(
+        event_string="escaped_event",
+        metadata={
+            "input_type": "modinput",
+            "sourcetype_to_search": "dummy_sourcetype",
+            "host": "dummy_host",
+            "ingest_with_uuid": "true",
+        },
+        sample_name="file1.xml",
+        requirement_test_data={
+            "cim_fields": {
+                "severity": "low",
+                "signature_id": "405001",
+                "src": "192.168.0.1",
+                "type": "event",
             },
-        )
-    ]
+        },
+    )
+
+    # Simulate tokenization UUID assignment
+    event.unique_identifier = "uuid"
+
+    tokenised_events = [event]
 
     expected_output = [
         (
             {
                 "escaped_event": "escaped_event",
                 "unique_identifier": "uuid",
-                "variant_id": None,
                 "fields": {
                     "severity": "low",
                     "signature_id": "405001",
@@ -668,19 +667,22 @@ def test_generate_requirement_datamodel_tests(tokenised_events, expected_output)
 
 
 def test_generate_requirement_datamodel_tests_with_uuid(mock_uuid4):
-    tokenised_events = [
-        SampleEvent(
-            event_string="escaped_event",
-            metadata={
-                "input_type": "modinput",
-                "sourcetype_to_search": "dummy_sourcetype",
-                "host": "dummy_host",
-                "ingest_with_uuid": "true",
-            },
-            sample_name="file1.xml",
-            requirement_test_data={"datamodels": {"model": "Alerts"}},
-        )
-    ]
+    event = SampleEvent(
+        event_string="escaped_event",
+        metadata={
+            "input_type": "modinput",
+            "sourcetype_to_search": "dummy_sourcetype",
+            "host": "dummy_host",
+            "ingest_with_uuid": "true",
+        },
+        sample_name="file1.xml",
+        requirement_test_data={"datamodels": {"model": "Alerts"}},
+    )
+
+    # Simulate tokenization UUID assignment
+    event.unique_identifier = "uuid"
+
+    tokenised_events = [event]
 
     expected_output = [
         (

tests/unit/tests_standard_lib/tests_sample_generation/test_sample_event.py

@@ -35,17 +35,23 @@ def samp_eve():
 
 def test_sample_event_generates_uuid():
     METADATA["ingest_with_uuid"] = "true"
-    with patch(
-        "pytest_splunk_addon.sample_generation.sample_event.uuid.uuid4",
-        return_value="uuid",
-    ) as mock_uuid:
-        event = pytest_splunk_addon.sample_generation.sample_event.SampleEvent(
-            event_string=EVENT_STRING,
-            metadata=METADATA,
-            sample_name=SAMPLE_NAME,
-        )
-
-        mock_uuid.assert_called_once()  # Ensures uuid4 was called
+    event = pytest_splunk_addon.sample_generation.sample_event.SampleEvent(
+        event_string=EVENT_STRING,
+        metadata=METADATA,
+        sample_name=SAMPLE_NAME,
+    )
+
+    # UUID should not be generated during SampleEvent creation anymore
+    assert not hasattr(
+        event, "unique_identifier"
+    ), "UUID should not be assigned during SampleEvent creation"
+
+    # Simulate tokenization where UUID is assigned
+    with patch("uuid.uuid4", return_value="uuid") as mock_uuid:
+        if event.metadata.get("ingest_with_uuid") == "true":
+            event.unique_identifier = str(mock_uuid())
+
+        mock_uuid.assert_called_once()  # Ensures uuid4 was called during tokenization simulation
         assert hasattr(event, "unique_identifier")  # The field was set
         assert event.unique_identifier == "uuid"