1 files changed, 206 insertions, 0 deletions
diff --git a/packages/Python/lldbsuite/test/python_api/process/io/TestProcessIO.py b/packages/Python/lldbsuite/test/python_api/process/io/TestProcessIO.py
new file mode 100644
index 0000000..2944ee7
--- /dev/null
+++ b/packages/Python/lldbsuite/test/python_api/process/io/TestProcessIO.py
@@ -0,0 +1,206 @@
+"""Test Python APIs for process IO."""
+
+from __future__ import print_function
+
+
+
+import os, sys, time
+import lldb
+from lldbsuite.test.lldbtest import *
+import lldbsuite.test.lldbutil as lldbutil
+
+class ProcessIOTestCase(TestBase):
+
+    mydir = TestBase.compute_mydir(__file__)
+
+    def setUp(self):
+        # Call super's setUp().
+        TestBase.setUp(self)
+        # Get the full path to our executable to be debugged.
+        self.exe = os.path.join(os.getcwd(), "process_io")
+        self.local_input_file  = os.path.join(os.getcwd(), "input.txt")
+        self.local_output_file = os.path.join(os.getcwd(), "output.txt")
+        self.local_error_file  = os.path.join(os.getcwd(), "error.txt")
+
+        self.input_file  = os.path.join(self.get_process_working_directory(), "input.txt")
+        self.output_file = os.path.join(self.get_process_working_directory(), "output.txt")
+        self.error_file  = os.path.join(self.get_process_working_directory(), "error.txt")
+        self.lines = ["Line 1", "Line 2", "Line 3"]
+
+    @skipIfWindows # stdio manipulation unsupported on Windows
+    @add_test_categories(['pyapi'])
+    def test_stdin_by_api(self):
+        """Exercise SBProcess.PutSTDIN()."""
+        self.build()
+        self.create_target()
+        self.run_process(True)
+        output = self.process.GetSTDOUT(1000)
+        self.check_process_output(output, output)
+
+    @skipIfWindows # stdio manipulation unsupported on Windows
+    @add_test_categories(['pyapi'])
+    def test_stdin_redirection(self):
+        """Exercise SBLaunchInfo::AddOpenFileAction() for STDIN without specifying STDOUT or STDERR."""
+        self.build()
+        self.create_target()
+        self.redirect_stdin()
+        self.run_process(False)
+        output = self.process.GetSTDOUT(1000)        
+        self.check_process_output(output, output)
+
+    @skipIfWindows # stdio manipulation unsupported on Windows
+    @add_test_categories(['pyapi'])
+    def test_stdout_redirection(self):
+        """Exercise SBLaunchInfo::AddOpenFileAction() for STDOUT without specifying STDIN or STDERR."""
+        self.build()
+        self.create_target()
+        self.redirect_stdout()
+        self.run_process(True)
+        output = self.read_output_file_and_delete()
+        error = self.process.GetSTDOUT(1000)
+        self.check_process_output(output, error)
+
+    @skipIfWindows # stdio manipulation unsupported on Windows
+    @add_test_categories(['pyapi'])
+    def test_stderr_redirection(self):
+        """Exercise SBLaunchInfo::AddOpenFileAction() for STDERR without specifying STDIN or STDOUT."""
+        self.build()
+        self.create_target()
+        self.redirect_stderr()
+        self.run_process(True)
+        output = self.process.GetSTDOUT(1000)
+        error = self.read_error_file_and_delete()
+        self.check_process_output(output, error)
+
+    @skipIfWindows # stdio manipulation unsupported on Windows
+    @add_test_categories(['pyapi'])
+    def test_stdout_stderr_redirection(self):
+        """Exercise SBLaunchInfo::AddOpenFileAction() for STDOUT and STDERR without redirecting STDIN."""
+        self.build()
+        self.create_target()
+        self.redirect_stdout()
+        self.redirect_stderr()
+        self.run_process(True)
+        output = self.read_output_file_and_delete()
+        error = self.read_error_file_and_delete()
+        self.check_process_output(output, error)
+
+    # target_file - path on local file system or remote file system if running remote
+    # local_file - path on local system
+    def read_file_and_delete(self, target_file, local_file):
+        if lldb.remote_platform:
+            self.runCmd('platform get-file "{remote}" "{local}"'.format(
+                remote=target_file, local=local_file))
+
+        self.assertTrue(os.path.exists(local_file), 'Make sure "{local}" file exists'.format(local=local_file))
+        f = open(local_file, 'r')
+        contents = f.read()
+        f.close()
+
+        #TODO: add 'platform delete-file' file command
+        #if lldb.remote_platform:
+        #    self.runCmd('platform delete-file "{remote}"'.format(remote=target_file))
+        os.unlink(local_file)
+        return contents
+
+    def read_output_file_and_delete(self):
+        return self.read_file_and_delete(self.output_file, self.local_output_file)
+
+    def read_error_file_and_delete(self):
+        return self.read_file_and_delete(self.error_file, self.local_error_file)
+
+    def create_target(self):
+        '''Create the target and launch info that will be used by all tests'''
+        self.target = self.dbg.CreateTarget(self.exe)        
+        self.launch_info = lldb.SBLaunchInfo([self.exe])
+        self.launch_info.SetWorkingDirectory(self.get_process_working_directory())
+    
+    def redirect_stdin(self):
+        '''Redirect STDIN (file descriptor 0) to use our input.txt file
+
+        Make the input.txt file to use when redirecting STDIN, setup a cleanup action
+        to delete the input.txt at the end of the test in case exceptions are thrown,
+        and redirect STDIN in the launch info.'''
+        f = open(self.local_input_file, 'w')
+        for line in self.lines:
+            f.write(line + "\n")
+        f.close()
+
+        if lldb.remote_platform:
+            self.runCmd('platform put-file "{local}" "{remote}"'.format(
+                local=self.local_input_file, remote=self.input_file))
+
+        # This is the function to remove the custom formats in order to have a
+        # clean slate for the next test case.
+        def cleanup():
+            os.unlink(self.local_input_file)
+            #TODO: add 'platform delete-file' file command
+            #if lldb.remote_platform:
+            #    self.runCmd('platform delete-file "{remote}"'.format(remote=self.input_file))
+
+        # Execute the cleanup function during test case tear down.
+        self.addTearDownHook(cleanup)
+        self.launch_info.AddOpenFileAction(0, self.input_file, True, False);
+        
+    def redirect_stdout(self):
+        '''Redirect STDOUT (file descriptor 1) to use our output.txt file'''
+        self.launch_info.AddOpenFileAction(1, self.output_file, False, True);
+    
+    def redirect_stderr(self):
+        '''Redirect STDERR (file descriptor 2) to use our error.txt file'''
+        self.launch_info.AddOpenFileAction(2, self.error_file, False, True);
+        
+    def run_process(self, put_stdin):
+        '''Run the process to completion and optionally put lines to STDIN via the API if "put_stdin" is True'''
+        # Set the breakpoints
+        self.breakpoint = self.target.BreakpointCreateBySourceRegex('Set breakpoint here', lldb.SBFileSpec("main.c"))
+        self.assertTrue(self.breakpoint.GetNumLocations() > 0, VALID_BREAKPOINT)
+
+        # Launch the process, and do not stop at the entry point.
+        error = lldb.SBError()
+        # This should launch the process and it should exit by the time we get back
+        # because we have synchronous mode enabled
+        self.process = self.target.Launch (self.launch_info, error)
+
+        self.assertTrue(error.Success(), "Make sure process launched successfully")
+        self.assertTrue(self.process, PROCESS_IS_VALID)
+
+        if self.TraceOn():
+            print("process launched.")
+
+        # Frame #0 should be at our breakpoint.
+        threads = lldbutil.get_threads_stopped_at_breakpoint (self.process, self.breakpoint)
+        
+        self.assertTrue(len(threads) == 1)
+        self.thread = threads[0]
+        self.frame = self.thread.frames[0]
+        self.assertTrue(self.frame, "Frame 0 is valid.")
+
+        if self.TraceOn():
+            print("process stopped at breakpoint, sending STDIN via LLDB API.")
+
+        # Write data to stdin via the public API if we were asked to
+        if put_stdin:
+            for line in self.lines:
+                self.process.PutSTDIN(line + "\n")
+        
+        # Let process continue so it will exit
+        self.process.Continue()
+        state = self.process.GetState()
+        self.assertTrue(state == lldb.eStateExited, PROCESS_IS_VALID)
+        
+    def check_process_output (self, output, error):
+            # Since we launched the process without specifying stdin/out/err,
+            # a pseudo terminal is used for stdout/err, and we are satisfied
+            # once "input line=>1" appears in stdout.
+            # See also main.c.
+        if self.TraceOn():
+            print("output = '%s'" % output)
+            print("error = '%s'" % error)
+        
+        for line in self.lines:
+            check_line = 'input line to stdout: %s' % (line)
+            self.assertTrue(check_line in output, "verify stdout line shows up in STDOUT")
+        for line in self.lines:
+            check_line = 'input line to stderr: %s' % (line)
+            self.assertTrue(check_line in error, "verify stderr line shows up in STDERR")