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