fix csc apps test

Author: Cecylia Borek

tests/searchcommands/test_csc_apps.py

@@ -64,17 +64,41 @@ def test_eventing_app(self):
         self.assertEqual(state.title, "eventing_app")
 
         jobs = self.service.jobs
-        stream = jobs.oneshot(
-            'search index="_internal" | head 4000 | eventingcsc status=200 | head 10',
-            output_mode="json",
+
+        base_search = self._create_test_data_search(count=100)
+        full_search = (
+            base_search
+            + """
+        | eventingcsc status=200
+        | head 10
+        """
         )
-        result = results.JSONResultsReader(stream)
-        ds = list(result)
+        stream = jobs.oneshot(full_search, output_mode="json")
+
+        reader = results.JSONResultsReader(stream)
+        items = list(reader)
+
+        self.assertEqual(reader.is_preview, False)
 
-        self.assertEqual(result.is_preview, False)
-        self.assertTrue(isinstance(ds[0], (dict, results.Message)))
-        nonmessages = [d for d in ds if isinstance(d, dict)]
-        self.assertTrue(len(nonmessages) <= 10)
+        actual_results = [item for item in items if isinstance(item, dict)]
+        informational_messages = [
+            item for item in items if isinstance(item, results.Message)
+        ]
+
+        self.assertTrue(len(actual_results) > 0)
+
+        fatal_messages = [
+            msg for msg in informational_messages if msg.type in ["FATAL", "ERROR"]
+        ]
+        self.assertEqual(
+            len(fatal_messages),
+            0,
+            f"Should not have FATAL/ERROR messages, but got: {[msg.message for msg in fatal_messages]}",
+        )
+
+        for res in actual_results:
+            self.assertIn("status", res)
+            self.assertEqual(res["status"], "200")
 
     def test_generating_app(self):
         app_name = "generating_app"
@@ -257,6 +281,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()