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