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 without redirecting STDIN to output files already exist.""" 103 self.setup_test() 104 self.build() 105 self.create_target() 106 self.write_file_with_placeholder(self.output_file) 107 self.write_file_with_placeholder(self.error_file) 108 self.redirect_stdout() 109 self.redirect_stderr() 110 self.run_process(True) 111 output = self.read_output_file_and_delete() 112 error = self.read_error_file_and_delete() 113 self.check_process_output(output, error) 114 115 def write_file_with_placeholder(self, target_file): 116 placeholder = "This content should be overwritten." 117 if lldb.remote_platform: 118 self.runCmd( 119 'platform file write "{target}" -d "{data}"'.format( 120 target=target_file, data=placeholder 121 ) 122 ) 123 else: 124 f = open(target_file, "w") 125 f.write(placeholder) 126 f.close() 127 128 # target_file - path on local file system or remote file system if running remote 129 # local_file - path on local system 130 def read_file_and_delete(self, target_file, local_file): 131 if lldb.remote_platform: 132 self.runCmd( 133 'platform get-file "{remote}" "{local}"'.format( 134 remote=target_file, local=local_file 135 ) 136 ) 137 138 self.assertTrue( 139 os.path.exists(local_file), 140 'Make sure "{local}" file exists'.format(local=local_file), 141 ) 142 f = open(local_file, "r") 143 contents = f.read() 144 f.close() 145 146 # TODO: add 'platform delete-file' file command 147 # if lldb.remote_platform: 148 # self.runCmd('platform delete-file "{remote}"'.format(remote=target_file)) 149 os.unlink(local_file) 150 return contents 151 152 def read_output_file_and_delete(self): 153 return self.read_file_and_delete(self.output_file, self.local_output_file) 154 155 def read_error_file_and_delete(self): 156 return self.read_file_and_delete(self.error_file, self.local_error_file) 157 158 def create_target(self): 159 """Create the target and launch info that will be used by all tests""" 160 self.target = self.dbg.CreateTarget(self.exe) 161 self.launch_info = self.target.GetLaunchInfo() 162 self.launch_info.SetWorkingDirectory(self.get_process_working_directory()) 163 164 def redirect_stdin(self): 165 """Redirect STDIN (file descriptor 0) to use our input.txt file 166 167 Make the input.txt file to use when redirecting STDIN, setup a cleanup action 168 to delete the input.txt at the end of the test in case exceptions are thrown, 169 and redirect STDIN in the launch info.""" 170 f = open(self.local_input_file, "w") 171 for line in self.lines: 172 f.write(line + "\n") 173 f.close() 174 175 if lldb.remote_platform: 176 self.runCmd( 177 'platform put-file "{local}" "{remote}"'.format( 178 local=self.local_input_file, remote=self.input_file 179 ) 180 ) 181 182 # This is the function to remove the custom formats in order to have a 183 # clean slate for the next test case. 184 def cleanup(): 185 os.unlink(self.local_input_file) 186 # TODO: add 'platform delete-file' file command 187 # if lldb.remote_platform: 188 # self.runCmd('platform delete-file "{remote}"'.format(remote=self.input_file)) 189 190 # Execute the cleanup function during test case tear down. 191 self.addTearDownHook(cleanup) 192 self.launch_info.AddOpenFileAction(0, self.input_file, True, False) 193 194 def redirect_stdout(self): 195 """Redirect STDOUT (file descriptor 1) to use our output.txt file""" 196 self.launch_info.AddOpenFileAction(1, self.output_file, False, True) 197 198 def redirect_stderr(self): 199 """Redirect STDERR (file descriptor 2) to use our error.txt file""" 200 self.launch_info.AddOpenFileAction(2, self.error_file, False, True) 201 202 def run_process(self, put_stdin): 203 """Run the process to completion and optionally put lines to STDIN via the API if "put_stdin" is True""" 204 # Set the breakpoints 205 self.breakpoint = self.target.BreakpointCreateBySourceRegex( 206 "Set breakpoint here", lldb.SBFileSpec("main.c") 207 ) 208 self.assertGreater(self.breakpoint.GetNumLocations(), 0, VALID_BREAKPOINT) 209 210 # Launch the process, and do not stop at the entry point. 211 error = lldb.SBError() 212 # This should launch the process and it should exit by the time we get back 213 # because we have synchronous mode enabled 214 self.process = self.target.Launch(self.launch_info, error) 215 216 self.assertTrue(error.Success(), "Make sure process launched successfully") 217 self.assertTrue(self.process, PROCESS_IS_VALID) 218 219 if self.TraceOn(): 220 print("process launched.") 221 222 # Frame #0 should be at our breakpoint. 223 threads = lldbutil.get_threads_stopped_at_breakpoint( 224 self.process, self.breakpoint 225 ) 226 227 self.assertEqual(len(threads), 1) 228 self.thread = threads[0] 229 self.frame = self.thread.frames[0] 230 self.assertTrue(self.frame, "Frame 0 is valid.") 231 232 if self.TraceOn(): 233 print("process stopped at breakpoint, sending STDIN via LLDB API.") 234 235 # Write data to stdin via the public API if we were asked to 236 if put_stdin: 237 for line in self.lines: 238 self.process.PutSTDIN(line + "\n") 239 240 # Let process continue so it will exit 241 self.process.Continue() 242 state = self.process.GetState() 243 self.assertState(state, lldb.eStateExited, PROCESS_IS_VALID) 244 245 def check_process_output(self, output, error): 246 # Since we launched the process without specifying stdin/out/err, 247 # a pseudo terminal is used for stdout/err, and we are satisfied 248 # once "input line=>1" appears in stdout. 249 # See also main.c. 250 if self.TraceOn(): 251 print("output = '%s'" % output) 252 print("error = '%s'" % error) 253 254 for line in self.lines: 255 check_line = "input line to stdout: %s" % (line) 256 self.assertIn(check_line, output, "verify stdout line shows up in STDOUT") 257 for line in self.lines: 258 check_line = "input line to stderr: %s" % (line) 259 self.assertIn(check_line, error, "verify stderr line shows up in STDERR") 260