test using sythetic data

Author: Cecylia Borek

tests/searchcommands/test_csc_apps.py

@@ -69,28 +69,20 @@ def test_eventing_app(self):
 
         jobs = self.service.jobs
 
-        test_index_stream = jobs.oneshot(
-            'search index="_internal" | head 10', output_mode='json'
+        base_search = self._create_test_data_search(count=100)
+        full_search = (
+            base_search
+            + '''
+        | eventingcsc status=200
+        | head 10
+        '''
         )
-        test_reader = results.JSONResultsReader(test_index_stream)
-        test_items = list(test_reader)
 
-        print("DEBUG: Test items from _internal index stream:")
-        for item in test_items:
-            print(item)
+        stream = jobs.oneshot(full_search, output_mode='json')
 
-        stream = jobs.oneshot(
-            'search index="_internal" | fields status | head 4000 | eventingcsc status=200 | head 10',
-            output_mode='json',
-        )
-       
         reader = results.JSONResultsReader(stream)
         items = list(reader)
 
-        print("DEBUG: Items from eventingcsc command:")
-        for item in items:
-            print(item)
-
         actual_results = [item for item in items if isinstance(item, dict)]
         informational_messages = [
             item for item in items if isinstance(item, results.Message)
@@ -332,6 +324,22 @@ def test_streaming_app(self):
         self.assertTrue(ds[0]["fahrenheit"] == "95.0")
         self.assertTrue(len(ds) == 5)
 
+    def _create_test_data_search(self, count=100):
+        """Helper to create deterministic test data using Splunk search commands."""
+        return f'''
+        | makeresults count={count}
+        | streamstats count as row_num
+        | eval _time=_time - (row_num * 60)
+        | eval status=case(
+            (row_num % 10) < 7, 200,
+            (row_num % 10) < 9, 404,
+            1=1, 500
+        )
+        | eval response_time=100 + ((row_num * 37) % 1000)
+        | eval user_id="user" + tostring(row_num % 50)
+        | eval _raw=strftime(_time, "%Y-%m-%d %H:%M:%S") + " status=" + tostring(status) + " response_time=" + tostring(response_time) + "ms user=" + user_id
+        '''
+
 
 if __name__ == "__main__":
     unittest.main()