initial end-to-end tests for streaming, reporting, generating

Author: zenmoto

splunklib/searchcommands/search_command.py

@@ -851,7 +851,8 @@ def _execute(self, ifile, process):
 
     @staticmethod
     def _as_binary_stream(ifile):
-        if six.PY2:
+        naught = ifile.read(0)
+        if isinstance(naught, bytes):
             return ifile
 
         try:

tests/searchcommands/chunked_data_stream.py

@@ -0,0 +1,19 @@
+import io
+from splunklib.searchcommands import StreamingCommand, Configuration
+
+
+def test_simple_streaming_command():
+    @Configuration()
+    class TestStreamingCommand(StreamingCommand):
+        count = 0
+
+        def stream(self, records):
+            for record in records:
+                record["idx"] = self.count
+                self.count += 1
+                yield record
+
+    cmd = TestStreamingCommand()
+    ifile = io.BytesIO()
+    ofile = io.BytesIO()
+    cmd._process_protocol_v2([], ifile, ofile)