fix csc apps test

debug: csc test

test using sythetic data

remove debug steps

Author: Cecylia Borek

tests/searchcommands/test_csc_apps.py

@@ -64,17 +64,80 @@ def test_eventing_app(self):
         self.assertEqual(state.title, "eventing_app")
 
         jobs = self.service.jobs
+
+        base_search = self._create_test_data_search(count=100)
+        full_search = (
+            base_search
+            + '''
+        | eventingcsc status=200
+        | head 10
+        '''
+        )
+
+        stream = jobs.oneshot(full_search, output_mode='json')
+
+        reader = results.JSONResultsReader(stream)
+        items = list(reader)
+
+        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]}",
+        )
+
+        first_result = actual_results[0]
+        self.assertIn('status', first_result)
+        self.assertEqual(first_result['status'], '200')
+
+    def test_eventing_app_malformed_query_missing_field(self):
+        app_name = "eventing_app"
+        # Fetch the app
+        app = self.service.apps[app_name]
+        app.refresh()
+
+        jobs = self.service.jobs
+        # wrong search query because `status` field is not being passed to eventing app
         stream = jobs.oneshot(
             'search index="_internal" | head 4000 | eventingcsc status=200 | head 10',
-            output_mode="json",
+            output_mode='json',
         )
-        result = results.JSONResultsReader(stream)
-        ds = list(result)
+        reader = results.JSONResultsReader(stream)
+        items = list(reader)
+
+        self.assertTrue(len(items) > 0, "Should have at least one item (error message)")
+
+        actual_results = [item for item in items if isinstance(item, dict)]
+        informational_messages = [
+            item for item in items if isinstance(item, results.Message)
+        ]
 
-        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)
+        self.assertTrue(len(informational_messages) > 0)
+
+        error_messages = [
+            msg for msg in informational_messages if msg.type in ['FATAL', 'ERROR']
+        ]
+        self.assertTrue(len(error_messages) > 0)
+
+        error_message = error_messages[0]
+        self.assertIn('eventingcsc', error_message.message)
+        self.assertIn('status', error_message.message)
+        self.assertIn('KeyError', error_message.message)
+
+        self.assertEqual(
+            len(actual_results),
+            0,
+            f"Should not have actual results, but got: {actual_results}",
+        )
 
     def test_generating_app(self):
         app_name = "generating_app"
@@ -257,6 +320,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()