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
|
"""
Test that thread plan listing, and deleting works.
"""
import lldb
from lldbsuite.test.decorators import *
import lldbsuite.test.lldbutil as lldbutil
from lldbsuite.test.lldbtest import *
class TestThreadPlanCommands(TestBase):
NO_DEBUG_INFO_TESTCASE = True
@skipIfWindows
def test_thread_plan_actions(self):
self.build()
self.main_source_file = lldb.SBFileSpec("main.c")
self.thread_plan_test()
def check_list_output(
self, command, active_plans=[], completed_plans=[], discarded_plans=[]
):
# Check the "thread plan list" output against a list of active & completed and discarded plans.
# If all three check arrays are empty, that means the command is expected to fail.
interp = self.dbg.GetCommandInterpreter()
result = lldb.SBCommandReturnObject()
num_active = len(active_plans)
num_completed = len(completed_plans)
num_discarded = len(discarded_plans)
interp.HandleCommand(command, result)
print("Command: %s" % (command))
print(result.GetOutput())
if num_active == 0 and num_completed == 0 and num_discarded == 0:
self.assertFalse(
result.Succeeded(),
"command: '%s' succeeded when it should have failed: '%s'"
% (command, result.GetError()),
)
return
self.assertTrue(
result.Succeeded(),
"command: '%s' failed: '%s'" % (command, result.GetError()),
)
result_arr = result.GetOutput().splitlines()
num_results = len(result_arr)
# Now iterate through the results array and pick out the results.
result_idx = 0
self.assertIn("thread #", result_arr[result_idx], "Found thread header")
result_idx += 1
self.assertIn(
"Active plan stack", result_arr[result_idx], "Found active header"
)
result_idx += 1
self.assertIn(
"Element 0: Base thread plan", result_arr[result_idx], "Found base plan"
)
result_idx += 1
for text in active_plans:
self.assertIn(
text, result_arr[result_idx], "Didn't find active plan: %s" % (text)
)
result_idx += 1
if len(completed_plans) > 0:
# First consume any remaining active plans:
while not "Completed plan stack:" in result_arr[result_idx]:
result_idx += 1
if result_idx == num_results:
self.fail(
"There should have been completed plans, but I never saw the completed stack header"
)
# We are at the Completed header, skip it:
result_idx += 1
for text in completed_plans:
self.assertIn(
text,
result_arr[result_idx],
"Didn't find completed plan: %s" % (text),
)
result_idx += 1
if len(discarded_plans) > 0:
# First consume any remaining completed plans:
while not "Discarded plan stack:" in result_arr[result_idx]:
result_idx += 1
if result_idx == num_results:
self.fail(
"There should have been discarded plans, but I never saw the discarded stack header"
)
# We are at the Discarded header, skip it:
result_idx += 1
for text in discarded_plans:
self.assertIn(
text,
result_arr[result_idx],
"Didn't find discarded plan: %s" % (text),
)
result_idx += 1
def thread_plan_test(self):
(target, process, thread, bkpt) = lldbutil.run_to_source_breakpoint(
self, "Set a breakpoint here", self.main_source_file
)
# We need to have an internal plan so we can test listing one.
# The most consistent way to do that is to use a scripted thread plan
# that uses a sub-plan. Source that in now.
source_path = os.path.join(self.getSourceDir(), "wrap_step_over.py")
self.runCmd("command script import '%s'" % (source_path))
# Now set a breakpoint that we will hit by running our scripted step.
call_me_bkpt = target.BreakpointCreateBySourceRegex(
"Set another here", self.main_source_file
)
self.assertTrue(
call_me_bkpt.GetNumLocations() > 0, "Set the breakpoint successfully"
)
thread.StepUsingScriptedThreadPlan("wrap_step_over.WrapStepOver")
threads = lldbutil.get_threads_stopped_at_breakpoint(process, call_me_bkpt)
self.assertEqual(len(threads), 1, "Hit my breakpoint while stepping over")
current_id = threads[0].GetIndexID()
current_tid = threads[0].GetThreadID()
# Run thread plan list without the -i flag:
command = "thread plan list %d" % (current_id)
self.check_list_output(command, ["wrap_step_over.WrapStepOver"], [])
# Run thread plan list with the -i flag:
command = "thread plan list -i %d" % (current_id)
self.check_list_output(command, ["WrapStepOver", "Stepping over line main.c"])
# Run thread plan list providing TID, output should be the same:
command = "thread plan list -t %d" % (current_tid)
self.check_list_output(command, ["wrap_step_over.WrapStepOver"])
# Provide both index & tid, and make sure we only print once:
command = "thread plan list -t %d %d" % (current_tid, current_id)
self.check_list_output(command, ["wrap_step_over.WrapStepOver"])
# Try a fake TID, and make sure that fails:
fake_tid = 0
for i in range(100, 10000, 100):
fake_tid = current_tid + i
thread = process.GetThreadByID(fake_tid)
if not thread:
break
command = "thread plan list -t %d" % (fake_tid)
self.check_list_output(command)
# Now continue, and make sure we printed the completed plan:
process.Continue()
threads = lldbutil.get_stopped_threads(process, lldb.eStopReasonPlanComplete)
self.assertEqual(len(threads), 1, "One thread completed a step")
# Run thread plan list - there aren't any private plans at this point:
command = "thread plan list %d" % (current_id)
self.check_list_output(command, [], ["wrap_step_over.WrapStepOver"])
# Set another breakpoint that we can run to, to try deleting thread plans.
second_step_bkpt = target.BreakpointCreateBySourceRegex(
"Run here to step over again", self.main_source_file
)
self.assertTrue(
second_step_bkpt.GetNumLocations() > 0, "Set the breakpoint successfully"
)
final_bkpt = target.BreakpointCreateBySourceRegex(
"Make sure we get here on last continue", self.main_source_file
)
self.assertTrue(
final_bkpt.GetNumLocations() > 0, "Set the breakpoint successfully"
)
threads = lldbutil.continue_to_breakpoint(process, second_step_bkpt)
self.assertEqual(len(threads), 1, "Hit the second step breakpoint")
threads[0].StepOver()
threads = lldbutil.get_threads_stopped_at_breakpoint(process, call_me_bkpt)
result = lldb.SBCommandReturnObject()
interp = self.dbg.GetCommandInterpreter()
interp.HandleCommand("thread plan discard 1", result)
self.assertTrue(
result.Succeeded(), "Deleted the step over plan: %s" % (result.GetOutput())
)
# Make sure the plan gets listed in the discarded plans:
command = "thread plan list %d" % (current_id)
self.check_list_output(command, [], [], ["Stepping over line main.c:"])
process.Continue()
threads = lldbutil.get_threads_stopped_at_breakpoint(process, final_bkpt)
self.assertEqual(len(threads), 1, "Ran to final breakpoint")
threads = lldbutil.get_stopped_threads(process, lldb.eStopReasonPlanComplete)
self.assertEqual(len(threads), 0, "Did NOT complete the step over plan")
|
