chore: adjusting host variants handling

Author: mkolasinski-splunk

pytest_splunk_addon/fields_tests/test_generator.py

@@ -264,15 +264,29 @@ def generate_requirements_tests(self):
                     for field, value in requirement_fields.items()
                     if field not in exceptions
                 }
-                sample_event = {
-                    "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
+                # Prefer UUID if present and include variant_id for disambiguation
+                # Fetch UUID from the SampleEvent when ingest_with_uuid is enabled
+                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}
+                )
                 yield pytest.param(
-                    sample_event,
+                    {
+                        **search_selector,
+                        "fields": requirement_fields,
+                        "modinput_params": modinput_params,
+                    },
                     id=f"sample_name::{event.sample_name}::host::{event.metadata.get('host')}",
                 )
 

pytest_splunk_addon/fields_tests/test_templates.py

@@ -170,7 +170,11 @@ def test_requirements_fields(
             "modinput_params", splunk_searchtime_fields_requirements["modinput_params"]
         )
 
-        escaped_event = splunk_searchtime_fields_requirements["escaped_event"]
+        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"]
 
@@ -185,15 +189,18 @@ def test_requirements_fields(
             if param_value is not None:
                 basic_search += f" {param}={param_value}"
 
-        if splunk_searchtime_fields_requirements.get("unique_identifier"):
-            unique_identifier = splunk_searchtime_fields_requirements[
-                "unique_identifier"
-            ]
-            record_property("Event_with", unique_identifier)
-
-            search = f'search {index_list} {basic_search} unique_identifier="{unique_identifier}" | fields *'
+        # Prefer UUID+variant_id, fallback to escaped_event+variant_id
+        if unique_identifier is not None:
+            selector = f'fields.unique_identifier="{unique_identifier}"'
+        elif escaped_event is not None:
+            selector = escaped_event
         else:
-            search = f"search {index_list} {basic_search} {escaped_event} | fields *"
+            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 *"
+        )
 
         self.logger.info(f"Executing the search query: {search}")
 

pytest_splunk_addon/sample_generation/sample_stanza.py

@@ -267,6 +267,9 @@ 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
@@ -306,8 +309,7 @@ def _get_raw_sample(self):
                 if "transport" in each_event.keys():
                     static_host = each_event["transport"].get("@host")
                     if static_host:
-                        # Preserve per-event uniqueness by appending variant counter
-                        event_metadata.update(host=f"{static_host}-{self.host_count}")
+                        event_metadata.update(host=static_host)
                     static_source = each_event["transport"].get("@source")
                     if static_source:
                         event_metadata.update(source=static_source)

tests/unit/tests_standard_lib/test_fields_tests/test_test_generator.py

@@ -486,6 +486,7 @@ def test_generate_field_tests(
                 (
                     {
                         "escaped_event": "escaped_event",
+                        "variant_id": None,
                         "fields": {
                             "severity": "low",
                             "signature_id": "405001",
@@ -501,6 +502,7 @@ def test_generate_field_tests(
                 (
                     {
                         "escaped_event": "escaped_event",
+                        "variant_id": None,
                         "fields": {
                             "src": "192.168.0.1",
                             "type": "event",
@@ -543,6 +545,7 @@ def test_generate_requirement_tests_with_uuid(mock_uuid4):
                 "sourcetype_to_search": "dummy_sourcetype",
                 "host": "dummy_host",
                 "ingest_with_uuid": "true",
+                "unique_identifier": "uuid",
             },
             sample_name="file1.xml",
             requirement_test_data={
@@ -561,6 +564,7 @@ def test_generate_requirement_tests_with_uuid(mock_uuid4):
             {
                 "escaped_event": "escaped_event",
                 "unique_identifier": "uuid",
+                "variant_id": None,
                 "fields": {
                     "severity": "low",
                     "signature_id": "405001",