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