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
|
from lldbsuite.test.decorators import *
from lldbsuite.test.lldbtest import *
from fork_testbase import GdbRemoteForkTestBase
class TestGdbRemoteForkNonStop(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_vfork_nonstop(self):
parent_pid, parent_tid = self.fork_and_detach_test("vfork",
nonstop=True)
# resume the parent
self.test_sequence.add_log_lines([
"read packet: $c#00",
"send packet: $OK#00",
{"direction": "send",
"regex": r"%Stop:T[0-9a-fA-F]{{2}}thread:p{}[.]{}.*vforkdone.*"
.format(parent_pid, parent_tid),
},
"read packet: $vStopped#00",
"send packet: $OK#00",
"read packet: $c#00",
"send packet: $OK#00",
"send packet: %Stop:W00;process:{}#00".format(parent_pid),
"read packet: $vStopped#00",
"send packet: $OK#00",
], True)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_fork_nonstop(self):
parent_pid, _ = self.fork_and_detach_test("fork", nonstop=True)
# resume the parent
self.test_sequence.add_log_lines([
"read packet: $c#00",
"send packet: $OK#00",
"send packet: %Stop:W00;process:{}#00".format(parent_pid),
"read packet: $vStopped#00",
"send packet: $OK#00",
], True)
self.expect_gdbremote_sequence()
@add_test_categories(["fork"])
def test_fork_follow_nonstop(self):
self.fork_and_follow_test("fork", nonstop=True)
@add_test_categories(["fork"])
def test_vfork_follow_nonstop(self):
self.fork_and_follow_test("vfork", nonstop=True)
@add_test_categories(["fork"])
def test_detach_all_nonstop(self):
self.detach_all_test(nonstop=True)
@add_test_categories(["fork"])
def test_kill_all_nonstop(self):
parent_pid, _, child_pid, _ = self.start_fork_test(["fork"],
nonstop=True)
exit_regex = "X09;process:([0-9a-f]+)"
# Depending on a potential race, the second kill may make it into
# the async queue before we issue vStopped or after. In the former
# case, we should expect the exit status in reply to vStopped.
# In the latter, we should expect an OK response (queue empty),
# followed by another async notification.
vstop_regex = "[$](OK|{})#.*".format(exit_regex)
self.test_sequence.add_log_lines([
# kill all processes
"read packet: $k#00",
"send packet: $OK#00",
{"direction": "send", "regex": "%Stop:{}#.*".format(exit_regex),
"capture": {1: "pid1"}},
"read packet: $vStopped#00",
{"direction": "send", "regex": vstop_regex,
"capture": {1: "vstop_reply", 2: "pid2"}},
], True)
ret = self.expect_gdbremote_sequence()
pid1 = ret["pid1"]
if ret["vstop_reply"] == "OK":
self.reset_test_sequence()
self.test_sequence.add_log_lines([
{"direction": "send", "regex": "%Stop:{}#.*".format(exit_regex),
"capture": {1: "pid2"}},
], True)
ret = self.expect_gdbremote_sequence()
pid2 = ret["pid2"]
self.reset_test_sequence()
self.test_sequence.add_log_lines([
"read packet: $vStopped#00",
"send packet: $OK#00",
], True)
self.expect_gdbremote_sequence()
self.assertEqual(set([pid1, pid2]), set([parent_pid, child_pid]))
@add_test_categories(["fork"])
def test_vkill_both_nonstop(self):
self.vkill_test(kill_parent=True, kill_child=True, nonstop=True)
@add_test_categories(["fork"])
def test_c_interspersed_nonstop(self):
self.resume_one_test(run_order=["parent", "child", "parent", "child"],
nonstop=True)
@add_test_categories(["fork"])
def test_vCont_interspersed_nonstop(self):
self.resume_one_test(run_order=["parent", "child", "parent", "child"],
use_vCont=True, nonstop=True)
@add_test_categories(["fork"])
def test_c_both_nonstop(self):
lock1 = self.getBuildArtifact("lock1")
lock2 = self.getBuildArtifact("lock2")
parent_pid, parent_tid, child_pid, child_tid = (
self.start_fork_test(["fork", "process:sync:" + lock1, "print-pid",
"process:sync:" + lock2, "stop"],
nonstop=True))
self.test_sequence.add_log_lines([
"read packet: $Hcp{}.{}#00".format(parent_pid, parent_tid),
"send packet: $OK#00",
"read packet: $c#00",
"send packet: $OK#00",
"read packet: $Hcp{}.{}#00".format(child_pid, child_tid),
"send packet: $OK#00",
"read packet: $c#00",
"send packet: $OK#00",
{"direction": "send", "regex": "%Stop:T.*"},
# see the comment in TestNonStop.py, test_stdio
"read packet: $vStdio#00",
"read packet: $vStdio#00",
"send packet: $OK#00",
], True)
ret = self.expect_gdbremote_sequence()
self.assertIn("PID: {}".format(int(parent_pid, 16)).encode(),
ret["O_content"])
self.assertIn("PID: {}".format(int(child_pid, 16)).encode(),
ret["O_content"])
|