Merge pull request #633 from splunk/dev/CB/DVPL-12497-fix-csc-apps-test

DVPL-12497: Fix CSC apps' tests

Author: Ickerday

tests/searchcommands/test_csc_apps.py

@@ -22,16 +22,17 @@
 
 
 @pytest.mark.smoke
-class TestCSC(testlib.SDKTestCase):
-    def test_eventing_app(self):
-        app_name = "eventing_app"
+class TestEventingApp(testlib.SDKTestCase):
+    app_name = "eventing_app"
 
+    def test_metadata(self):
         self.assertTrue(
-            app_name in self.service.apps, msg="%s is not installed." % app_name
+            TestEventingApp.app_name in self.service.apps,
+            msg=f"{TestEventingApp.app_name} is not installed.",
         )
 
         # Fetch the app
-        app = self.service.apps[app_name]
+        app = self.service.apps[TestEventingApp.app_name]
         app.refresh()
 
         # Extract app info
@@ -40,6 +41,8 @@ def test_eventing_app(self):
         state = app.state
 
         # App info assertions
+        self.assertEqual(state.title, TestEventingApp.app_name)
+
         self.assertEqual(access.app, "system")
         self.assertEqual(access.can_change_perms, "1")
         self.assertEqual(access.can_list, "1")
@@ -61,30 +64,51 @@ def test_eventing_app(self):
         self.assertEqual(content.version, "1.0.0")
         self.assertEqual(content.visible, "1")
 
-        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",
+    def test_behavior(self):
+        makeresults_count = 20
+        expected_results_count = 10
+        expected_status = "200"
+
+        search_query = f"""
+        | makeresults count={makeresults_count}
+        | streamstats count as row_num
+        | eval status=case(
+            (row_num % 2) == 1, 200,
+            1=1, 500
         )
-        result = results.JSONResultsReader(stream)
-        ds = list(result)
+        | eventingcsc status={expected_status}
+        """
+        stream = self.service.jobs.oneshot(search_query, output_mode="json")
+
+        results_reader = results.JSONResultsReader(stream)
+        items = list(results_reader)
 
-        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.assertFalse(results_reader.is_preview)
 
-    def test_generating_app(self):
-        app_name = "generating_app"
+        # filter out informational messages and keep only search results
+        actual_results = [
+            item for item in items if not isinstance(item, results.Message)
+        ]
 
+        self.assertTrue(len(actual_results) == expected_results_count)
+
+        for res in actual_results:
+            self.assertIn("status", res)
+            self.assertEqual(res["status"], expected_status)
+
+
+@pytest.mark.smoke
+class TestGeneratingApp(testlib.SDKTestCase):
+    app_name = "generating_app"
+
+    def test_metadata(self):
         self.assertTrue(
-            app_name in self.service.apps, msg="%s is not installed." % app_name
+            TestGeneratingApp.app_name in self.service.apps,
+            msg=f"{TestGeneratingApp.app_name} is not installed.",
         )
 
         # Fetch the app
-        app = self.service.apps[app_name]
+        app = self.service.apps[TestGeneratingApp.app_name]
         app.refresh()
 
         # Extract app info
@@ -93,6 +117,8 @@ def test_generating_app(self):
         state = app.state
 
         # App info assertions
+        self.assertEqual(state.title, TestGeneratingApp.app_name)
+
         self.assertEqual(access.app, "system")
         self.assertEqual(access.can_change_perms, "1")
         self.assertEqual(access.can_list, "1")
@@ -116,23 +142,27 @@ def test_generating_app(self):
         self.assertEqual(content.version, "1.0.0")
         self.assertEqual(content.visible, "1")
 
-        self.assertEqual(state.title, "generating_app")
-
-        jobs = self.service.jobs
-        stream = jobs.oneshot("| generatingcsc count=4", output_mode="json")
+    def test_behavior(self):
+        stream = self.service.jobs.oneshot(
+            "| generatingcsc count=4", output_mode="json"
+        )
         result = results.JSONResultsReader(stream)
         ds = list(result)
         self.assertTrue(len(ds) == 4)
 
-    def test_reporting_app(self):
-        app_name = "reporting_app"
 
+@pytest.mark.smoke
+class TestReportingApp(testlib.SDKTestCase):
+    app_name = "reporting_app"
+
+    def test_metadata(self):
         self.assertTrue(
-            app_name in self.service.apps, msg="%s is not installed." % app_name
+            TestReportingApp.app_name in self.service.apps,
+            msg=f"{TestReportingApp.app_name} is not installed.",
         )
 
         # Fetch the app
-        app = self.service.apps[app_name]
+        app = self.service.apps[TestReportingApp.app_name]
         app.refresh()
 
         # Extract app info
@@ -141,6 +171,8 @@ def test_reporting_app(self):
         state = app.state
 
         # App info assertions
+        self.assertEqual(state.title, TestReportingApp.app_name)
+
         self.assertEqual(access.app, "system")
         self.assertEqual(access.can_change_perms, "1")
         self.assertEqual(access.can_list, "1")
@@ -164,11 +196,9 @@ def test_reporting_app(self):
         self.assertEqual(content.version, "1.0.0")
         self.assertEqual(content.visible, "1")
 
-        self.assertEqual(state.title, "reporting_app")
-
+    def test_behavior_all_entries_above_cutoff(self):
         jobs = self.service.jobs
 
-        # All above 150
         stream = jobs.oneshot(
             "| makeresults count=10 | eval math=100, eng=100, cs=100 | reportingcsc cutoff=150 math eng cs",
             output_mode="json",
@@ -183,8 +213,8 @@ def test_reporting_app(self):
         no_of_students = int(list(ds[0].values())[0])
         self.assertTrue(no_of_students == 10)
 
-        # All below 150
-        stream = jobs.oneshot(
+    def test_behavior_all_entries_below_cutoff(self):
+        stream = self.service.jobs.oneshot(
             "| makeresults count=10 | eval math=45, eng=45, cs=45 | reportingcsc cutoff=150 math eng cs",
             output_mode="json",
         )
@@ -198,15 +228,19 @@ def test_reporting_app(self):
         no_of_students = int(list(ds[0].values())[0])
         self.assertTrue(no_of_students == 0)
 
-    def test_streaming_app(self):
-        app_name = "streaming_app"
 
+@pytest.mark.smoke
+class TestStreamingApp(testlib.SDKTestCase):
+    app_name = "streaming_app"
+
+    def test_metadata(self):
         self.assertTrue(
-            app_name in self.service.apps, msg="%s is not installed." % app_name
+            TestStreamingApp.app_name in self.service.apps,
+            msg=f"{TestStreamingApp.app_name} is not installed.",
         )
 
         # Fetch the app
-        app = self.service.apps[app_name]
+        app = self.service.apps[TestStreamingApp.app_name]
         app.refresh()
 
         # Extract app info
@@ -215,6 +249,8 @@ def test_streaming_app(self):
         state = app.state
 
         # App info assertions
+        self.assertEqual(state.title, TestStreamingApp.app_name)
+
         self.assertEqual(access.app, "system")
         self.assertEqual(access.can_change_perms, "1")
         self.assertEqual(access.can_list, "1")
@@ -238,11 +274,8 @@ def test_streaming_app(self):
         self.assertEqual(content.version, "1.0.0")
         self.assertEqual(content.visible, "1")
 
-        self.assertEqual(state.title, "streaming_app")
-
-        jobs = self.service.jobs
-
-        stream = jobs.oneshot(
+    def test_behavior(self):
+        stream = self.service.jobs.oneshot(
             "| makeresults count=5 | eval celsius = 35 | streamingcsc",
             output_mode="json",
         )