from lldbsuite.test.decorators import *
from lldbsuite.test.lldbtest import *
import gdbremote_testcase
class LldbGdbServerTestCase(gdbremote_testcase.GdbRemoteTestCaseBase):
@skipIfWindows # no SIGSEGV support
@add_test_categories(["llgs"])
def test_run(self):
self.build()
self.set_inferior_startup_launch()
thread_num = 3
procs = self.prep_debug_monitor_and_inferior(
inferior_args=["thread:segfault"] + thread_num * ["thread:new"]
)
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
],
True,
)
self.expect_gdbremote_sequence()
segv_signo = lldbutil.get_signal_number("SIGSEGV")
all_threads = set()
all_segv_threads = []
# we should get segfaults from all the threads
for segv_no in range(thread_num):
# first wait for the notification event
self.reset_test_sequence()
self.test_sequence.add_log_lines(
[
{
"direction": "send",
"regex": r"^%Stop:(T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);)",
"capture": {1: "packet", 2: "signo", 3: "thread_id"},
},
],
True,
)
m = self.expect_gdbremote_sequence()
del m["O_content"]
threads = [m]
# then we may get events for the remaining threads
# (but note that not all threads may have been started yet)
while True:
self.reset_test_sequence()
self.test_sequence.add_log_lines(
[
"read packet: $vStopped#00",
{
"direction": "send",
"regex": r"^\$(OK|T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);)",
"capture": {1: "packet", 2: "signo", 3: "thread_id"},
},
],
True,
)
m = self.expect_gdbremote_sequence()
if m["packet"] == "OK":
break
del m["O_content"]
threads.append(m)
segv_threads = []
other_threads = []
for t in threads:
signo = int(t["signo"], 16)
if signo == segv_signo:
segv_threads.append(t["thread_id"])
else:
self.assertEqual(signo, 0)
other_threads.append(t["thread_id"])
# verify that exactly one thread segfaulted
self.assertEqual(len(segv_threads), 1)
# we should get only one segv from every thread
self.assertNotIn(segv_threads[0], all_segv_threads)
all_segv_threads.extend(segv_threads)
# segv_threads + other_threads should always be a superset
# of all_threads, i.e. we should get states for all threads
# already started
self.assertFalse(all_threads.difference(other_threads + segv_threads))
all_threads.update(other_threads + segv_threads)
# verify that `?` returns the same result
self.reset_test_sequence()
self.test_sequence.add_log_lines(
[
"read packet: $?#00",
],
True,
)
threads_verify = []
while True:
self.test_sequence.add_log_lines(
[
{
"direction": "send",
"regex": r"^\$(OK|T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);)",
"capture": {1: "packet", 2: "signo", 3: "thread_id"},
},
],
True,
)
m = self.expect_gdbremote_sequence()
if m["packet"] == "OK":
break
del m["O_content"]
threads_verify.append(m)
self.reset_test_sequence()
self.test_sequence.add_log_lines(
[
"read packet: $vStopped#00",
],
True,
)
self.assertEqual(threads, threads_verify)
self.reset_test_sequence()
self.test_sequence.add_log_lines(
[
"read packet: $vCont;C{:02x}:{};c#00".format(
segv_signo, segv_threads[0]
),
"send packet: $OK#00",
],
True,
)
self.expect_gdbremote_sequence()
# finally, verify that all threads have started
self.assertEqual(len(all_threads), thread_num + 1)
@add_test_categories(["llgs"])
def test_vCtrlC(self):
self.build()
self.set_inferior_startup_launch()
procs = self.prep_debug_monitor_and_inferior(inferior_args=["thread:new"])
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
"read packet: $vCtrlC#00",
"send packet: $OK#00",
{
"direction": "send",
"regex": r"^%Stop:T",
},
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["llgs"])
def test_exit(self):
self.build()
self.set_inferior_startup_launch()
procs = self.prep_debug_monitor_and_inferior()
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
"send packet: %Stop:W00#00",
"read packet: $vStopped#00",
"send packet: $OK#00",
],
True,
)
self.expect_gdbremote_sequence()
@skipIfWindows # no clue, the result makes zero sense
@add_test_categories(["llgs"])
def test_exit_query(self):
self.build()
self.set_inferior_startup_launch()
procs = self.prep_debug_monitor_and_inferior()
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
"send packet: %Stop:W00#00",
"read packet: $?#00",
"send packet: $W00#00",
"read packet: $vStopped#00",
"send packet: $OK#00",
],
True,
)
self.expect_gdbremote_sequence()
def multiple_resume_test(self, second_command):
self.build()
self.set_inferior_startup_launch()
procs = self.prep_debug_monitor_and_inferior(inferior_args=["sleep:15"])
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
"read packet: ${}#00".format(second_command),
"send packet: $E37#00",
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["llgs"])
def test_multiple_C_continue_with_signal(self):
self.multiple_resume_test("C05")
@add_test_categories(["llgs"])
def test_multiple_c_continue_with_addr(self):
self.multiple_resume_test("c")
@add_test_categories(["llgs"])
def test_multiple_s_single_step_with_addr(self):
self.multiple_resume_test("s")
@skipIfWindows
@add_test_categories(["llgs"])
def test_multiple_vCont(self):
self.build()
self.set_inferior_startup_launch()
procs = self.prep_debug_monitor_and_inferior(
inferior_args=["thread:new", "stop", "sleep:15"]
)
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
{
"direction": "send",
"regex": r"^%Stop:T[0-9a-fA-F]{2}thread:([0-9a-fA-F]+);",
"capture": {1: "tid1"},
},
"read packet: $vStopped#63",
{
"direction": "send",
"regex": r"^[$]T[0-9a-fA-F]{2}thread:([0-9a-fA-F]+);",
"capture": {1: "tid2"},
},
"read packet: $vStopped#63",
"send packet: $OK#00",
],
True,
)
ret = self.expect_gdbremote_sequence()
self.reset_test_sequence()
self.test_sequence.add_log_lines(
[
"read packet: $vCont;c:{}#00".format(ret["tid1"]),
"send packet: $OK#00",
"read packet: $vCont;c:{}#00".format(ret["tid2"]),
"send packet: $E37#00",
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["llgs"])
@skipIfWindows # Sometimes results in '$E37' instead of expected '$OK'
def test_vCont_then_stop(self):
self.build()
self.set_inferior_startup_launch()
procs = self.prep_debug_monitor_and_inferior(inferior_args=["sleep:15"])
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
"read packet: $vCont;t#00",
"send packet: $OK#00",
],
True,
)
self.expect_gdbremote_sequence()
def vCont_then_partial_stop_test(self, run_both):
self.build()
self.set_inferior_startup_launch()
procs = self.prep_debug_monitor_and_inferior(
inferior_args=["thread:new", "stop", "sleep:15"]
)
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
{
"direction": "send",
"regex": r"^%Stop:T[0-9a-fA-F]{2}thread:([0-9a-fA-F]+);",
"capture": {1: "tid1"},
},
"read packet: $vStopped#63",
{
"direction": "send",
"regex": r"^[$]T[0-9a-fA-F]{2}thread:([0-9a-fA-F]+);",
"capture": {1: "tid2"},
},
"read packet: $vStopped#63",
"send packet: $OK#00",
],
True,
)
ret = self.expect_gdbremote_sequence()
self.reset_test_sequence()
if run_both:
self.test_sequence.add_log_lines(
[
"read packet: $vCont;c#00",
],
True,
)
else:
self.test_sequence.add_log_lines(
[
"read packet: $vCont;c:{}#00".format(ret["tid1"]),
],
True,
)
self.test_sequence.add_log_lines(
[
"send packet: $OK#00",
"read packet: $vCont;t:{}#00".format(ret["tid2"]),
"send packet: $E03#00",
],
True,
)
self.expect_gdbremote_sequence()
@skipIfWindows
@add_test_categories(["llgs"])
def test_vCont_then_partial_stop(self):
self.vCont_then_partial_stop_test(False)
@skipIfWindows
@add_test_categories(["llgs"])
def test_vCont_then_partial_stop_run_both(self):
self.vCont_then_partial_stop_test(True)
@skipIfWindows
@add_test_categories(["llgs"])
def test_stdio(self):
self.build()
self.set_inferior_startup_launch()
# Since we can't easily ensure that lldb will send output in two parts,
# just put a stop in the middle. Since we don't clear vStdio,
# the second message won't be delivered immediately.
self.prep_debug_monitor_and_inferior(
inferior_args=["message 1", "stop", "message 2"]
)
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
{"direction": "send", "regex": r"^%Stop:T.*"},
"read packet: $vStopped#00",
"send packet: $OK#00",
"read packet: $c#63",
"send packet: $OK#00",
"send packet: %Stop:W00#00",
],
True,
)
ret = self.expect_gdbremote_sequence()
# We know there will be at least two messages, but there may be more.
# Loop until we have everything. The first message waiting for us in the
# packet queue.
count = 1
output = self._server.get_raw_output_packet()
while not (b"message 2\r\n" in output):
self._server.send_packet(b"vStdio")
output += self._server.get_raw_output_packet()
count += 1
self.assertGreaterEqual(count, 2)
self.reset_test_sequence()
self.test_sequence.add_log_lines(
[
"read packet: $vStdio#00",
"send packet: $OK#00",
"read packet: $vStopped#00",
"send packet: $OK#00",
],
True,
)
self.expect_gdbremote_sequence()
@skipIfWindows
@add_test_categories(["llgs"])
def test_stop_reason_while_running(self):
self.build()
self.set_inferior_startup_launch()
procs = self.prep_debug_monitor_and_inferior(
inferior_args=["thread:new", "thread:new", "stop", "sleep:15"]
)
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
# stop is used to synchronize starting threads
"read packet: $c#63",
"send packet: $OK#00",
{"direction": "send", "regex": "%Stop:T.*"},
"read packet: $c#63",
"send packet: $OK#00",
"read packet: $?#00",
"send packet: $OK#00",
],
True,
)
self.expect_gdbremote_sequence()
@skipIfWindows
@add_test_categories(["llgs"])
def test_leave_nonstop(self):
self.build()
self.set_inferior_startup_launch()
procs = self.prep_debug_monitor_and_inferior(
inferior_args=["thread:new", "thread:new", "stop", "sleep:15"]
)
self.test_sequence.add_log_lines(
[
"read packet: $QNonStop:1#00",
"send packet: $OK#00",
# stop is used to synchronize starting threads
"read packet: $c#63",
"send packet: $OK#00",
{"direction": "send", "regex": "%Stop:T.*"},
"read packet: $c#63",
"send packet: $OK#00",
# verify that the threads are running now
"read packet: $?#00",
"send packet: $OK#00",
"read packet: $QNonStop:0#00",
"send packet: $OK#00",
# we should issue some random request now to verify that the stub
# did not send stop reasons -- we may verify whether notification
# queue was cleared while at it
"read packet: $vStopped#00",
"send packet: $Eff#00",
"read packet: $?#00",
{"direction": "send", "regex": "[$]T.*"},
],
True,
)
self.expect_gdbremote_sequence()