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