From ccf4e8ed2fb531b8f542b9986d48b3f91b2fe95f Mon Sep 17 00:00:00 2001 From: Cecylia Borek Date: Fri, 8 Aug 2025 19:31:53 +0200 Subject: [PATCH 1/3] fix csc apps test --- tests/searchcommands/test_csc_apps.py | 58 ++++++++++++++++++++++----- 1 file changed, 49 insertions(+), 9 deletions(-) diff --git a/tests/searchcommands/test_csc_apps.py b/tests/searchcommands/test_csc_apps.py index b3eb18b4..dfb6cb47 100755 --- a/tests/searchcommands/test_csc_apps.py +++ b/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() From 22238252e34e5ef73ad846e3fb0763447452e375 Mon Sep 17 00:00:00 2001 From: Cecylia Borek Date: Mon, 18 Aug 2025 16:52:42 +0200 Subject: [PATCH 2/3] simplify eventing app test --- tests/searchcommands/test_csc_apps.py | 62 +++++++++------------------ 1 file changed, 20 insertions(+), 42 deletions(-) diff --git a/tests/searchcommands/test_csc_apps.py b/tests/searchcommands/test_csc_apps.py index dfb6cb47..5fb645ac 100755 --- a/tests/searchcommands/test_csc_apps.py +++ b/tests/searchcommands/test_csc_apps.py @@ -63,42 +63,36 @@ def test_eventing_app(self): self.assertEqual(state.title, "eventing_app") - jobs = self.service.jobs + makeresults_count = 20 + expected_results_count = 10 + expected_status = "200" - base_search = self._create_test_data_search(count=100) - full_search = ( - base_search - + """ - | eventingcsc status=200 - | head 10 - """ + search_query = f""" + | makeresults count={makeresults_count} + | streamstats count as row_num + | eval status=case( + (row_num % 2) == 1, 200, + 1=1, 500 ) - stream = jobs.oneshot(full_search, output_mode="json") + | eventingcsc status={expected_status} + """ + stream = self.service.jobs.oneshot(search_query, output_mode="json") - reader = results.JSONResultsReader(stream) - items = list(reader) + results_reader = results.JSONResultsReader(stream) + items = list(results_reader) - self.assertEqual(reader.is_preview, False) + self.assertFalse(results_reader.is_preview) - actual_results = [item for item in items if isinstance(item, dict)] - informational_messages = [ - item for item in items if isinstance(item, results.Message) + # 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) > 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]}", - ) + self.assertTrue(len(actual_results) == expected_results_count) for res in actual_results: self.assertIn("status", res) - self.assertEqual(res["status"], "200") + self.assertEqual(res["status"], expected_status) def test_generating_app(self): app_name = "generating_app" @@ -281,22 +275,6 @@ 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() From 23f02636f851194334f1ded9016cfdcd2f7afdb0 Mon Sep 17 00:00:00 2001 From: Cecylia Borek Date: Mon, 18 Aug 2025 17:09:11 +0200 Subject: [PATCH 3/3] split csc tests into behavior and metadata --- tests/searchcommands/test_csc_apps.py | 81 ++++++++++++++++----------- 1 file changed, 48 insertions(+), 33 deletions(-) diff --git a/tests/searchcommands/test_csc_apps.py b/tests/searchcommands/test_csc_apps.py index 5fb645ac..2b71afe8 100755 --- a/tests/searchcommands/test_csc_apps.py +++ b/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,8 +64,7 @@ def test_eventing_app(self): self.assertEqual(content.version, "1.0.0") self.assertEqual(content.visible, "1") - self.assertEqual(state.title, "eventing_app") - + def test_behavior(self): makeresults_count = 20 expected_results_count = 10 expected_status = "200" @@ -94,15 +96,19 @@ def test_eventing_app(self): self.assertIn("status", res) self.assertEqual(res["status"], expected_status) - def test_generating_app(self): - app_name = "generating_app" +@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 @@ -111,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") @@ -134,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 @@ -159,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") @@ -182,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", @@ -201,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", ) @@ -216,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 @@ -233,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") @@ -256,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", )