debug: csc test

Author: Cecylia Borek

.github/workflows/test.yml

@@ -44,8 +44,8 @@ jobs:
       - name: Install tox
         run: pip install tox
 
-      - name: Test Execution
-        run: tox -e py
+      - name: Debug Test Execution
+        run: tox -e py39 -- -s tests/searchcommands/test_csc_apps.py
   fossa-scan:
     uses: splunk/oss-scanning-public/.github/workflows/oss-scan.yml@main
     secrets: inherit

splunklib/results.py

@@ -327,6 +327,7 @@ def _parse_results(self, stream):
             if strip_line.__len__() == 0:
                 continue
             parsed_line = json_loads(strip_line)
+            print(f"JSONResultReader: Parsed line: {parsed_line}")
             if "preview" in parsed_line:
                 self.is_preview = parsed_line["preview"]
             if "messages" in parsed_line and parsed_line["messages"].__len__() > 0:

tests/searchcommands/test_csc_apps.py

@@ -14,12 +14,16 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+import logging
 import unittest
 import pytest
 
+import splunklib
 from tests import testlib
 from splunklib import results
 
+splunklib.setup_logging(logging.DEBUG)
+
 
 @pytest.mark.smoke
 class TestCSC(testlib.SDKTestCase):
@@ -64,13 +68,29 @@ def test_eventing_app(self):
         self.assertEqual(state.title, "eventing_app")
 
         jobs = self.service.jobs
+
+        test_index_stream = jobs.oneshot(
+            'search index="_internal" | head 10', output_mode='json'
+        )
+        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(
             '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)