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
|
import lldb
from lldbsuite.test.decorators import *
from lldbsuite.test.lldbtest import *
from lldbsuite.test import lldbutil
from lldbgdbserverutils import get_debugserver_exe
import os
import platform
import shutil
import time
import socket
class TestStopAtEntry(TestBase):
NO_DEBUG_INFO_TESTCASE = True
# The port used by debugserver.
PORT = 54638
# The number of attempts.
ATTEMPTS = 10
# Time given to the binary to launch and to debugserver to attach to it for
# every attempt. We'll wait a maximum of 10 times 2 seconds while the
# inferior will wait 10 times 10 seconds.
TIMEOUT = 2
def no_debugserver(self):
if get_debugserver_exe() is None:
return "no debugserver"
return None
def port_not_available(self):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if s.connect_ex(("127.0.0.1", self.PORT)) == 0:
return "{} not available".format(self.PORT)
return None
@skipUnlessDarwin
@skipIfRemote
def test_stop_default_platform_sync(self):
self.do_test_stop_at_entry(True, False)
@skipUnlessDarwin
@skipIfRemote
def test_stop_default_platform_async(self):
self.do_test_stop_at_entry(False, False)
@skipUnlessDarwin
@skipIfRemote
@expectedFailureIfFn(no_debugserver)
@expectedFailureIfFn(port_not_available)
def test_stop_remote_platform_sync(self):
self.do_test_stop_at_entry(True, True)
@skipUnlessDarwin
@skipIfRemote
@expectedFailureIfFn(no_debugserver)
@expectedFailureIfFn(port_not_available)
def test_stop_remote_platform_async(self):
self.do_test_stop_at_entry(False, True)
def do_test_stop_at_entry(self, synchronous, remote):
"""Test the normal launch path in either sync or async mode"""
self.build()
target = lldbutil.run_to_breakpoint_make_target(self)
launch_info = target.GetLaunchInfo()
launch_info.SetLaunchFlags(lldb.eLaunchFlagStopAtEntry)
old_async = self.dbg.GetAsync()
def cleanup():
self.dbg.SetAsync(old_async)
self.addTearDownHook(cleanup)
if not synchronous:
self.dbg.SetAsync(True)
listener = lldb.SBListener("test-process-listener")
mask = listener.StartListeningForEventClass(
self.dbg,
lldb.SBProcess.GetBroadcasterClassName(),
lldb.SBProcess.eBroadcastBitStateChanged,
)
self.assertEqual(
mask,
lldb.SBProcess.eBroadcastBitStateChanged,
"Got right mask for listener",
)
launch_info.SetListener(listener)
else:
self.dbg.SetAsync(False)
if remote:
self.setup_remote_platform()
error = lldb.SBError()
process = target.Launch(launch_info, error)
self.assertSuccess(error, "Launch failed")
# If we are asynchronous, we have to wait for the events:
if not synchronous:
listener = launch_info.GetListener()
event = lldb.SBEvent()
result = listener.WaitForEvent(30, event)
self.assertTrue(result, "Timed out waiting for event from process")
state = lldb.SBProcess.GetStateFromEvent(event)
self.assertState(
state, lldb.eStateStopped, "Didn't get a stopped state after launch"
)
# Okay, we should be stopped. Make sure we are indeed at the
# entry point. I only know how to do this on darwin:
self.assertEqual(
len(process.threads), 1, "Should only have one thread at entry"
)
thread = process.threads[0]
frame = thread.GetFrameAtIndex(0)
stop_func = frame.name
self.assertEqual(stop_func, "_dyld_start")
# Now make sure that we can resume the process and have it exit.
error = process.Continue()
self.assertSuccess(error, "Error continuing")
# Fetch events till we get eStateExited:
if not synchronous:
# Get events till exited.
listener = launch_info.GetListener()
event = lldb.SBEvent()
# We get two running events in a row here??? That's a bug
# but not the one I'm testing for, so for now just fetch as
# many as were sent.
num_running = 0
state = lldb.eStateRunning
while state == lldb.eStateRunning:
num_running += 1
result = listener.WaitForEvent(30, event)
self.assertTrue(result, "Timed out waiting for running")
state = lldb.SBProcess.GetStateFromEvent(event)
if num_running == 1:
self.assertState(state, lldb.eStateRunning, "Got running event")
# The last event we should get is the exited event
self.assertState(state, lldb.eStateExited, "Got exit event")
else:
# Make sure that the process has indeed exited
state = process.GetState()
self.assertState(state, lldb.eStateExited)
def setup_remote_platform(self):
return
self.build()
exe = self.getBuildArtifact("a.out")
# Launch our test binary.
# Attach to it with debugserver.
debugserver = get_debugserver_exe()
debugserver_args = ["localhost:{}".format(self.PORT)]
self.spawnSubprocess(debugserver, debugserver_args)
# Select the platform.
self.expect("platform select remote-macosx", substrs=[sdk_dir])
# Connect to debugserver
interpreter = self.dbg.GetCommandInterpreter()
connected = False
for i in range(self.ATTEMPTS):
result = lldb.SBCommandReturnObject()
interpreter.HandleCommand("gdb-remote {}".format(self.PORT), result)
connected = result.Succeeded()
if connected:
break
time.sleep(self.TIMEOUT)
self.assertTrue(connected, "could not connect to debugserver")
|
