packages/Python/lldbsuite/test/python_api/process/io/TestProcessIO.py

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