1import lldb
2import binascii
3from lldbsuite.test.lldbtest import *
4from lldbsuite.test.decorators import *
5from gdbclientutils import *
6
7
8class TestGDBRemoteClient(GDBRemoteTestBase):
9
10    class gPacketResponder(MockGDBServerResponder):
11        def readRegisters(self):
12            return '0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000'
13
14    def test_connect(self):
15        """Test connecting to a remote gdb server"""
16        target = self.createTarget("a.yaml")
17        process = self.connect(target)
18        self.assertPacketLogContains(["qProcessInfo", "qfThreadInfo"])
19
20    @skipIfReproducer # Unexpected packet during replay
21    def test_attach_fail(self):
22        error_msg = "mock-error-msg"
23
24        class MyResponder(MockGDBServerResponder):
25            # Pretend we don't have any process during the initial queries.
26            def qC(self):
27                return "E42"
28
29            def qfThreadInfo(self):
30                return "OK" # No threads.
31
32            # Then, when we are asked to attach, error out.
33            def vAttach(self, pid):
34                return "E42;" + binascii.hexlify(error_msg.encode()).decode()
35
36        self.server.responder = MyResponder()
37
38        target = self.dbg.CreateTarget("")
39        process = self.connect(target)
40        lldbutil.expect_state_changes(self, self.dbg.GetListener(), process, [lldb.eStateConnected])
41
42        error = lldb.SBError()
43        target.AttachToProcessWithID(lldb.SBListener(), 47, error)
44        self.assertEquals(error_msg, error.GetCString())
45
46    def test_launch_fail(self):
47        class MyResponder(MockGDBServerResponder):
48            # Pretend we don't have any process during the initial queries.
49            def qC(self):
50                return "E42"
51
52            def qfThreadInfo(self):
53                return "OK" # No threads.
54
55            # Then, when we are asked to attach, error out.
56            def A(self, packet):
57                return "E47"
58
59        self.runCmd("log enable gdb-remote packets")
60        self.server.responder = MyResponder()
61
62        target = self.createTarget("a.yaml")
63        process = self.connect(target)
64        lldbutil.expect_state_changes(self, self.dbg.GetListener(), process, [lldb.eStateConnected])
65
66        error = lldb.SBError()
67        target.Launch(lldb.SBListener(), None, None, None, None, None,
68                None, 0, True, error)
69        self.assertEquals("'A' packet returned an error: 71", error.GetCString())
70
71    def test_read_registers_using_g_packets(self):
72        """Test reading registers using 'g' packets (default behavior)"""
73        self.dbg.HandleCommand(
74                "settings set plugin.process.gdb-remote.use-g-packet-for-reading true")
75        self.addTearDownHook(lambda:
76                self.runCmd("settings set plugin.process.gdb-remote.use-g-packet-for-reading false"))
77        self.server.responder = self.gPacketResponder()
78        target = self.createTarget("a.yaml")
79        process = self.connect(target)
80
81        self.assertEquals(1, self.server.responder.packetLog.count("g"))
82        self.server.responder.packetLog = []
83        self.read_registers(process)
84        # Reading registers should not cause any 'p' packets to be exchanged.
85        self.assertEquals(
86                0, len([p for p in self.server.responder.packetLog if p.startswith("p")]))
87
88    def test_read_registers_using_p_packets(self):
89        """Test reading registers using 'p' packets"""
90        self.dbg.HandleCommand(
91                "settings set plugin.process.gdb-remote.use-g-packet-for-reading false")
92        target = self.createTarget("a.yaml")
93        process = self.connect(target)
94
95        self.read_registers(process)
96        self.assertFalse("g" in self.server.responder.packetLog)
97        self.assertGreater(
98                len([p for p in self.server.responder.packetLog if p.startswith("p")]), 0)
99
100    def test_write_registers_using_P_packets(self):
101        """Test writing registers using 'P' packets (default behavior)"""
102        self.server.responder = self.gPacketResponder()
103        target = self.createTarget("a.yaml")
104        process = self.connect(target)
105
106        self.write_registers(process)
107        self.assertEquals(0, len(
108                [p for p in self.server.responder.packetLog if p.startswith("G")]))
109        self.assertGreater(
110                len([p for p in self.server.responder.packetLog if p.startswith("P")]), 0)
111
112    def test_write_registers_using_G_packets(self):
113        """Test writing registers using 'G' packets"""
114
115        class MyResponder(self.gPacketResponder):
116            def readRegister(self, register):
117                # empty string means unsupported
118                return ""
119
120        self.server.responder = MyResponder()
121        target = self.createTarget("a.yaml")
122        process = self.connect(target)
123
124        self.write_registers(process)
125        self.assertEquals(0, len(
126                [p for p in self.server.responder.packetLog if p.startswith("P")]))
127        self.assertGreater(len(
128                [p for p in self.server.responder.packetLog if p.startswith("G")]), 0)
129
130    def read_registers(self, process):
131        self.for_each_gpr(
132                process, lambda r: self.assertEquals("0x00000000", r.GetValue()))
133
134    def write_registers(self, process):
135        self.for_each_gpr(
136                process, lambda r: r.SetValueFromCString("0x00000000"))
137
138    def for_each_gpr(self, process, operation):
139        registers = process.GetThreadAtIndex(0).GetFrameAtIndex(0).GetRegisters()
140        self.assertGreater(registers.GetSize(), 0)
141        regSet = registers[0]
142        numChildren = regSet.GetNumChildren()
143        self.assertGreater(numChildren, 0)
144        for i in range(numChildren):
145            operation(regSet.GetChildAtIndex(i))
146