624 lines · python
1import random2 3from lldbsuite.test.decorators import *4from lldbsuite.test.lldbtest import *5from lldbsuite.support import seven6 7from fork_testbase import GdbRemoteForkTestBase8 9 10class TestGdbRemoteFork(GdbRemoteForkTestBase):11 def setUp(self):12 GdbRemoteForkTestBase.setUp(self)13 if self.getPlatform() == "linux" and self.getArchitecture() in [14 "arm",15 "aarch64",16 ]:17 self.skipTest("Unsupported for Arm/AArch64 Linux")18 19 @add_test_categories(["fork"])20 def test_fork_multithreaded(self):21 _, _, child_pid, _ = self.start_fork_test(["thread:new"] * 2 + ["fork"])22 23 # detach the forked child24 self.test_sequence.add_log_lines(25 [26 "read packet: $D;{}#00".format(child_pid),27 "send packet: $OK#00",28 "read packet: $k#00",29 ],30 True,31 )32 self.expect_gdbremote_sequence()33 34 @add_test_categories(["fork"])35 def test_fork(self):36 parent_pid, _ = self.fork_and_detach_test("fork")37 38 # resume the parent39 self.test_sequence.add_log_lines(40 [41 "read packet: $c#00",42 "send packet: $W00;process:{}#00".format(parent_pid),43 ],44 True,45 )46 self.expect_gdbremote_sequence()47 48 @add_test_categories(["fork"])49 def test_vfork(self):50 parent_pid, parent_tid = self.fork_and_detach_test("vfork")51 52 # resume the parent53 self.test_sequence.add_log_lines(54 [55 "read packet: $c#00",56 {57 "direction": "send",58 "regex": r"[$]T[0-9a-fA-F]{{2}}thread:p{}[.]{}.*vforkdone.*".format(59 parent_pid, parent_tid60 ),61 },62 "read packet: $c#00",63 "send packet: $W00;process:{}#00".format(parent_pid),64 ],65 True,66 )67 self.expect_gdbremote_sequence()68 69 @add_test_categories(["fork"])70 def test_fork_follow(self):71 self.fork_and_follow_test("fork")72 73 @add_test_categories(["fork"])74 def test_vfork_follow(self):75 self.fork_and_follow_test("vfork")76 77 @add_test_categories(["fork"])78 def test_select_wrong_pid(self):79 self.build()80 self.prep_debug_monitor_and_inferior()81 self.add_qSupported_packets(["multiprocess+"])82 ret = self.expect_gdbremote_sequence()83 self.assertIn("multiprocess+", ret["qSupported_response"])84 self.reset_test_sequence()85 86 # get process pid87 self.test_sequence.add_log_lines(88 [89 "read packet: $qC#00",90 {91 "direction": "send",92 "regex": "[$]QCp([0-9a-f]+).([0-9a-f]+)#.*",93 "capture": {1: "pid", 2: "tid"},94 },95 ],96 True,97 )98 ret = self.expect_gdbremote_sequence()99 pid, tid = (int(ret[x], 16) for x in ("pid", "tid"))100 self.reset_test_sequence()101 102 self.test_sequence.add_log_lines(103 [104 # try switching to correct pid105 "read packet: $Hgp{:x}.{:x}#00".format(pid, tid),106 "send packet: $OK#00",107 "read packet: $Hcp{:x}.{:x}#00".format(pid, tid),108 "send packet: $OK#00",109 # try switching to invalid tid110 "read packet: $Hgp{:x}.{:x}#00".format(pid, tid + 1),111 "send packet: $E15#00",112 "read packet: $Hcp{:x}.{:x}#00".format(pid, tid + 1),113 "send packet: $E15#00",114 # try switching to invalid pid115 "read packet: $Hgp{:x}.{:x}#00".format(pid + 1, tid),116 "send packet: $Eff#00",117 "read packet: $Hcp{:x}.{:x}#00".format(pid + 1, tid),118 "send packet: $Eff#00",119 ],120 True,121 )122 self.expect_gdbremote_sequence()123 124 @add_test_categories(["fork"])125 def test_detach_current(self):126 self.build()127 self.prep_debug_monitor_and_inferior()128 self.add_qSupported_packets(["multiprocess+"])129 ret = self.expect_gdbremote_sequence()130 self.assertIn("multiprocess+", ret["qSupported_response"])131 self.reset_test_sequence()132 133 # get process pid134 self.test_sequence.add_log_lines(135 [136 "read packet: $qC#00",137 {138 "direction": "send",139 "regex": "[$]QCp([0-9a-f]+).[0-9a-f]+#.*",140 "capture": {1: "pid"},141 },142 ],143 True,144 )145 ret = self.expect_gdbremote_sequence()146 pid = ret["pid"]147 self.reset_test_sequence()148 149 # detach the process150 self.test_sequence.add_log_lines(151 [152 "read packet: $D;{}#00".format(pid),153 "send packet: $OK#00",154 "read packet: $qC#00",155 "send packet: $E44#00",156 ],157 True,158 )159 self.expect_gdbremote_sequence()160 161 @add_test_categories(["fork"])162 def test_detach_all(self):163 self.detach_all_test()164 165 @add_test_categories(["fork"])166 def test_kill_all(self):167 parent_pid, _, child_pid, _ = self.start_fork_test(["fork"])168 169 exit_regex = "[$]X09;process:([0-9a-f]+)#.*"170 self.test_sequence.add_log_lines(171 [172 # kill all processes173 "read packet: $k#00",174 {"direction": "send", "regex": exit_regex, "capture": {1: "pid1"}},175 {"direction": "send", "regex": exit_regex, "capture": {1: "pid2"}},176 ],177 True,178 )179 ret = self.expect_gdbremote_sequence()180 self.assertEqual(set([ret["pid1"], ret["pid2"]]), set([parent_pid, child_pid]))181 182 @add_test_categories(["fork"])183 def test_vkill_child(self):184 self.vkill_test(kill_child=True)185 186 @add_test_categories(["fork"])187 def test_vkill_parent(self):188 self.vkill_test(kill_parent=True)189 190 @add_test_categories(["fork"])191 def test_vkill_both(self):192 self.vkill_test(kill_parent=True, kill_child=True)193 194 @add_test_categories(["fork"])195 def test_vCont_two_processes(self):196 parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(197 ["fork", "stop"]198 )199 200 self.test_sequence.add_log_lines(201 [202 # try to resume both processes203 "read packet: $vCont;c:p{}.{};c:p{}.{}#00".format(204 parent_pid, parent_tid, child_pid, child_tid205 ),206 "send packet: $E03#00",207 ],208 True,209 )210 self.expect_gdbremote_sequence()211 212 @add_test_categories(["fork"])213 def test_vCont_all_processes_explicit(self):214 self.start_fork_test(["fork", "stop"])215 216 self.test_sequence.add_log_lines(217 [218 # try to resume all processes implicitly219 "read packet: $vCont;c:p-1.-1#00",220 "send packet: $E03#00",221 ],222 True,223 )224 self.expect_gdbremote_sequence()225 226 @add_test_categories(["fork"])227 def test_vCont_all_processes_implicit(self):228 self.start_fork_test(["fork", "stop"])229 230 self.test_sequence.add_log_lines(231 [232 # try to resume all processes implicitly233 "read packet: $vCont;c#00",234 "send packet: $E03#00",235 ],236 True,237 )238 self.expect_gdbremote_sequence()239 240 @add_test_categories(["fork"])241 def test_threadinfo(self):242 parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(243 ["fork", "thread:new", "stop"]244 )245 pidtids = [246 (parent_pid, parent_tid),247 (child_pid, child_tid),248 ]249 250 self.add_threadinfo_collection_packets()251 ret = self.expect_gdbremote_sequence()252 prev_pidtids = set(self.parse_threadinfo_packets(ret))253 self.assertEqual(254 prev_pidtids,255 frozenset((int(pid, 16), int(tid, 16)) for pid, tid in pidtids),256 )257 self.reset_test_sequence()258 259 for pidtid in pidtids:260 self.test_sequence.add_log_lines(261 [262 "read packet: $Hcp{}.{}#00".format(*pidtid),263 "send packet: $OK#00",264 "read packet: $c#00",265 {266 "direction": "send",267 "regex": self.stop_regex.format(*pidtid),268 },269 ],270 True,271 )272 self.add_threadinfo_collection_packets()273 ret = self.expect_gdbremote_sequence()274 self.reset_test_sequence()275 new_pidtids = set(self.parse_threadinfo_packets(ret))276 added_pidtid = new_pidtids - prev_pidtids277 prev_pidtids = new_pidtids278 279 # verify that we've got exactly one new thread, and that280 # the PID matches281 self.assertEqual(len(added_pidtid), 1)282 self.assertEqual(added_pidtid.pop()[0], int(pidtid[0], 16))283 284 for pidtid in new_pidtids:285 self.test_sequence.add_log_lines(286 [287 "read packet: $Hgp{:x}.{:x}#00".format(*pidtid),288 "send packet: $OK#00",289 ],290 True,291 )292 self.expect_gdbremote_sequence()293 294 @add_test_categories(["fork"])295 def test_memory_read_write(self):296 self.build()297 INITIAL_DATA = "Initial message"298 self.prep_debug_monitor_and_inferior(299 inferior_args=[300 "set-message:{}".format(INITIAL_DATA),301 "get-data-address-hex:g_message",302 "fork",303 "print-message:",304 "stop",305 ]306 )307 self.add_qSupported_packets(["multiprocess+", "fork-events+"])308 ret = self.expect_gdbremote_sequence()309 self.assertIn("fork-events+", ret["qSupported_response"])310 self.reset_test_sequence()311 312 # continue and expect fork313 self.test_sequence.add_log_lines(314 [315 "read packet: $c#00",316 {317 "type": "output_match",318 "regex": self.maybe_strict_output_regex(319 r"data address: 0x([0-9a-fA-F]+)\r\n"320 ),321 "capture": {1: "addr"},322 },323 {324 "direction": "send",325 "regex": self.fork_regex.format("fork"),326 "capture": self.fork_capture,327 },328 ],329 True,330 )331 ret = self.expect_gdbremote_sequence()332 pidtids = {333 "parent": (ret["parent_pid"], ret["parent_tid"]),334 "child": (ret["child_pid"], ret["child_tid"]),335 }336 addr = ret["addr"]337 self.reset_test_sequence()338 339 for name, pidtid in pidtids.items():340 self.test_sequence.add_log_lines(341 [342 "read packet: $Hgp{}.{}#00".format(*pidtid),343 "send packet: $OK#00",344 # read the current memory contents345 "read packet: $m{},{:x}#00".format(addr, len(INITIAL_DATA) + 1),346 {347 "direction": "send",348 "regex": r"^[$](.+)#.*$",349 "capture": {1: "data"},350 },351 # write a new value352 "read packet: $M{},{:x}:{}#00".format(353 addr, len(name) + 1, seven.hexlify(name + "\0")354 ),355 "send packet: $OK#00",356 # resume the process and wait for the trap357 "read packet: $Hcp{}.{}#00".format(*pidtid),358 "send packet: $OK#00",359 "read packet: $c#00",360 {361 "type": "output_match",362 "regex": self.maybe_strict_output_regex(r"message: (.*)\r\n"),363 "capture": {1: "printed_message"},364 },365 {366 "direction": "send",367 "regex": self.stop_regex.format(*pidtid),368 },369 ],370 True,371 )372 ret = self.expect_gdbremote_sequence()373 data = seven.unhexlify(ret["data"])374 self.assertEqual(data, INITIAL_DATA + "\0")375 self.assertEqual(ret["printed_message"], name)376 self.reset_test_sequence()377 378 # we do the second round separately to make sure that initial data379 # is correctly preserved while writing into the first process380 381 for name, pidtid in pidtids.items():382 self.test_sequence.add_log_lines(383 [384 "read packet: $Hgp{}.{}#00".format(*pidtid),385 "send packet: $OK#00",386 # read the current memory contents387 "read packet: $m{},{:x}#00".format(addr, len(name) + 1),388 {389 "direction": "send",390 "regex": r"^[$](.+)#.*$",391 "capture": {1: "data"},392 },393 ],394 True,395 )396 ret = self.expect_gdbremote_sequence()397 self.assertIsNotNone(ret.get("data"))398 data = seven.unhexlify(ret.get("data"))399 self.assertEqual(data, name + "\0")400 self.reset_test_sequence()401 402 @add_test_categories(["fork"])403 def test_register_read_write(self):404 parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(405 ["fork", "thread:new", "stop"]406 )407 pidtids = [408 (parent_pid, parent_tid),409 (child_pid, child_tid),410 ]411 412 for pidtid in pidtids:413 self.test_sequence.add_log_lines(414 [415 "read packet: $Hcp{}.{}#00".format(*pidtid),416 "send packet: $OK#00",417 "read packet: $c#00",418 {419 "direction": "send",420 "regex": self.stop_regex.format(*pidtid),421 },422 ],423 True,424 )425 426 self.add_threadinfo_collection_packets()427 ret = self.expect_gdbremote_sequence()428 self.reset_test_sequence()429 430 pidtids = set(self.parse_threadinfo_packets(ret))431 self.assertEqual(len(pidtids), 4)432 # first, save register values from all the threads433 thread_regs = {}434 for pidtid in pidtids:435 for regno in range(256):436 self.test_sequence.add_log_lines(437 [438 "read packet: $Hgp{:x}.{:x}#00".format(*pidtid),439 "send packet: $OK#00",440 "read packet: $p{:x}#00".format(regno),441 {442 "direction": "send",443 "regex": r"^[$](.+)#.*$",444 "capture": {1: "data"},445 },446 ],447 True,448 )449 ret = self.expect_gdbremote_sequence()450 data = ret.get("data")451 self.assertIsNotNone(data)452 # ignore registers shorter than 32 bits (this also catches453 # "Exx" errors)454 if len(data) >= 8:455 break456 else:457 self.skipTest("no usable register found")458 thread_regs[pidtid] = (regno, data)459 460 vals = set(x[1] for x in thread_regs.values())461 # NB: cheap hack to make the loop below easier462 new_val = next(iter(vals))463 464 # then, start altering them and verify that we don't unexpectedly465 # change the value from another thread466 for pidtid in pidtids:467 old_val = thread_regs[pidtid]468 regno = old_val[0]469 old_val_length = len(old_val[1])470 # generate a unique new_val471 while new_val in vals:472 new_val = "{{:0{}x}}".format(old_val_length).format(473 random.getrandbits(old_val_length * 4)474 )475 vals.add(new_val)476 477 self.test_sequence.add_log_lines(478 [479 "read packet: $Hgp{:x}.{:x}#00".format(*pidtid),480 "send packet: $OK#00",481 "read packet: $p{:x}#00".format(regno),482 {483 "direction": "send",484 "regex": r"^[$](.+)#.*$",485 "capture": {1: "data"},486 },487 "read packet: $P{:x}={}#00".format(regno, new_val),488 "send packet: $OK#00",489 ],490 True,491 )492 ret = self.expect_gdbremote_sequence()493 data = ret.get("data")494 self.assertIsNotNone(data)495 self.assertEqual(data, old_val[1])496 thread_regs[pidtid] = (regno, new_val)497 498 # finally, verify that new values took effect499 for pidtid in pidtids:500 old_val = thread_regs[pidtid]501 self.test_sequence.add_log_lines(502 [503 "read packet: $Hgp{:x}.{:x}#00".format(*pidtid),504 "send packet: $OK#00",505 "read packet: $p{:x}#00".format(old_val[0]),506 {507 "direction": "send",508 "regex": r"^[$](.+)#.*$",509 "capture": {1: "data"},510 },511 ],512 True,513 )514 ret = self.expect_gdbremote_sequence()515 data = ret.get("data")516 self.assertIsNotNone(data)517 self.assertEqual(data, old_val[1])518 519 @add_test_categories(["fork"])520 def test_qC(self):521 parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(522 ["fork", "thread:new", "stop"]523 )524 pidtids = [525 (parent_pid, parent_tid),526 (child_pid, child_tid),527 ]528 529 for pidtid in pidtids:530 self.test_sequence.add_log_lines(531 [532 "read packet: $Hcp{}.{}#00".format(*pidtid),533 "send packet: $OK#00",534 "read packet: $c#00",535 {536 "direction": "send",537 "regex": self.stop_regex.format(*pidtid),538 },539 ],540 True,541 )542 543 self.add_threadinfo_collection_packets()544 ret = self.expect_gdbremote_sequence()545 self.reset_test_sequence()546 547 pidtids = set(self.parse_threadinfo_packets(ret))548 self.assertEqual(len(pidtids), 4)549 for pidtid in pidtids:550 self.test_sequence.add_log_lines(551 [552 "read packet: $Hgp{:x}.{:x}#00".format(*pidtid),553 "send packet: $OK#00",554 "read packet: $qC#00",555 "send packet: $QCp{:x}.{:x}#00".format(*pidtid),556 ],557 True,558 )559 self.expect_gdbremote_sequence()560 561 @add_test_categories(["fork"])562 def test_T(self):563 parent_pid, parent_tid, child_pid, child_tid = self.start_fork_test(564 ["fork", "thread:new", "stop"]565 )566 pidtids = [567 (parent_pid, parent_tid),568 (child_pid, child_tid),569 ]570 571 for pidtid in pidtids:572 self.test_sequence.add_log_lines(573 [574 "read packet: $Hcp{}.{}#00".format(*pidtid),575 "send packet: $OK#00",576 "read packet: $c#00",577 {578 "direction": "send",579 "regex": self.stop_regex.format(*pidtid),580 },581 ],582 True,583 )584 585 self.add_threadinfo_collection_packets()586 ret = self.expect_gdbremote_sequence()587 self.reset_test_sequence()588 589 pidtids = set(self.parse_threadinfo_packets(ret))590 self.assertEqual(len(pidtids), 4)591 max_pid = max(pid for pid, tid in pidtids)592 max_tid = max(tid for pid, tid in pidtids)593 bad_pidtids = (594 (max_pid, max_tid + 1, "E02"),595 (max_pid + 1, max_tid, "E01"),596 (max_pid + 1, max_tid + 1, "E01"),597 )598 599 for pidtid in pidtids:600 self.test_sequence.add_log_lines(601 [602 # test explicit PID+TID603 "read packet: $Tp{:x}.{:x}#00".format(*pidtid),604 "send packet: $OK#00",605 # test implicit PID via Hg606 "read packet: $Hgp{:x}.{:x}#00".format(*pidtid),607 "send packet: $OK#00",608 "read packet: $T{:x}#00".format(max_tid + 1),609 "send packet: $E02#00",610 "read packet: $T{:x}#00".format(pidtid[1]),611 "send packet: $OK#00",612 ],613 True,614 )615 for pid, tid, expected in bad_pidtids:616 self.test_sequence.add_log_lines(617 [618 "read packet: $Tp{:x}.{:x}#00".format(pid, tid),619 "send packet: ${}#00".format(expected),620 ],621 True,622 )623 self.expect_gdbremote_sequence()624