feat: Added options to generate ingest events and run test

Author: rajkarkar-crest

pytest_splunk_addon/plugin.py

@@ -1,6 +1,6 @@
 import logging
 import pytest
-import sys
+from .standard_lib.sample_generation.sample_xdist_generator import SampleXdistGenerator
 import traceback
 from .standard_lib import AppTestGenerator
 from .standard_lib.cim_compliance import CIMReportPlugin
@@ -78,11 +78,32 @@ def pytest_unconfigure(config):
         del config._markdown
         config.pluginmanager.unregister(markdown)
 
+def pytest_sessionstart(session):
+
+    SampleXdistGenerator.event_path = session.config.getoption("event_path")
+    SampleXdistGenerator.event_stored = False
+    SampleXdistGenerator.tokenized_event_source = session.config.getoption("tokenized_event_source").lower()
+    # For scenario (store_new, False, False) get_samples from SampleXdistGenerator will not be invoked
+    # Because all teh tests are skipped, So event.pickle file will not be generated
+    # For this scenarios tokenized_event is generated here
+    if (
+        SampleXdistGenerator.tokenized_event_source == "store_new"
+        and session.config.getoption("ingest_events").lower() in ["no", "n", "false", "f"]
+        and session.config.getoption("execute_test").lower() in ["no", "n", "false", "f"] 
+    ):
+        app_path = session.config.getoption("splunk_app")
+        config_path = session.config.getoption("splunk_data_generator")
+        store_events = session.config.getoption("store_events")
+        sample_generator = SampleXdistGenerator(app_path, config_path)
+        sample_generator.get_samples(store_events)
+
 
 def pytest_generate_tests(metafunc):
     """
     Parse the fixture dynamically.
     """
+    if metafunc.config.getoption("execute_test").lower() in ["no","n","false","f"]:
+        return
     global test_generator
     for fixture in metafunc.fixturenames:
         if fixture.startswith("splunk_searchtime") or fixture.startswith(
@@ -111,6 +132,13 @@ def pytest_generate_tests(metafunc):
                     f"\nLogs:\n{log_message}"
                 )
 
+def pytest_collection_modifyitems(config, items):
+    ingest_events_flag = config.getoption("ingest_events")
+    is_ingest_false = ingest_events_flag.lower() in ["no","n","false","f"]
+    execute_test_flag = config.getoption("execute_test")
+    if execute_test_flag.lower() in ["no","n","false","f"]:
+        for item in items.copy():
+            item.add_marker(pytest.mark.skipif(item.name != 'test_events_with_untokenised_values' or is_ingest_false, reason=f'--execute-test={execute_test_flag} provided'))
 
 def init_pytest_splunk_addon_logger():
     """
@@ -128,6 +156,5 @@ def init_pytest_splunk_addon_logger():
     logger.setLevel(logging.INFO)
     return logger
 
-
 init_pytest_splunk_addon_logger()
 LOGGER = logging.getLogger("pytest-splunk-addon")

pytest_splunk_addon/splunk.py

@@ -286,7 +286,34 @@ def pytest_addoption(parser):
         default="Chang3d!",
         help="Password of the Universal Forwarder user",
     )
-
+    group.addoption(
+        "--event-file-path",
+        action="store",
+        dest="event_path",
+        help="Path to tokenised event directory",
+        default="events.pickel"
+    )
+    group.addoption(
+        "--tokenized-event-source",
+        action="store",
+        dest="tokenized_event_source",
+        help="One of (new|pregenerated|store_new)",
+        default="new"
+    )
+    group.addoption(
+        "--ingest-events",
+        action="store",
+        dest="ingest_events",
+        help="Should ingest events or not (True|False)",
+        default="True"
+    )
+    group.addoption(
+        "--execute-test",
+        action="store",
+        dest="execute_test",
+        help="Should ingest events or not (True|False)",
+        default="True"
+    )
 
 @pytest.fixture(scope="session")
 def splunk_setup(splunk):
@@ -666,6 +693,8 @@ def splunk_ingest_data(request, splunk_hec_uri, sc4s, uf, splunk_events_cleanup)
     TODO:
         For splunk_type=external, data will not be ingested as manual configurations are required.
     """
+    if request.config.getoption("ingest_events").lower() in ['n','no','false','f']:
+        return
     global PYTEST_XDIST_TESTRUNUID
     if (
         "PYTEST_XDIST_WORKER" not in os.environ

pytest_splunk_addon/standard_lib/sample_generation/sample_xdist_generator.py

@@ -4,6 +4,7 @@
 import pickle
 from filelock import FileLock
 import json
+import pytest
 
 class SampleXdistGenerator():
 
@@ -13,6 +14,21 @@ def __init__(self, addon_path, config_path=None, process_count=4):
         self.config_path = config_path
 
     def get_samples(self, store_events):
+
+        if self.tokenized_event_source == 'pregenerated':
+            with open(self.event_path,"rb") as file_obj:
+                store_sample = pickle.load(file_obj)
+                if store_events and (
+                    "PYTEST_XDIST_WORKER" not in os.environ 
+                    or os.environ.get("PYTEST_XDIST_WORKER") == "gw0"
+                ):
+                    try:
+                        tokenized_events = store_sample.get("tokenized_events")
+                        self.store_events(tokenized_events)
+                    except Exception as e:
+                        pytest.exit(str(e))
+                return store_sample
+        
         if "PYTEST_XDIST_WORKER" in os.environ:
             file_path = os.environ.get("PYTEST_XDIST_TESTRUNUID") + "_events"
             with FileLock(str(file_path) + ".lock"):
@@ -35,6 +51,10 @@ def get_samples(self, store_events):
             store_sample = {"conf_name" : SampleGenerator.conf_name, "tokenized_events" : tokenized_events}
             if store_events:
                 self.store_events(tokenized_events)
+        if self.tokenized_event_source == "store_new" and not self.event_stored:
+            with open(self.event_path, "wb") as file_obj:
+                pickle.dump(store_sample, file_obj)
+            self.event_stored = True
         return store_sample
 
     def store_events(self, tokenized_events):