1"""Test Python APIs for process IO.""" 2 3from __future__ import print_function 4 5 6import os 7import lldb 8from lldbsuite.test.decorators import * 9from lldbsuite.test.lldbtest import * 10from lldbsuite.test import lldbutil 11 12 13class ProcessIOTestCase(TestBase): 14 15 mydir = TestBase.compute_mydir(__file__) 16 NO_DEBUG_INFO_TESTCASE = True 17 18 @skipIfReproducer 19 def setup_test(self): 20 # Get the full path to our executable to be debugged. 21 self.exe = self.getBuildArtifact("process_io") 22 self.local_input_file = self.getBuildArtifact("input.txt") 23 self.local_output_file = self.getBuildArtifact("output.txt") 24 self.local_error_file = self.getBuildArtifact("error.txt") 25 26 self.input_file = os.path.join( 27 self.get_process_working_directory(), "input.txt") 28 self.output_file = os.path.join( 29 self.get_process_working_directory(), "output.txt") 30 self.error_file = os.path.join( 31 self.get_process_working_directory(), "error.txt") 32 self.lines = ["Line 1", "Line 2", "Line 3"] 33 34 @skipIfWindows # stdio manipulation unsupported on Windows 35 @add_test_categories(['pyapi']) 36 @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") 37 @skipIfDarwinEmbedded # I/O redirection like this is not supported on remote iOS devices yet <rdar://problem/54581135> 38 def test_stdin_by_api(self): 39 """Exercise SBProcess.PutSTDIN().""" 40 self.setup_test() 41 self.build() 42 self.create_target() 43 self.run_process(True) 44 output = self.process.GetSTDOUT(1000) 45 self.check_process_output(output, output) 46 47 @skipIfWindows # stdio manipulation unsupported on Windows 48 @add_test_categories(['pyapi']) 49 @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") 50 def test_stdin_redirection(self): 51 """Exercise SBLaunchInfo::AddOpenFileAction() for STDIN without specifying STDOUT or STDERR.""" 52 self.setup_test() 53 self.build() 54 self.create_target() 55 self.redirect_stdin() 56 self.run_process(False) 57 output = self.process.GetSTDOUT(1000) 58 self.check_process_output(output, output) 59 60 @skipIfWindows # stdio manipulation unsupported on Windows 61 @add_test_categories(['pyapi']) 62 @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") 63 @skipIfDarwinEmbedded # debugserver can't create/write files on the device 64 def test_stdout_redirection(self): 65 """Exercise SBLaunchInfo::AddOpenFileAction() for STDOUT without specifying STDIN or STDERR.""" 66 self.setup_test() 67 self.build() 68 self.create_target() 69 self.redirect_stdout() 70 self.run_process(True) 71 output = self.read_output_file_and_delete() 72 error = self.process.GetSTDOUT(1000) 73 self.check_process_output(output, error) 74 75 @skipIfWindows # stdio manipulation unsupported on Windows 76 @add_test_categories(['pyapi']) 77 @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") 78 @skipIfDarwinEmbedded # debugserver can't create/write files on the device 79 def test_stderr_redirection(self): 80 """Exercise SBLaunchInfo::AddOpenFileAction() for STDERR without specifying STDIN or STDOUT.""" 81 self.setup_test() 82 self.build() 83 self.create_target() 84 self.redirect_stderr() 85 self.run_process(True) 86 output = self.process.GetSTDOUT(1000) 87 error = self.read_error_file_and_delete() 88 self.check_process_output(output, error) 89 90 @skipIfWindows # stdio manipulation unsupported on Windows 91 @add_test_categories(['pyapi']) 92 @expectedFlakeyLinux(bugnumber="llvm.org/pr26437") 93 @skipIfDarwinEmbedded # debugserver can't create/write files on the device 94 def test_stdout_stderr_redirection(self): 95 """Exercise SBLaunchInfo::AddOpenFileAction() for STDOUT and STDERR without redirecting STDIN.""" 96 self.setup_test() 97 self.build() 98 self.create_target() 99 self.redirect_stdout() 100 self.redirect_stderr() 101 self.run_process(True) 102 output = self.read_output_file_and_delete() 103 error = self.read_error_file_and_delete() 104 self.check_process_output(output, error) 105 106 # target_file - path on local file system or remote file system if running remote 107 # local_file - path on local system 108 def read_file_and_delete(self, target_file, local_file): 109 if lldb.remote_platform: 110 self.runCmd('platform get-file "{remote}" "{local}"'.format( 111 remote=target_file, local=local_file)) 112 113 self.assertTrue( 114 os.path.exists(local_file), 115 'Make sure "{local}" file exists'.format( 116 local=local_file)) 117 f = open(local_file, 'r') 118 contents = f.read() 119 f.close() 120 121 # TODO: add 'platform delete-file' file command 122 # if lldb.remote_platform: 123 # self.runCmd('platform delete-file "{remote}"'.format(remote=target_file)) 124 os.unlink(local_file) 125 return contents 126 127 def read_output_file_and_delete(self): 128 return self.read_file_and_delete( 129 self.output_file, self.local_output_file) 130 131 def read_error_file_and_delete(self): 132 return self.read_file_and_delete( 133 self.error_file, self.local_error_file) 134 135 def create_target(self): 136 '''Create the target and launch info that will be used by all tests''' 137 self.target = self.dbg.CreateTarget(self.exe) 138 self.launch_info = self.target.GetLaunchInfo() 139 self.launch_info.SetWorkingDirectory( 140 self.get_process_working_directory()) 141 142 def redirect_stdin(self): 143 '''Redirect STDIN (file descriptor 0) to use our input.txt file 144 145 Make the input.txt file to use when redirecting STDIN, setup a cleanup action 146 to delete the input.txt at the end of the test in case exceptions are thrown, 147 and redirect STDIN in the launch info.''' 148 f = open(self.local_input_file, 'w') 149 for line in self.lines: 150 f.write(line + "\n") 151 f.close() 152 153 if lldb.remote_platform: 154 self.runCmd('platform put-file "{local}" "{remote}"'.format( 155 local=self.local_input_file, remote=self.input_file)) 156 157 # This is the function to remove the custom formats in order to have a 158 # clean slate for the next test case. 159 def cleanup(): 160 os.unlink(self.local_input_file) 161 # TODO: add 'platform delete-file' file command 162 # if lldb.remote_platform: 163 # self.runCmd('platform delete-file "{remote}"'.format(remote=self.input_file)) 164 165 # Execute the cleanup function during test case tear down. 166 self.addTearDownHook(cleanup) 167 self.launch_info.AddOpenFileAction(0, self.input_file, True, False) 168 169 def redirect_stdout(self): 170 '''Redirect STDOUT (file descriptor 1) to use our output.txt file''' 171 self.launch_info.AddOpenFileAction(1, self.output_file, False, True) 172 173 def redirect_stderr(self): 174 '''Redirect STDERR (file descriptor 2) to use our error.txt file''' 175 self.launch_info.AddOpenFileAction(2, self.error_file, False, True) 176 177 def run_process(self, put_stdin): 178 '''Run the process to completion and optionally put lines to STDIN via the API if "put_stdin" is True''' 179 # Set the breakpoints 180 self.breakpoint = self.target.BreakpointCreateBySourceRegex( 181 'Set breakpoint here', lldb.SBFileSpec("main.c")) 182 self.assertTrue( 183 self.breakpoint.GetNumLocations() > 0, 184 VALID_BREAKPOINT) 185 186 # Launch the process, and do not stop at the entry point. 187 error = lldb.SBError() 188 # This should launch the process and it should exit by the time we get back 189 # because we have synchronous mode enabled 190 self.process = self.target.Launch(self.launch_info, error) 191 192 self.assertTrue( 193 error.Success(), 194 "Make sure process launched successfully") 195 self.assertTrue(self.process, PROCESS_IS_VALID) 196 197 if self.TraceOn(): 198 print("process launched.") 199 200 # Frame #0 should be at our breakpoint. 201 threads = lldbutil.get_threads_stopped_at_breakpoint( 202 self.process, self.breakpoint) 203 204 self.assertEqual(len(threads), 1) 205 self.thread = threads[0] 206 self.frame = self.thread.frames[0] 207 self.assertTrue(self.frame, "Frame 0 is valid.") 208 209 if self.TraceOn(): 210 print("process stopped at breakpoint, sending STDIN via LLDB API.") 211 212 # Write data to stdin via the public API if we were asked to 213 if put_stdin: 214 for line in self.lines: 215 self.process.PutSTDIN(line + "\n") 216 217 # Let process continue so it will exit 218 self.process.Continue() 219 state = self.process.GetState() 220 self.assertEqual(state, lldb.eStateExited, PROCESS_IS_VALID) 221 222 def check_process_output(self, output, error): 223 # Since we launched the process without specifying stdin/out/err, 224 # a pseudo terminal is used for stdout/err, and we are satisfied 225 # once "input line=>1" appears in stdout. 226 # See also main.c. 227 if self.TraceOn(): 228 print("output = '%s'" % output) 229 print("error = '%s'" % error) 230 231 for line in self.lines: 232 check_line = 'input line to stdout: %s' % (line) 233 self.assertTrue( 234 check_line in output, 235 "verify stdout line shows up in STDOUT") 236 for line in self.lines: 237 check_line = 'input line to stderr: %s' % (line) 238 self.assertTrue( 239 check_line in error, 240 "verify stderr line shows up in STDERR") 241