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