"""Test Python APIs for process IO.""" from __future__ import print_function import os import lldb from lldbsuite.test.decorators import * from lldbsuite.test.lldbtest import * from lldbsuite.test import lldbutil class ProcessIOTestCase(TestBase): mydir = TestBase.compute_mydir(__file__) NO_DEBUG_INFO_TESTCASE = True def setup_test(self): # Get the full path to our executable to be debugged. self.exe = self.getBuildArtifact("process_io") self.local_input_file = self.getBuildArtifact("input.txt") self.local_output_file = self.getBuildArtifact("output.txt") self.local_error_file = self.getBuildArtifact("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']) @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") @skipIfDarwinEmbedded # I/O redirection like this is not supported on remote iOS devices yet def test_stdin_by_api(self): """Exercise SBProcess.PutSTDIN().""" self.setup_test() 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']) @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") def test_stdin_redirection(self): """Exercise SBLaunchInfo::AddOpenFileAction() for STDIN without specifying STDOUT or STDERR.""" self.setup_test() 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']) @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") @skipIfDarwinEmbedded # debugserver can't create/write files on the device def test_stdout_redirection(self): """Exercise SBLaunchInfo::AddOpenFileAction() for STDOUT without specifying STDIN or STDERR.""" self.setup_test() 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']) @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") @skipIfDarwinEmbedded # debugserver can't create/write files on the device def test_stderr_redirection(self): """Exercise SBLaunchInfo::AddOpenFileAction() for STDERR without specifying STDIN or STDOUT.""" self.setup_test() 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']) @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") @skipIfDarwinEmbedded # debugserver can't create/write files on the device def test_stdout_stderr_redirection(self): """Exercise SBLaunchInfo::AddOpenFileAction() for STDOUT and STDERR without redirecting STDIN.""" self.setup_test() 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")