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