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()