import random
from lldbsuite.test.decorators import *
from lldbsuite.test.lldbtest import *
from fork_testbase import GdbRemoteForkTestBase
class TestGdbRemoteFork(GdbRemoteForkTestBase):
def setUp(self):
GdbRemoteForkTestBase.setUp(self)
if self.getPlatform() == "linux" and self.getArchitecture() in [
"arm",
"aarch64",
]:
self.skipTest("Unsupported for Arm/AArch64 Linux")
@add_test_categories(["fork"])
def test_fork_multithreaded(self):
_, _, child_pid, _ = self.start_fork_test(["thread:new"] * 2 + ["fork"])
# detach the forked child
self.test_sequence.add_log_lines(
[
"read packet: $D;{}#00".format(child_pid),
"send packet: $OK#00",
"read packet: $k#00",
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_fork(self):
parent_pid, _ = self.fork_and_detach_test("fork")
# resume the parent
self.test_sequence.add_log_lines(
[
"read packet: $c#00",
"send packet: $W00;process:{}#00".format(parent_pid),
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_vfork(self):
parent_pid, parent_tid = self.fork_and_detach_test("vfork")
# resume the parent
self.test_sequence.add_log_lines(
[
"read packet: $c#00",
{
"direction": "send",
"regex": r"[$]T[0-9a-fA-F]{{2}}thread:p{}[.]{}.*vforkdone.*".format(
parent_pid, parent_tid
),
},
"read packet: $c#00",
"send packet: $W00;process:{}#00".format(parent_pid),
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_fork_follow(self):
self.fork_and_follow_test("fork")
@add_test_categories(["fork"])
def test_vfork_follow(self):
self.fork_and_follow_test("vfork")
@add_test_categories(["fork"])
def test_select_wrong_pid(self):
self.build()
self.prep_debug_monitor_and_inferior()
self.add_qSupported_packets(["multiprocess+"])
ret = self.expect_gdbremote_sequence()
self.assertIn("multiprocess+", ret["qSupported_response"])
self.reset_test_sequence()
# get process pid
self.test_sequence.add_log_lines(
[
"read packet: $qC#00",
{
"direction": "send",
"regex": "[$]QCp([0-9a-f]+).([0-9a-f]+)#.*",
"capture": {1: "pid", 2: "tid"},
},
],
True,
)
ret = self.expect_gdbremote_sequence()
pid, tid = (int(ret[x], 16) for x in ("pid", "tid"))
self.reset_test_sequence()
self.test_sequence.add_log_lines(
[
# try switching to correct pid
"read packet: $Hgp{:x}.{:x}#00".format(pid, tid),
"send packet: $OK#00",
"read packet: $Hcp{:x}.{:x}#00".format(pid, tid),
"send packet: $OK#00",
# try switching to invalid tid
"read packet: $Hgp{:x}.{:x}#00".format(pid, tid + 1),
"send packet: $E15#00",
"read packet: $Hcp{:x}.{:x}#00".format(pid, tid + 1),
"send packet: $E15#00",
# try switching to invalid pid
"read packet: $Hgp{:x}.{:x}#00".format(pid + 1, tid),
"send packet: $Eff#00",
"read packet: $Hcp{:x}.{:x}#00".format(pid + 1, tid),
"send packet: $Eff#00",
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_detach_current(self):
self.build()
self.prep_debug_monitor_and_inferior()
self.add_qSupported_packets(["multiprocess+"])
ret = self.expect_gdbremote_sequence()
self.assertIn("multiprocess+", ret["qSupported_response"])
self.reset_test_sequence()
# get process pid
self.test_sequence.add_log_lines(
[
"read packet: $qC#00",
{
"direction": "send",
"regex": "[$]QCp([0-9a-f]+).[0-9a-f]+#.*",
"capture": {1: "pid"},
},
],
True,
)
ret = self.expect_gdbremote_sequence()
pid = ret["pid"]
self.reset_test_sequence()
# detach the process
self.test_sequence.add_log_lines(
[
"read packet: $D;{}#00".format(pid),
"send packet: $OK#00",
"read packet: $qC#00",
"send packet: $E44#00",
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_detach_all(self):
self.detach_all_test()
@add_test_categories(["fork"])
def test_kill_all(self):
parent_pid, _, child_pid, _ = self.start_fork_test(["fork"])
exit_regex = "[$]X09;process:([0-9a-f]+)#.*"
self.test_sequence.add_log_lines(
[
# kill all processes
"read packet: $k#00",
{"direction": "send", "regex": exit_regex, "capture": {1: "pid1"}},
{"direction": "send", "regex": exit_regex, "capture": {1: "pid2"}},
],
True,
)
ret = self.expect_gdbremote_sequence()
self.assertEqual(set([ret["pid1"], ret["pid2"]]), set([parent_pid, child_pid]))
@add_test_categories(["fork"])
def test_vkill_child(self):
self.vkill_test(kill_child=True)
@add_test_categories(["fork"])
def test_vkill_parent(self):
self.vkill_test(kill_parent=True)
@add_test_categories(["fork"])
def test_vkill_both(self):
self.vkill_test(kill_parent=True, kill_child=True)
@add_test_categories(["fork"])
def test_c_parent(self):
self.resume_one_test(run_order=["parent", "parent"])
@add_test_categories(["fork"])
def test_c_child(self):
self.resume_one_test(run_order=["child", "child"])
@add_test_categories(["fork"])
def test_c_parent_then_child(self):
self.resume_one_test(run_order=["parent", "parent", "child", "child"])
@add_test_categories(["fork"])
def test_c_child_then_parent(self):
self.resume_one_test(run_order=["child", "child", "parent", "parent"])
@add_test_categories(["fork"])
def test_c_interspersed(self):
self.resume_one_test(run_order=["parent", "child", "parent", "child"])
@add_test_categories(["fork"])
def test_vCont_parent(self):
self.resume_one_test(run_order=["parent", "parent"], use_vCont=True)
@add_test_categories(["fork"])
def test_vCont_child(self):
self.resume_one_test(run_order=["child", "child"], use_vCont=True)
@add_test_categories(["fork"])
def test_vCont_parent_then_child(self):
self.resume_one_test(
run_order=["parent", "parent", "child", "child"], use_vCont=True
)
@add_test_categories(["fork"])
def test_vCont_child_then_parent(self):
self.resume_one_test(
run_order=["child", "child", "parent", "parent"], use_vCont=True
)
@add_test_categories(["fork"])
def test_vCont_interspersed(self):
self.resume_one_test(
run_order=["parent", "child", "parent", "child"], use_vCont=True
)
@add_test_categories(["fork"])
def test_vCont_two_processes(self):
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
["fork", "stop"]
)
self.test_sequence.add_log_lines(
[
# try to resume both processes
"read packet: $vCont;c:p{}.{};c:p{}.{}#00".format(
parent_pid, parent_tid, child_pid, child_tid
),
"send packet: $E03#00",
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_vCont_all_processes_explicit(self):
self.start_fork_test(["fork", "stop"])
self.test_sequence.add_log_lines(
[
# try to resume all processes implicitly
"read packet: $vCont;c:p-1.-1#00",
"send packet: $E03#00",
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_vCont_all_processes_implicit(self):
self.start_fork_test(["fork", "stop"])
self.test_sequence.add_log_lines(
[
# try to resume all processes implicitly
"read packet: $vCont;c#00",
"send packet: $E03#00",
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_threadinfo(self):
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
["fork", "thread:new", "stop"]
)
pidtids = [
(parent_pid, parent_tid),
(child_pid, child_tid),
]
self.add_threadinfo_collection_packets()
ret = self.expect_gdbremote_sequence()
prev_pidtids = set(self.parse_threadinfo_packets(ret))
self.assertEqual(
prev_pidtids,
frozenset((int(pid, 16), int(tid, 16)) for pid, tid in pidtids),
)
self.reset_test_sequence()
for pidtid in pidtids:
self.test_sequence.add_log_lines(
[
"read packet: $Hcp{}.{}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $c#00",
{
"direction": "send",
"regex": self.stop_regex.format(*pidtid),
},
],
True,
)
self.add_threadinfo_collection_packets()
ret = self.expect_gdbremote_sequence()
self.reset_test_sequence()
new_pidtids = set(self.parse_threadinfo_packets(ret))
added_pidtid = new_pidtids - prev_pidtids
prev_pidtids = new_pidtids
# verify that we've got exactly one new thread, and that
# the PID matches
self.assertEqual(len(added_pidtid), 1)
self.assertEqual(added_pidtid.pop()[0], int(pidtid[0], 16))
for pidtid in new_pidtids:
self.test_sequence.add_log_lines(
[
"read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
"send packet: $OK#00",
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_memory_read_write(self):
self.build()
INITIAL_DATA = "Initial message"
self.prep_debug_monitor_and_inferior(
inferior_args=[
"set-message:{}".format(INITIAL_DATA),
"get-data-address-hex:g_message",
"fork",
"print-message:",
"stop",
]
)
self.add_qSupported_packets(["multiprocess+", "fork-events+"])
ret = self.expect_gdbremote_sequence()
self.assertIn("fork-events+", ret["qSupported_response"])
self.reset_test_sequence()
# continue and expect fork
self.test_sequence.add_log_lines(
[
"read packet: $c#00",
{
"type": "output_match",
"regex": self.maybe_strict_output_regex(
r"data address: 0x([0-9a-fA-F]+)\r\n"
),
"capture": {1: "addr"},
},
{
"direction": "send",
"regex": self.fork_regex.format("fork"),
"capture": self.fork_capture,
},
],
True,
)
ret = self.expect_gdbremote_sequence()
pidtids = {
"parent": (ret["parent_pid"], ret["parent_tid"]),
"child": (ret["child_pid"], ret["child_tid"]),
}
addr = ret["addr"]
self.reset_test_sequence()
for name, pidtid in pidtids.items():
self.test_sequence.add_log_lines(
[
"read packet: $Hgp{}.{}#00".format(*pidtid),
"send packet: $OK#00",
# read the current memory contents
"read packet: $m{},{:x}#00".format(addr, len(INITIAL_DATA) + 1),
{
"direction": "send",
"regex": r"^[$](.+)#.*$",
"capture": {1: "data"},
},
# write a new value
"read packet: $M{},{:x}:{}#00".format(
addr, len(name) + 1, seven.hexlify(name + "\0")
),
"send packet: $OK#00",
# resume the process and wait for the trap
"read packet: $Hcp{}.{}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $c#00",
{
"type": "output_match",
"regex": self.maybe_strict_output_regex(r"message: (.*)\r\n"),
"capture": {1: "printed_message"},
},
{
"direction": "send",
"regex": self.stop_regex.format(*pidtid),
},
],
True,
)
ret = self.expect_gdbremote_sequence()
data = seven.unhexlify(ret["data"])
self.assertEqual(data, INITIAL_DATA + "\0")
self.assertEqual(ret["printed_message"], name)
self.reset_test_sequence()
# we do the second round separately to make sure that initial data
# is correctly preserved while writing into the first process
for name, pidtid in pidtids.items():
self.test_sequence.add_log_lines(
[
"read packet: $Hgp{}.{}#00".format(*pidtid),
"send packet: $OK#00",
# read the current memory contents
"read packet: $m{},{:x}#00".format(addr, len(name) + 1),
{
"direction": "send",
"regex": r"^[$](.+)#.*$",
"capture": {1: "data"},
},
],
True,
)
ret = self.expect_gdbremote_sequence()
self.assertIsNotNone(ret.get("data"))
data = seven.unhexlify(ret.get("data"))
self.assertEqual(data, name + "\0")
self.reset_test_sequence()
@add_test_categories(["fork"])
def test_register_read_write(self):
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
["fork", "thread:new", "stop"]
)
pidtids = [
(parent_pid, parent_tid),
(child_pid, child_tid),
]
for pidtid in pidtids:
self.test_sequence.add_log_lines(
[
"read packet: $Hcp{}.{}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $c#00",
{
"direction": "send",
"regex": self.stop_regex.format(*pidtid),
},
],
True,
)
self.add_threadinfo_collection_packets()
ret = self.expect_gdbremote_sequence()
self.reset_test_sequence()
pidtids = set(self.parse_threadinfo_packets(ret))
self.assertEqual(len(pidtids), 4)
# first, save register values from all the threads
thread_regs = {}
for pidtid in pidtids:
for regno in range(256):
self.test_sequence.add_log_lines(
[
"read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $p{:x}#00".format(regno),
{
"direction": "send",
"regex": r"^[$](.+)#.*$",
"capture": {1: "data"},
},
],
True,
)
ret = self.expect_gdbremote_sequence()
data = ret.get("data")
self.assertIsNotNone(data)
# ignore registers shorter than 32 bits (this also catches
# "Exx" errors)
if len(data) >= 8:
break
else:
self.skipTest("no usable register found")
thread_regs[pidtid] = (regno, data)
vals = set(x[1] for x in thread_regs.values())
# NB: cheap hack to make the loop below easier
new_val = next(iter(vals))
# then, start altering them and verify that we don't unexpectedly
# change the value from another thread
for pidtid in pidtids:
old_val = thread_regs[pidtid]
regno = old_val[0]
old_val_length = len(old_val[1])
# generate a unique new_val
while new_val in vals:
new_val = "{{:0{}x}}".format(old_val_length).format(
random.getrandbits(old_val_length * 4)
)
vals.add(new_val)
self.test_sequence.add_log_lines(
[
"read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $p{:x}#00".format(regno),
{
"direction": "send",
"regex": r"^[$](.+)#.*$",
"capture": {1: "data"},
},
"read packet: $P{:x}={}#00".format(regno, new_val),
"send packet: $OK#00",
],
True,
)
ret = self.expect_gdbremote_sequence()
data = ret.get("data")
self.assertIsNotNone(data)
self.assertEqual(data, old_val[1])
thread_regs[pidtid] = (regno, new_val)
# finally, verify that new values took effect
for pidtid in pidtids:
old_val = thread_regs[pidtid]
self.test_sequence.add_log_lines(
[
"read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $p{:x}#00".format(old_val[0]),
{
"direction": "send",
"regex": r"^[$](.+)#.*$",
"capture": {1: "data"},
},
],
True,
)
ret = self.expect_gdbremote_sequence()
data = ret.get("data")
self.assertIsNotNone(data)
self.assertEqual(data, old_val[1])
@add_test_categories(["fork"])
def test_qC(self):
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
["fork", "thread:new", "stop"]
)
pidtids = [
(parent_pid, parent_tid),
(child_pid, child_tid),
]
for pidtid in pidtids:
self.test_sequence.add_log_lines(
[
"read packet: $Hcp{}.{}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $c#00",
{
"direction": "send",
"regex": self.stop_regex.format(*pidtid),
},
],
True,
)
self.add_threadinfo_collection_packets()
ret = self.expect_gdbremote_sequence()
self.reset_test_sequence()
pidtids = set(self.parse_threadinfo_packets(ret))
self.assertEqual(len(pidtids), 4)
for pidtid in pidtids:
self.test_sequence.add_log_lines(
[
"read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $qC#00",
"send packet: $QCp{:x}.{:x}#00".format(*pidtid),
],
True,
)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_T(self):
parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(
["fork", "thread:new", "stop"]
)
pidtids = [
(parent_pid, parent_tid),
(child_pid, child_tid),
]
for pidtid in pidtids:
self.test_sequence.add_log_lines(
[
"read packet: $Hcp{}.{}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $c#00",
{
"direction": "send",
"regex": self.stop_regex.format(*pidtid),
},
],
True,
)
self.add_threadinfo_collection_packets()
ret = self.expect_gdbremote_sequence()
self.reset_test_sequence()
pidtids = set(self.parse_threadinfo_packets(ret))
self.assertEqual(len(pidtids), 4)
max_pid = max(pid for pid, tid in pidtids)
max_tid = max(tid for pid, tid in pidtids)
bad_pidtids = (
(max_pid, max_tid + 1, "E02"),
(max_pid + 1, max_tid, "E01"),
(max_pid + 1, max_tid + 1, "E01"),
)
for pidtid in pidtids:
self.test_sequence.add_log_lines(
[
# test explicit PID+TID
"read packet: $Tp{:x}.{:x}#00".format(*pidtid),
"send packet: $OK#00",
# test implicit PID via Hg
"read packet: $Hgp{:x}.{:x}#00".format(*pidtid),
"send packet: $OK#00",
"read packet: $T{:x}#00".format(max_tid + 1),
"send packet: $E02#00",
"read packet: $T{:x}#00".format(pidtid[1]),
"send packet: $OK#00",
],
True,
)
for pid, tid, expected in bad_pidtids:
self.test_sequence.add_log_lines(
[
"read packet: $Tp{:x}.{:x}#00".format(pid, tid),
"send packet: ${}#00".format(expected),
],
True,
)
self.expect_gdbremote_sequence()
Codebase Browser
llvm