1"""
2Test case for testing the gdbremote protocol.
3
4Tests run against debugserver and lldb-server (llgs).
5lldb-server tests run where the lldb-server exe is
6available.
7
8This class will be broken into smaller test case classes by
9gdb remote packet functional areas.  For now it contains
10the initial set of tests implemented.
11"""
12
13import binascii
14import itertools
15import struct
16
17import unittest2
18import gdbremote_testcase
19import lldbgdbserverutils
20from lldbsuite.support import seven
21from lldbsuite.test.decorators import *
22from lldbsuite.test.lldbtest import *
23from lldbsuite.test.lldbdwarf import *
24from lldbsuite.test import lldbutil, lldbplatformutil
25
26
27class LldbGdbServerTestCase(gdbremote_testcase.GdbRemoteTestCaseBase, DwarfOpcodeParser):
28
29    mydir = TestBase.compute_mydir(__file__)
30
31    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
32    @skipIf(dwarf_version=['<', '3'])
33    def test_thread_suffix_supported(self):
34        server = self.connect_to_debug_monitor()
35        self.assertIsNotNone(server)
36
37        self.do_handshake()
38        self.test_sequence.add_log_lines(
39            ["lldb-server <  26> read packet: $QThreadSuffixSupported#e4",
40             "lldb-server <   6> send packet: $OK#9a"],
41            True)
42
43        self.expect_gdbremote_sequence()
44
45
46    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
47    @skipIf(dwarf_version=['<', '3'])
48    def test_list_threads_in_stop_reply_supported(self):
49        server = self.connect_to_debug_monitor()
50        self.assertIsNotNone(server)
51
52        self.do_handshake()
53        self.test_sequence.add_log_lines(
54            ["lldb-server <  27> read packet: $QListThreadsInStopReply#21",
55             "lldb-server <   6> send packet: $OK#9a"],
56            True)
57        self.expect_gdbremote_sequence()
58
59    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
60    @skipIf(dwarf_version=['<', '3'])
61    def test_c_packet_works(self):
62        self.build()
63        procs = self.prep_debug_monitor_and_inferior()
64        self.test_sequence.add_log_lines(
65            ["read packet: $c#63",
66             "send packet: $W00#00"],
67            True)
68
69        self.expect_gdbremote_sequence()
70
71    @skipIfWindows # No pty support to test any inferior output
72    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
73    @skipIf(dwarf_version=['<', '3'])
74    def test_inferior_print_exit(self):
75        self.build()
76        procs = self.prep_debug_monitor_and_inferior(
77                inferior_args=["hello, world"])
78        self.test_sequence.add_log_lines(
79            ["read packet: $vCont;c#a8",
80             {"type": "output_match", "regex": self.maybe_strict_output_regex(r"hello, world\r\n")},
81             "send packet: $W00#00"],
82            True)
83
84        context = self.expect_gdbremote_sequence()
85        self.assertIsNotNone(context)
86
87    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
88    @skipIf(dwarf_version=['<', '3'])
89    def test_first_launch_stop_reply_thread_matches_first_qC(self):
90        self.build()
91        procs = self.prep_debug_monitor_and_inferior()
92        self.test_sequence.add_log_lines(["read packet: $qC#00",
93                                          {"direction": "send",
94                                           "regex": r"^\$QC([0-9a-fA-F]+)#",
95                                           "capture": {1: "thread_id_QC"}},
96                                          "read packet: $?#00",
97                                          {"direction": "send",
98                                              "regex": r"^\$T[0-9a-fA-F]{2}thread:([0-9a-fA-F]+)",
99                                              "capture": {1: "thread_id_?"}}],
100                                         True)
101        context = self.expect_gdbremote_sequence()
102        self.assertEqual(context.get("thread_id_QC"), context.get("thread_id_?"))
103
104    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
105    @skipIf(dwarf_version=['<', '3'])
106    def test_attach_commandline_continue_app_exits(self):
107        self.build()
108        self.set_inferior_startup_attach()
109        procs = self.prep_debug_monitor_and_inferior()
110        self.test_sequence.add_log_lines(
111            ["read packet: $vCont;c#a8",
112             "send packet: $W00#00"],
113            True)
114        self.expect_gdbremote_sequence()
115
116        # Wait a moment for completed and now-detached inferior process to
117        # clear.
118        time.sleep(1)
119
120        if not lldb.remote_platform:
121            # Process should be dead now. Reap results.
122            poll_result = procs["inferior"].poll()
123            self.assertIsNotNone(poll_result)
124
125        # Where possible, verify at the system level that the process is not
126        # running.
127        self.assertFalse(
128            lldbgdbserverutils.process_is_running(
129                procs["inferior"].pid, False))
130
131    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
132    @skipIf(dwarf_version=['<', '3'])
133    def test_qRegisterInfo_returns_one_valid_result(self):
134        self.build()
135        self.prep_debug_monitor_and_inferior()
136        self.test_sequence.add_log_lines(
137            ["read packet: $qRegisterInfo0#00",
138             {"direction": "send", "regex": r"^\$(.+);#[0-9A-Fa-f]{2}", "capture": {1: "reginfo_0"}}],
139            True)
140
141        # Run the stream
142        context = self.expect_gdbremote_sequence()
143        self.assertIsNotNone(context)
144
145        reg_info_packet = context.get("reginfo_0")
146        self.assertIsNotNone(reg_info_packet)
147        self.assert_valid_reg_info(
148            lldbgdbserverutils.parse_reg_info_response(reg_info_packet))
149
150    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
151    @skipIf(dwarf_version=['<', '3'])
152    def test_qRegisterInfo_returns_all_valid_results(self):
153        self.build()
154        self.prep_debug_monitor_and_inferior()
155        self.add_register_info_collection_packets()
156
157        # Run the stream.
158        context = self.expect_gdbremote_sequence()
159        self.assertIsNotNone(context)
160
161        # Validate that each register info returned validates.
162        for reg_info in self.parse_register_info_packets(context):
163            self.assert_valid_reg_info(reg_info)
164
165    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
166    @skipIf(dwarf_version=['<', '3'])
167    def test_qRegisterInfo_contains_required_generics_debugserver(self):
168        self.build()
169        self.prep_debug_monitor_and_inferior()
170        self.add_register_info_collection_packets()
171
172        # Run the packet stream.
173        context = self.expect_gdbremote_sequence()
174        self.assertIsNotNone(context)
175
176        # Gather register info entries.
177        reg_infos = self.parse_register_info_packets(context)
178
179        # Collect all generic registers found.
180        generic_regs = {
181            reg_info['generic']: 1 for reg_info in reg_infos if 'generic' in reg_info}
182
183        # Ensure we have a program counter register.
184        self.assertIn('pc', generic_regs)
185
186        # Ensure we have a frame pointer register. PPC64le's FP is the same as SP
187        if self.getArchitecture() != 'powerpc64le':
188            self.assertIn('fp', generic_regs)
189
190        # Ensure we have a stack pointer register.
191        self.assertIn('sp', generic_regs)
192
193        # Ensure we have a flags register.
194        self.assertIn('flags', generic_regs)
195
196    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
197    @skipIf(dwarf_version=['<', '3'])
198    def test_qRegisterInfo_contains_at_least_one_register_set(self):
199        self.build()
200        self.prep_debug_monitor_and_inferior()
201        self.add_register_info_collection_packets()
202
203        # Run the packet stream.
204        context = self.expect_gdbremote_sequence()
205        self.assertIsNotNone(context)
206
207        # Gather register info entries.
208        reg_infos = self.parse_register_info_packets(context)
209
210        # Collect all register sets found.
211        register_sets = {
212            reg_info['set']: 1 for reg_info in reg_infos if 'set' in reg_info}
213        self.assertTrue(len(register_sets) >= 1)
214
215    def targetHasAVX(self):
216        triple = self.dbg.GetSelectedPlatform().GetTriple()
217
218        # TODO other platforms, please implement this function
219        if not re.match(".*-.*-linux", triple):
220            return True
221
222        # Need to do something different for non-Linux/Android targets
223        if lldb.remote_platform:
224            self.runCmd('platform get-file "/proc/cpuinfo" "cpuinfo"')
225            cpuinfo_path = "cpuinfo"
226            self.addTearDownHook(lambda: os.unlink("cpuinfo"))
227        else:
228            cpuinfo_path = "/proc/cpuinfo"
229
230        f = open(cpuinfo_path, 'r')
231        cpuinfo = f.read()
232        f.close()
233        return " avx " in cpuinfo
234
235    @expectedFailureAll(oslist=["windows"]) # no avx for now.
236    @skipIf(archs=no_match(['amd64', 'i386', 'x86_64']))
237    @add_test_categories(["llgs"])
238    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
239    @skipIf(dwarf_version=['<', '3'])
240    def test_qRegisterInfo_contains_avx_registers(self):
241        self.build()
242        self.prep_debug_monitor_and_inferior()
243        self.add_register_info_collection_packets()
244
245        # Run the packet stream.
246        context = self.expect_gdbremote_sequence()
247        self.assertIsNotNone(context)
248
249        # Gather register info entries.
250        reg_infos = self.parse_register_info_packets(context)
251
252        # Collect all generics found.
253        register_sets = {
254            reg_info['set']: 1 for reg_info in reg_infos if 'set' in reg_info}
255        self.assertEqual(
256            self.targetHasAVX(),
257            "Advanced Vector Extensions" in register_sets)
258
259    def qThreadInfo_contains_thread(self):
260        procs = self.prep_debug_monitor_and_inferior()
261        self.add_threadinfo_collection_packets()
262
263        # Run the packet stream.
264        context = self.expect_gdbremote_sequence()
265        self.assertIsNotNone(context)
266
267        # Gather threadinfo entries.
268        threads = self.parse_threadinfo_packets(context)
269        self.assertIsNotNone(threads)
270
271        # We should have exactly one thread.
272        self.assertEqual(len(threads), 1)
273
274    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
275    @skipIf(dwarf_version=['<', '3'])
276    def test_qThreadInfo_contains_thread_launch(self):
277        self.build()
278        self.set_inferior_startup_launch()
279        self.qThreadInfo_contains_thread()
280
281    @expectedFailureAll(oslist=["windows"]) # expect one more thread stopped
282    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
283    @skipIf(dwarf_version=['<', '3'])
284    def test_qThreadInfo_contains_thread_attach(self):
285        self.build()
286        self.set_inferior_startup_attach()
287        self.qThreadInfo_contains_thread()
288
289    def qThreadInfo_matches_qC(self):
290        procs = self.prep_debug_monitor_and_inferior()
291
292        self.add_threadinfo_collection_packets()
293        self.test_sequence.add_log_lines(
294            ["read packet: $qC#00",
295             {"direction": "send", "regex": r"^\$QC([0-9a-fA-F]+)#", "capture": {1: "thread_id"}}
296             ], True)
297
298        # Run the packet stream.
299        context = self.expect_gdbremote_sequence()
300        self.assertIsNotNone(context)
301
302        # Gather threadinfo entries.
303        threads = self.parse_threadinfo_packets(context)
304        self.assertIsNotNone(threads)
305
306        # We should have exactly one thread from threadinfo.
307        self.assertEqual(len(threads), 1)
308
309        # We should have a valid thread_id from $QC.
310        QC_thread_id_hex = context.get("thread_id")
311        self.assertIsNotNone(QC_thread_id_hex)
312        QC_thread_id = int(QC_thread_id_hex, 16)
313
314        # Those two should be the same.
315        self.assertEqual(threads[0], QC_thread_id)
316
317    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
318    @skipIf(dwarf_version=['<', '3'])
319    def test_qThreadInfo_matches_qC_launch(self):
320        self.build()
321        self.set_inferior_startup_launch()
322        self.qThreadInfo_matches_qC()
323
324    @expectedFailureAll(oslist=["windows"]) # expect one more thread stopped
325    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
326    @skipIf(dwarf_version=['<', '3'])
327    def test_qThreadInfo_matches_qC_attach(self):
328        self.build()
329        self.set_inferior_startup_attach()
330        self.qThreadInfo_matches_qC()
331
332    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
333    @skipIf(dwarf_version=['<', '3'])
334    def test_p_returns_correct_data_size_for_each_qRegisterInfo_launch(self):
335        self.build()
336        self.set_inferior_startup_launch()
337        procs = self.prep_debug_monitor_and_inferior()
338        self.add_register_info_collection_packets()
339
340        # Run the packet stream.
341        context = self.expect_gdbremote_sequence()
342        self.assertIsNotNone(context)
343
344        # Gather register info entries.
345        reg_infos = self.parse_register_info_packets(context)
346        self.assertIsNotNone(reg_infos)
347        self.assertTrue(len(reg_infos) > 0)
348
349        byte_order = self.get_target_byte_order()
350
351        # Read value for each register.
352        reg_index = 0
353        for reg_info in reg_infos:
354            # Skip registers that don't have a register set.  For x86, these are
355            # the DRx registers, which have no LLDB-kind register number and thus
356            # cannot be read via normal
357            # NativeRegisterContext::ReadRegister(reg_info,...) calls.
358            if not "set" in reg_info:
359                continue
360
361            # Clear existing packet expectations.
362            self.reset_test_sequence()
363
364            # Run the register query
365            self.test_sequence.add_log_lines(
366                ["read packet: $p{0:x}#00".format(reg_index),
367                 {"direction": "send", "regex": r"^\$([0-9a-fA-F]+)#", "capture": {1: "p_response"}}],
368                True)
369            context = self.expect_gdbremote_sequence()
370            self.assertIsNotNone(context)
371
372            # Verify the response length.
373            p_response = context.get("p_response")
374            self.assertIsNotNone(p_response)
375
376            # Skip erraneous (unsupported) registers.
377            # TODO: remove this once we make unsupported registers disappear.
378            if p_response.startswith("E") and len(p_response) == 3:
379                continue
380
381            if "dynamic_size_dwarf_expr_bytes" in reg_info:
382                self.updateRegInfoBitsize(reg_info, byte_order)
383            self.assertEqual(len(p_response), 2 * int(reg_info["bitsize"]) / 8,
384                             reg_info)
385
386            # Increment loop
387            reg_index += 1
388
389    def Hg_switches_to_3_threads(self, pass_pid=False):
390        # Startup the inferior with three threads (main + 2 new ones).
391        procs = self.prep_debug_monitor_and_inferior(
392            inferior_args=["thread:new", "thread:new"])
393
394        # Let the inferior process have a few moments to start up the thread
395        # when launched.  (The launch scenario has no time to run, so threads
396        # won't be there yet.)
397        self.run_process_then_stop(run_seconds=1)
398
399        # Wait at most x seconds for 3 threads to be present.
400        threads = self.wait_for_thread_count(3)
401        self.assertEqual(len(threads), 3)
402
403        pid_str = ""
404        if pass_pid:
405            pid_str = "p{0:x}.".format(procs["inferior"].pid)
406
407        # verify we can $H to each thead, and $qC matches the thread we set.
408        for thread in threads:
409            # Change to each thread, verify current thread id.
410            self.reset_test_sequence()
411            self.test_sequence.add_log_lines(
412                ["read packet: $Hg{0}{1:x}#00".format(pid_str, thread),  # Set current thread.
413                 "send packet: $OK#00",
414                 "read packet: $qC#00",
415                 {"direction": "send", "regex": r"^\$QC([0-9a-fA-F]+)#", "capture": {1: "thread_id"}}],
416                True)
417
418            context = self.expect_gdbremote_sequence()
419            self.assertIsNotNone(context)
420
421            # Verify the thread id.
422            self.assertIsNotNone(context.get("thread_id"))
423            self.assertEqual(int(context.get("thread_id"), 16), thread)
424
425    @expectedFailureAll(oslist=["windows"]) # expect 4 threads
426    @skipIf(compiler="clang", compiler_version=['<', '11.0'])
427    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
428    @skipIf(dwarf_version=['<', '3'])
429    def test_Hg_switches_to_3_threads_launch(self):
430        self.build()
431        self.set_inferior_startup_launch()
432        self.Hg_switches_to_3_threads()
433
434    @expectedFailureAll(oslist=["windows"]) # expecting one more thread
435    @skipIf(compiler="clang", compiler_version=['<', '11.0'])
436    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
437    @skipIf(dwarf_version=['<', '3'])
438    def test_Hg_switches_to_3_threads_attach(self):
439        self.build()
440        self.set_inferior_startup_attach()
441        self.Hg_switches_to_3_threads()
442
443    @expectedFailureAll(oslist=["windows"]) # expect 4 threads
444    @add_test_categories(["llgs"])
445    @skipIf(compiler="clang", compiler_version=['<', '11.0'])
446    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
447    @skipIf(dwarf_version=['<', '3'])
448    def test_Hg_switches_to_3_threads_attach_pass_correct_pid(self):
449        self.build()
450        self.set_inferior_startup_attach()
451        self.Hg_switches_to_3_threads(pass_pid=True)
452
453    def Hg_fails_on_pid(self, pass_pid):
454        # Start the inferior.
455        procs = self.prep_debug_monitor_and_inferior(
456            inferior_args=["thread:new"])
457
458        self.run_process_then_stop(run_seconds=1)
459
460        threads = self.wait_for_thread_count(2)
461        self.assertEqual(len(threads), 2)
462
463        if pass_pid == -1:
464            pid_str = "p-1."
465        else:
466            pid_str = "p{0:x}.".format(pass_pid)
467        thread = threads[1]
468
469        self.test_sequence.add_log_lines(
470            ["read packet: $Hg{0}{1:x}#00".format(pid_str, thread),  # Set current thread.
471             "send packet: $Eff#00"],
472            True)
473
474        self.expect_gdbremote_sequence()
475
476    @expectedFailureAll(oslist=["windows"])
477    @add_test_categories(["llgs"])
478    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
479    @skipIf(dwarf_version=['<', '3'])
480    def test_Hg_fails_on_another_pid(self):
481        self.build()
482        self.set_inferior_startup_launch()
483        self.Hg_fails_on_pid(1)
484
485    @expectedFailureAll(oslist=["windows"])
486    @add_test_categories(["llgs"])
487    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
488    @skipIf(dwarf_version=['<', '3'])
489    def test_Hg_fails_on_zero_pid(self):
490        self.build()
491        self.set_inferior_startup_launch()
492        self.Hg_fails_on_pid(0)
493
494    @expectedFailureAll(oslist=["windows"])
495    @add_test_categories(["llgs"])
496    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
497    @skipIf(dwarf_version=['<', '3'])
498    def test_Hg_fails_on_minus_one_pid(self):
499        self.build()
500        self.set_inferior_startup_launch()
501        self.Hg_fails_on_pid(-1)
502
503    def Hc_then_Csignal_signals_correct_thread(self, segfault_signo):
504        # NOTE only run this one in inferior-launched mode: we can't grab inferior stdout when running attached,
505        # and the test requires getting stdout from the exe.
506
507        NUM_THREADS = 3
508
509        # Startup the inferior with three threads (main + NUM_THREADS-1 worker threads).
510        # inferior_args=["thread:print-ids"]
511        inferior_args = ["thread:segfault"]
512        for i in range(NUM_THREADS - 1):
513            # if i > 0:
514                # Give time between thread creation/segfaulting for the handler to work.
515                # inferior_args.append("sleep:1")
516            inferior_args.append("thread:new")
517        inferior_args.append("sleep:10")
518
519        # Launch/attach.  (In our case, this should only ever be launched since
520        # we need inferior stdout/stderr).
521        procs = self.prep_debug_monitor_and_inferior(
522            inferior_args=inferior_args)
523        self.test_sequence.add_log_lines(["read packet: $c#63"], True)
524        context = self.expect_gdbremote_sequence()
525
526        # Let the inferior process have a few moments to start up the thread when launched.
527        # context = self.run_process_then_stop(run_seconds=1)
528
529        # Wait at most x seconds for all threads to be present.
530        # threads = self.wait_for_thread_count(NUM_THREADS)
531        # self.assertEquals(len(threads), NUM_THREADS)
532
533        signaled_tids = {}
534        print_thread_ids = {}
535
536        # Switch to each thread, deliver a signal, and verify signal delivery
537        for i in range(NUM_THREADS - 1):
538            # Run until SIGSEGV comes in.
539            self.reset_test_sequence()
540            self.test_sequence.add_log_lines([{"direction": "send",
541                                               "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);",
542                                               "capture": {1: "signo",
543                                                            2: "thread_id"}}],
544                                             True)
545
546            context = self.expect_gdbremote_sequence()
547            self.assertIsNotNone(context)
548            signo = context.get("signo")
549            self.assertEqual(int(signo, 16), segfault_signo)
550
551            # Ensure we haven't seen this tid yet.
552            thread_id = int(context.get("thread_id"), 16)
553            self.assertNotIn(thread_id, signaled_tids)
554            signaled_tids[thread_id] = 1
555
556            # Send SIGUSR1 to the thread that signaled the SIGSEGV.
557            self.reset_test_sequence()
558            self.test_sequence.add_log_lines(
559                [
560                    # Set the continue thread.
561                    # Set current thread.
562                    "read packet: $Hc{0:x}#00".format(thread_id),
563                    "send packet: $OK#00",
564
565                    # Continue sending the signal number to the continue thread.
566                    # The commented out packet is a way to do this same operation without using
567                    # a $Hc (but this test is testing $Hc, so we'll stick with the former).
568                    "read packet: $C{0:x}#00".format(lldbutil.get_signal_number('SIGUSR1')),
569                    # "read packet: $vCont;C{0:x}:{1:x};c#00".format(lldbutil.get_signal_number('SIGUSR1'), thread_id),
570
571                    # FIXME: Linux does not report the thread stop on the delivered signal (SIGUSR1 here).  MacOSX debugserver does.
572                    # But MacOSX debugserver isn't guaranteeing the thread the signal handler runs on, so currently its an XFAIL.
573                    # Need to rectify behavior here.  The linux behavior is more intuitive to me since we're essentially swapping out
574                    # an about-to-be-delivered signal (for which we already sent a stop packet) to a different signal.
575                    # {"direction":"send", "regex":r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture":{1:"stop_signo", 2:"stop_thread_id"} },
576                    #  "read packet: $c#63",
577                    {"type": "output_match", "regex": r"^received SIGUSR1 on thread id: ([0-9a-fA-F]+)\r\nthread ([0-9a-fA-F]+): past SIGSEGV\r\n", "capture": {1: "print_thread_id", 2: "post_handle_thread_id"}},
578                ],
579                True)
580
581            # Run the sequence.
582            context = self.expect_gdbremote_sequence()
583            self.assertIsNotNone(context)
584
585            # Ensure the stop signal is the signal we delivered.
586            # stop_signo = context.get("stop_signo")
587            # self.assertIsNotNone(stop_signo)
588            # self.assertEquals(int(stop_signo,16), lldbutil.get_signal_number('SIGUSR1'))
589
590            # Ensure the stop thread is the thread to which we delivered the signal.
591            # stop_thread_id = context.get("stop_thread_id")
592            # self.assertIsNotNone(stop_thread_id)
593            # self.assertEquals(int(stop_thread_id,16), thread_id)
594
595            # Ensure we haven't seen this thread id yet.  The inferior's
596            # self-obtained thread ids are not guaranteed to match the stub
597            # tids (at least on MacOSX).
598            print_thread_id = context.get("print_thread_id")
599            self.assertIsNotNone(print_thread_id)
600            print_thread_id = int(print_thread_id, 16)
601            self.assertNotIn(print_thread_id, print_thread_ids)
602
603            # Now remember this print (i.e. inferior-reflected) thread id and
604            # ensure we don't hit it again.
605            print_thread_ids[print_thread_id] = 1
606
607            # Ensure post signal-handle thread id matches the thread that
608            # initially raised the SIGSEGV.
609            post_handle_thread_id = context.get("post_handle_thread_id")
610            self.assertIsNotNone(post_handle_thread_id)
611            post_handle_thread_id = int(post_handle_thread_id, 16)
612            self.assertEqual(post_handle_thread_id, print_thread_id)
613
614    @expectedFailureDarwin
615    @skipIfWindows # no SIGSEGV support
616    @expectedFailureAll(oslist=["freebsd"], bugnumber="llvm.org/pr48419")
617    @expectedFailureNetBSD
618    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
619    @skipIf(dwarf_version=['<', '3'])
620    def test_Hc_then_Csignal_signals_correct_thread_launch(self):
621        self.build()
622        self.set_inferior_startup_launch()
623
624        if self.platformIsDarwin():
625            # Darwin debugserver translates some signals like SIGSEGV into some gdb
626            # expectations about fixed signal numbers.
627            self.Hc_then_Csignal_signals_correct_thread(self.TARGET_EXC_BAD_ACCESS)
628        else:
629            self.Hc_then_Csignal_signals_correct_thread(
630                lldbutil.get_signal_number('SIGSEGV'))
631
632    @skipIfWindows # No pty support to test any inferior output
633    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
634    @skipIf(dwarf_version=['<', '3'])
635    def test_m_packet_reads_memory(self):
636        self.build()
637        self.set_inferior_startup_launch()
638        # This is the memory we will write into the inferior and then ensure we
639        # can read back with $m.
640        MEMORY_CONTENTS = "Test contents 0123456789 ABCDEFGHIJKLMNOPQRSTUVWXYZ abcdefghijklmnopqrstuvwxyz"
641
642        # Start up the inferior.
643        procs = self.prep_debug_monitor_and_inferior(
644            inferior_args=[
645                "set-message:%s" %
646                MEMORY_CONTENTS,
647                "get-data-address-hex:g_message",
648                "sleep:5"])
649
650        # Run the process
651        self.test_sequence.add_log_lines(
652            [
653                # Start running after initial stop.
654                "read packet: $c#63",
655                # Match output line that prints the memory address of the message buffer within the inferior.
656                # Note we require launch-only testing so we can get inferior otuput.
657                {"type": "output_match", "regex": self.maybe_strict_output_regex(r"data address: 0x([0-9a-fA-F]+)\r\n"),
658                 "capture": {1: "message_address"}},
659                # Now stop the inferior.
660                "read packet: {}".format(chr(3)),
661                # And wait for the stop notification.
662                {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}],
663            True)
664
665        # Run the packet stream.
666        context = self.expect_gdbremote_sequence()
667        self.assertIsNotNone(context)
668
669        # Grab the message address.
670        self.assertIsNotNone(context.get("message_address"))
671        message_address = int(context.get("message_address"), 16)
672
673        # Grab contents from the inferior.
674        self.reset_test_sequence()
675        self.test_sequence.add_log_lines(
676            ["read packet: $m{0:x},{1:x}#00".format(message_address, len(MEMORY_CONTENTS)),
677             {"direction": "send", "regex": r"^\$(.+)#[0-9a-fA-F]{2}$", "capture": {1: "read_contents"}}],
678            True)
679
680        # Run the packet stream.
681        context = self.expect_gdbremote_sequence()
682        self.assertIsNotNone(context)
683
684        # Ensure what we read from inferior memory is what we wrote.
685        self.assertIsNotNone(context.get("read_contents"))
686        read_contents = seven.unhexlify(context.get("read_contents"))
687        self.assertEqual(read_contents, MEMORY_CONTENTS)
688
689    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
690    @skipIf(dwarf_version=['<', '3'])
691    def test_qMemoryRegionInfo_is_supported(self):
692        self.build()
693        self.set_inferior_startup_launch()
694        # Start up the inferior.
695        procs = self.prep_debug_monitor_and_inferior()
696
697        # Ask if it supports $qMemoryRegionInfo.
698        self.test_sequence.add_log_lines(
699            ["read packet: $qMemoryRegionInfo#00",
700             "send packet: $OK#00"
701             ], True)
702        self.expect_gdbremote_sequence()
703
704    @skipIfWindows # No pty support to test any inferior output
705    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
706    @skipIf(dwarf_version=['<', '3'])
707    def test_qMemoryRegionInfo_reports_code_address_as_executable(self):
708        self.build()
709        self.set_inferior_startup_launch()
710
711        # Start up the inferior.
712        procs = self.prep_debug_monitor_and_inferior(
713            inferior_args=["get-code-address-hex:hello", "sleep:5"])
714
715        # Run the process
716        self.test_sequence.add_log_lines(
717            [
718                # Start running after initial stop.
719                "read packet: $c#63",
720                # Match output line that prints the memory address of the message buffer within the inferior.
721                # Note we require launch-only testing so we can get inferior otuput.
722                {"type": "output_match", "regex": self.maybe_strict_output_regex(r"code address: 0x([0-9a-fA-F]+)\r\n"),
723                 "capture": {1: "code_address"}},
724                # Now stop the inferior.
725                "read packet: {}".format(chr(3)),
726                # And wait for the stop notification.
727                {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}],
728            True)
729
730        # Run the packet stream.
731        context = self.expect_gdbremote_sequence()
732        self.assertIsNotNone(context)
733
734        # Grab the code address.
735        self.assertIsNotNone(context.get("code_address"))
736        code_address = int(context.get("code_address"), 16)
737
738        # Grab memory region info from the inferior.
739        self.reset_test_sequence()
740        self.add_query_memory_region_packets(code_address)
741
742        # Run the packet stream.
743        context = self.expect_gdbremote_sequence()
744        self.assertIsNotNone(context)
745        mem_region_dict = self.parse_memory_region_packet(context)
746
747        # Ensure there are no errors reported.
748        self.assertNotIn("error", mem_region_dict)
749
750        # Ensure code address is readable and executable.
751        self.assertIn("permissions", mem_region_dict)
752        self.assertIn("r", mem_region_dict["permissions"])
753        self.assertIn("x", mem_region_dict["permissions"])
754
755        # Ensure the start address and size encompass the address we queried.
756        self.assert_address_within_memory_region(code_address, mem_region_dict)
757
758    @skipIfWindows # No pty support to test any inferior output
759    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
760    @skipIf(dwarf_version=['<', '3'])
761    def test_qMemoryRegionInfo_reports_stack_address_as_rw(self):
762        self.build()
763        self.set_inferior_startup_launch()
764
765        # Start up the inferior.
766        procs = self.prep_debug_monitor_and_inferior(
767            inferior_args=["get-stack-address-hex:", "sleep:5"])
768
769        # Run the process
770        self.test_sequence.add_log_lines(
771            [
772                # Start running after initial stop.
773                "read packet: $c#63",
774                # Match output line that prints the memory address of the message buffer within the inferior.
775                # Note we require launch-only testing so we can get inferior otuput.
776                {"type": "output_match", "regex": self.maybe_strict_output_regex(r"stack address: 0x([0-9a-fA-F]+)\r\n"),
777                 "capture": {1: "stack_address"}},
778                # Now stop the inferior.
779                "read packet: {}".format(chr(3)),
780                # And wait for the stop notification.
781                {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}],
782            True)
783
784        # Run the packet stream.
785        context = self.expect_gdbremote_sequence()
786        self.assertIsNotNone(context)
787
788        # Grab the address.
789        self.assertIsNotNone(context.get("stack_address"))
790        stack_address = int(context.get("stack_address"), 16)
791
792        # Grab memory region info from the inferior.
793        self.reset_test_sequence()
794        self.add_query_memory_region_packets(stack_address)
795
796        # Run the packet stream.
797        context = self.expect_gdbremote_sequence()
798        self.assertIsNotNone(context)
799        mem_region_dict = self.parse_memory_region_packet(context)
800
801        # Ensure there are no errors reported.
802        self.assertNotIn("error", mem_region_dict)
803
804        # Ensure address is readable and executable.
805        self.assertIn("permissions", mem_region_dict)
806        self.assertIn("r", mem_region_dict["permissions"])
807        self.assertIn("w", mem_region_dict["permissions"])
808
809        # Ensure the start address and size encompass the address we queried.
810        self.assert_address_within_memory_region(
811            stack_address, mem_region_dict)
812
813    @skipIfWindows # No pty support to test any inferior output
814    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
815    @skipIf(dwarf_version=['<', '3'])
816    def test_qMemoryRegionInfo_reports_heap_address_as_rw(self):
817        self.build()
818        self.set_inferior_startup_launch()
819
820        # Start up the inferior.
821        procs = self.prep_debug_monitor_and_inferior(
822            inferior_args=["get-heap-address-hex:", "sleep:5"])
823
824        # Run the process
825        self.test_sequence.add_log_lines(
826            [
827                # Start running after initial stop.
828                "read packet: $c#63",
829                # Match output line that prints the memory address of the message buffer within the inferior.
830                # Note we require launch-only testing so we can get inferior otuput.
831                {"type": "output_match", "regex": self.maybe_strict_output_regex(r"heap address: 0x([0-9a-fA-F]+)\r\n"),
832                 "capture": {1: "heap_address"}},
833                # Now stop the inferior.
834                "read packet: {}".format(chr(3)),
835                # And wait for the stop notification.
836                {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}],
837            True)
838
839        # Run the packet stream.
840        context = self.expect_gdbremote_sequence()
841        self.assertIsNotNone(context)
842
843        # Grab the address.
844        self.assertIsNotNone(context.get("heap_address"))
845        heap_address = int(context.get("heap_address"), 16)
846
847        # Grab memory region info from the inferior.
848        self.reset_test_sequence()
849        self.add_query_memory_region_packets(heap_address)
850
851        # Run the packet stream.
852        context = self.expect_gdbremote_sequence()
853        self.assertIsNotNone(context)
854        mem_region_dict = self.parse_memory_region_packet(context)
855
856        # Ensure there are no errors reported.
857        self.assertNotIn("error", mem_region_dict)
858
859        # Ensure address is readable and executable.
860        self.assertIn("permissions", mem_region_dict)
861        self.assertIn("r", mem_region_dict["permissions"])
862        self.assertIn("w", mem_region_dict["permissions"])
863
864        # Ensure the start address and size encompass the address we queried.
865        self.assert_address_within_memory_region(heap_address, mem_region_dict)
866
867    def breakpoint_set_and_remove_work(self, want_hardware):
868        # Start up the inferior.
869        procs = self.prep_debug_monitor_and_inferior(
870            inferior_args=[
871                "get-code-address-hex:hello",
872                "sleep:1",
873                "call-function:hello"])
874
875        # Run the process
876        self.add_register_info_collection_packets()
877        self.add_process_info_collection_packets()
878        self.test_sequence.add_log_lines(
879            [  # Start running after initial stop.
880                "read packet: $c#63",
881                # Match output line that prints the memory address of the function call entry point.
882                # Note we require launch-only testing so we can get inferior otuput.
883                {"type": "output_match", "regex": self.maybe_strict_output_regex(r"code address: 0x([0-9a-fA-F]+)\r\n"),
884                 "capture": {1: "function_address"}},
885                # Now stop the inferior.
886                "read packet: {}".format(chr(3)),
887                # And wait for the stop notification.
888                {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}],
889            True)
890
891        # Run the packet stream.
892        context = self.expect_gdbremote_sequence()
893        self.assertIsNotNone(context)
894
895        # Gather process info - we need endian of target to handle register
896        # value conversions.
897        process_info = self.parse_process_info_response(context)
898        endian = process_info.get("endian")
899        self.assertIsNotNone(endian)
900
901        # Gather register info entries.
902        reg_infos = self.parse_register_info_packets(context)
903        (pc_lldb_reg_index, pc_reg_info) = self.find_pc_reg_info(reg_infos)
904        self.assertIsNotNone(pc_lldb_reg_index)
905        self.assertIsNotNone(pc_reg_info)
906
907        # Grab the function address.
908        self.assertIsNotNone(context.get("function_address"))
909        function_address = int(context.get("function_address"), 16)
910
911        # Get current target architecture
912        target_arch = self.getArchitecture()
913
914        # Set the breakpoint.
915        if (target_arch == "arm") or (target_arch == "aarch64"):
916            # TODO: Handle case when setting breakpoint in thumb code
917            BREAKPOINT_KIND = 4
918        else:
919            BREAKPOINT_KIND = 1
920
921        # Set default packet type to Z0 (software breakpoint)
922        z_packet_type = 0
923
924        # If hardware breakpoint is requested set packet type to Z1
925        if want_hardware == True:
926            z_packet_type = 1
927
928        self.reset_test_sequence()
929        self.add_set_breakpoint_packets(
930            function_address,
931            z_packet_type,
932            do_continue=True,
933            breakpoint_kind=BREAKPOINT_KIND)
934
935        # Run the packet stream.
936        context = self.expect_gdbremote_sequence()
937        self.assertIsNotNone(context)
938
939        # Verify the stop signal reported was the breakpoint signal number.
940        stop_signo = context.get("stop_signo")
941        self.assertIsNotNone(stop_signo)
942        self.assertEqual(int(stop_signo, 16),
943                         lldbutil.get_signal_number('SIGTRAP'))
944
945        # Ensure we did not receive any output.  If the breakpoint was not set, we would
946        # see output (from a launched process with captured stdio) printing a hello, world message.
947        # That would indicate the breakpoint didn't take.
948        self.assertEqual(len(context["O_content"]), 0)
949
950        # Verify that the PC for the main thread is where we expect it - right at the breakpoint address.
951        # This acts as a another validation on the register reading code.
952        self.reset_test_sequence()
953        self.test_sequence.add_log_lines(
954            [
955                # Print the PC.  This should match the breakpoint address.
956                "read packet: $p{0:x}#00".format(pc_lldb_reg_index),
957                # Capture $p results.
958                {"direction": "send",
959                 "regex": r"^\$([0-9a-fA-F]+)#",
960                 "capture": {1: "p_response"}},
961            ], True)
962
963        context = self.expect_gdbremote_sequence()
964        self.assertIsNotNone(context)
965
966        # Verify the PC is where we expect.  Note response is in endianness of
967        # the inferior.
968        p_response = context.get("p_response")
969        self.assertIsNotNone(p_response)
970
971        # Convert from target endian to int.
972        returned_pc = lldbgdbserverutils.unpack_register_hex_unsigned(
973            endian, p_response)
974        self.assertEqual(returned_pc, function_address)
975
976        # Verify that a breakpoint remove and continue gets us the expected
977        # output.
978        self.reset_test_sequence()
979
980        # Add breakpoint remove packets
981        self.add_remove_breakpoint_packets(
982            function_address,
983            z_packet_type,
984            breakpoint_kind=BREAKPOINT_KIND)
985
986        self.test_sequence.add_log_lines(
987            [
988                # Continue running.
989                "read packet: $c#63",
990                # We should now receive the output from the call.
991                {"type": "output_match", "regex": r"^hello, world\r\n$"},
992                # And wait for program completion.
993                {"direction": "send", "regex": r"^\$W00(.*)#[0-9a-fA-F]{2}$"},
994            ], True)
995
996        context = self.expect_gdbremote_sequence()
997        self.assertIsNotNone(context)
998
999    @skipIfWindows # No pty support to test any inferior output
1000    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1001    @skipIf(dwarf_version=['<', '3'])
1002    def test_software_breakpoint_set_and_remove_work(self):
1003        if self.getArchitecture() == "arm":
1004            # TODO: Handle case when setting breakpoint in thumb code
1005            self.build(dictionary={'CFLAGS_EXTRAS': '-marm'})
1006        else:
1007            self.build()
1008        self.set_inferior_startup_launch()
1009        self.breakpoint_set_and_remove_work(want_hardware=False)
1010
1011    @skipUnlessPlatform(oslist=['linux'])
1012    @skipIf(archs=no_match(['arm', 'aarch64']))
1013    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1014    @skipIf(dwarf_version=['<', '3'])
1015    def test_hardware_breakpoint_set_and_remove_work(self):
1016        if self.getArchitecture() == "arm":
1017            # TODO: Handle case when setting breakpoint in thumb code
1018            self.build(dictionary={'CFLAGS_EXTRAS': '-marm'})
1019        else:
1020            self.build()
1021        self.set_inferior_startup_launch()
1022        self.breakpoint_set_and_remove_work(want_hardware=True)
1023
1024    def get_qSupported_dict(self, features=[]):
1025        self.build()
1026        self.set_inferior_startup_launch()
1027
1028        # Start up the stub and start/prep the inferior.
1029        procs = self.prep_debug_monitor_and_inferior()
1030        self.add_qSupported_packets(features)
1031
1032        # Run the packet stream.
1033        context = self.expect_gdbremote_sequence()
1034        self.assertIsNotNone(context)
1035
1036        # Retrieve the qSupported features.
1037        return self.parse_qSupported_response(context)
1038
1039    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1040    @skipIf(dwarf_version=['<', '3'])
1041    def test_qSupported_returns_known_stub_features(self):
1042        supported_dict = self.get_qSupported_dict()
1043        self.assertIsNotNone(supported_dict)
1044        self.assertTrue(len(supported_dict) > 0)
1045
1046    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1047    @skipIf(dwarf_version=['<','3'])
1048    def test_qSupported_auvx(self):
1049        expected = ('+' if lldbplatformutil.getPlatform()
1050                    in ["freebsd", "linux", "netbsd"] else '-')
1051        supported_dict = self.get_qSupported_dict()
1052        self.assertEqual(supported_dict.get('qXfer:auxv:read', '-'), expected)
1053
1054    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1055    @skipIf(dwarf_version=['<', '3'])
1056    def test_qSupported_libraries_svr4(self):
1057        expected = ('+' if lldbplatformutil.getPlatform()
1058                    in ["freebsd", "linux", "netbsd"] else '-')
1059        supported_dict = self.get_qSupported_dict()
1060        self.assertEqual(supported_dict.get('qXfer:libraries-svr4:read', '-'),
1061                         expected)
1062
1063    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1064    @skipIf(dwarf_version=['<', '3'])
1065    def test_qSupported_siginfo_read(self):
1066        expected = ('+' if lldbplatformutil.getPlatform()
1067                    in ["freebsd", "linux"] else '-')
1068        supported_dict = self.get_qSupported_dict()
1069        self.assertEqual(supported_dict.get('qXfer:siginfo:read', '-'),
1070                         expected)
1071
1072    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1073    @skipIf(dwarf_version=['<', '3'])
1074    def test_qSupported_QPassSignals(self):
1075        expected = ('+' if lldbplatformutil.getPlatform()
1076                    in ["freebsd", "linux", "netbsd"] else '-')
1077        supported_dict = self.get_qSupported_dict()
1078        self.assertEqual(supported_dict.get('QPassSignals', '-'), expected)
1079
1080    @add_test_categories(["fork"])
1081    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1082    @skipIf(dwarf_version=['<', '3'])
1083    def test_qSupported_fork_events(self):
1084        supported_dict = (
1085            self.get_qSupported_dict(['multiprocess+', 'fork-events+']))
1086        self.assertEqual(supported_dict.get('multiprocess', '-'), '+')
1087        self.assertEqual(supported_dict.get('fork-events', '-'), '+')
1088        self.assertEqual(supported_dict.get('vfork-events', '-'), '-')
1089
1090    @add_test_categories(["fork"])
1091    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1092    @skipIf(dwarf_version=['<', '3'])
1093    def test_qSupported_fork_events_without_multiprocess(self):
1094        supported_dict = (
1095            self.get_qSupported_dict(['fork-events+']))
1096        self.assertEqual(supported_dict.get('multiprocess', '-'), '-')
1097        self.assertEqual(supported_dict.get('fork-events', '-'), '-')
1098        self.assertEqual(supported_dict.get('vfork-events', '-'), '-')
1099
1100    @add_test_categories(["fork"])
1101    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1102    @skipIf(dwarf_version=['<', '3'])
1103    def test_qSupported_vfork_events(self):
1104        supported_dict = (
1105            self.get_qSupported_dict(['multiprocess+', 'vfork-events+']))
1106        self.assertEqual(supported_dict.get('multiprocess', '-'), '+')
1107        self.assertEqual(supported_dict.get('fork-events', '-'), '-')
1108        self.assertEqual(supported_dict.get('vfork-events', '-'), '+')
1109
1110    @add_test_categories(["fork"])
1111    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1112    @skipIf(dwarf_version=['<', '3'])
1113    def test_qSupported_vfork_events_without_multiprocess(self):
1114        supported_dict = (
1115            self.get_qSupported_dict(['vfork-events+']))
1116        self.assertEqual(supported_dict.get('multiprocess', '-'), '-')
1117        self.assertEqual(supported_dict.get('fork-events', '-'), '-')
1118        self.assertEqual(supported_dict.get('vfork-events', '-'), '-')
1119
1120    # We need to be able to self.runCmd to get cpuinfo,
1121    # which is not possible when using a remote platform.
1122    @skipIfRemote
1123    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1124    @skipIf(dwarf_version=['<', '3'])
1125    def test_qSupported_memory_tagging(self):
1126        supported_dict = self.get_qSupported_dict()
1127        self.assertEqual(supported_dict.get("memory-tagging", '-'),
1128                         '+' if self.isAArch64MTE() else '-')
1129
1130    @skipIfWindows # No pty support to test any inferior output
1131    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1132    @skipIf(dwarf_version=['<', '3'])
1133    def test_written_M_content_reads_back_correctly(self):
1134        self.build()
1135        self.set_inferior_startup_launch()
1136
1137        TEST_MESSAGE = "Hello, memory"
1138
1139        # Start up the stub and start/prep the inferior.
1140        procs = self.prep_debug_monitor_and_inferior(
1141            inferior_args=[
1142                "set-message:xxxxxxxxxxxxxX",
1143                "get-data-address-hex:g_message",
1144                "sleep:1",
1145                "print-message:"])
1146        self.test_sequence.add_log_lines(
1147            [
1148                # Start running after initial stop.
1149                "read packet: $c#63",
1150                # Match output line that prints the memory address of the message buffer within the inferior.
1151                # Note we require launch-only testing so we can get inferior otuput.
1152                {"type": "output_match", "regex": self.maybe_strict_output_regex(r"data address: 0x([0-9a-fA-F]+)\r\n"),
1153                 "capture": {1: "message_address"}},
1154                # Now stop the inferior.
1155                "read packet: {}".format(chr(3)),
1156                # And wait for the stop notification.
1157                {"direction": "send", "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);", "capture": {1: "stop_signo", 2: "stop_thread_id"}}],
1158            True)
1159        context = self.expect_gdbremote_sequence()
1160        self.assertIsNotNone(context)
1161
1162        # Grab the message address.
1163        self.assertIsNotNone(context.get("message_address"))
1164        message_address = int(context.get("message_address"), 16)
1165
1166        # Hex-encode the test message, adding null termination.
1167        hex_encoded_message = seven.hexlify(TEST_MESSAGE)
1168
1169        # Write the message to the inferior. Verify that we can read it with the hex-encoded (m)
1170        # and binary (x) memory read packets.
1171        self.reset_test_sequence()
1172        self.test_sequence.add_log_lines(
1173            ["read packet: $M{0:x},{1:x}:{2}#00".format(message_address, len(TEST_MESSAGE), hex_encoded_message),
1174             "send packet: $OK#00",
1175             "read packet: $m{0:x},{1:x}#00".format(message_address, len(TEST_MESSAGE)),
1176             "send packet: ${0}#00".format(hex_encoded_message),
1177             "read packet: $x{0:x},{1:x}#00".format(message_address, len(TEST_MESSAGE)),
1178             "send packet: ${0}#00".format(TEST_MESSAGE),
1179             "read packet: $m{0:x},4#00".format(message_address),
1180             "send packet: ${0}#00".format(hex_encoded_message[0:8]),
1181             "read packet: $x{0:x},4#00".format(message_address),
1182             "send packet: ${0}#00".format(TEST_MESSAGE[0:4]),
1183             "read packet: $c#63",
1184             {"type": "output_match", "regex": r"^message: (.+)\r\n$", "capture": {1: "printed_message"}},
1185             "send packet: $W00#00",
1186             ], True)
1187        context = self.expect_gdbremote_sequence()
1188        self.assertIsNotNone(context)
1189
1190        # Ensure what we read from inferior memory is what we wrote.
1191        printed_message = context.get("printed_message")
1192        self.assertIsNotNone(printed_message)
1193        self.assertEqual(printed_message, TEST_MESSAGE + "X")
1194
1195    # Note: as of this moment, a hefty number of the GPR writes are failing with E32 (everything except rax-rdx, rdi, rsi, rbp).
1196    # Come back to this.  I have the test rigged to verify that at least some
1197    # of the bit-flip writes work.
1198    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1199    @skipIf(dwarf_version=['<', '3'])
1200    def test_P_writes_all_gpr_registers(self):
1201        self.build()
1202        self.set_inferior_startup_launch()
1203
1204        # Start inferior debug session, grab all register info.
1205        procs = self.prep_debug_monitor_and_inferior(inferior_args=["sleep:2"])
1206        self.add_register_info_collection_packets()
1207        self.add_process_info_collection_packets()
1208
1209        context = self.expect_gdbremote_sequence()
1210        self.assertIsNotNone(context)
1211
1212        # Process register infos.
1213        reg_infos = self.parse_register_info_packets(context)
1214        self.assertIsNotNone(reg_infos)
1215        self.add_lldb_register_index(reg_infos)
1216
1217        # Process endian.
1218        process_info = self.parse_process_info_response(context)
1219        endian = process_info.get("endian")
1220        self.assertIsNotNone(endian)
1221
1222        # Pull out the register infos that we think we can bit flip
1223        # successfully,.
1224        gpr_reg_infos = [
1225            reg_info for reg_info in reg_infos if self.is_bit_flippable_register(reg_info)]
1226        self.assertTrue(len(gpr_reg_infos) > 0)
1227
1228        # Write flipped bit pattern of existing value to each register.
1229        (successful_writes, failed_writes) = self.flip_all_bits_in_each_register_value(
1230            gpr_reg_infos, endian)
1231        self.trace("successful writes: {}, failed writes: {}".format(successful_writes, failed_writes))
1232        self.assertTrue(successful_writes > 0)
1233
1234    # Note: as of this moment, a hefty number of the GPR writes are failing
1235    # with E32 (everything except rax-rdx, rdi, rsi, rbp).
1236    @skipIfWindows
1237    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1238    @skipIf(dwarf_version=['<', '3'])
1239    def test_P_and_p_thread_suffix_work(self):
1240        self.build()
1241        self.set_inferior_startup_launch()
1242
1243        # Startup the inferior with three threads.
1244        procs = self.prep_debug_monitor_and_inferior(
1245            inferior_args=["thread:new", "thread:new"])
1246        self.add_thread_suffix_request_packets()
1247        self.add_register_info_collection_packets()
1248        self.add_process_info_collection_packets()
1249
1250        context = self.expect_gdbremote_sequence()
1251        self.assertIsNotNone(context)
1252
1253        process_info = self.parse_process_info_response(context)
1254        self.assertIsNotNone(process_info)
1255        endian = process_info.get("endian")
1256        self.assertIsNotNone(endian)
1257
1258        reg_infos = self.parse_register_info_packets(context)
1259        self.assertIsNotNone(reg_infos)
1260        self.add_lldb_register_index(reg_infos)
1261
1262        reg_index = self.select_modifiable_register(reg_infos)
1263        self.assertIsNotNone(reg_index)
1264        reg_byte_size = int(reg_infos[reg_index]["bitsize"]) // 8
1265        self.assertTrue(reg_byte_size > 0)
1266
1267        # Run the process a bit so threads can start up, and collect register
1268        # info.
1269        context = self.run_process_then_stop(run_seconds=1)
1270        self.assertIsNotNone(context)
1271
1272        # Wait for 3 threads to be present.
1273        threads = self.wait_for_thread_count(3)
1274        self.assertEqual(len(threads), 3)
1275
1276        expected_reg_values = []
1277        register_increment = 1
1278        next_value = None
1279
1280        # Set the same register in each of 3 threads to a different value.
1281        # Verify each one has the unique value.
1282        for thread in threads:
1283            # If we don't have a next value yet, start it with the initial read
1284            # value + 1
1285            if not next_value:
1286                # Read pre-existing register value.
1287                self.reset_test_sequence()
1288                self.test_sequence.add_log_lines(
1289                    ["read packet: $p{0:x};thread:{1:x}#00".format(reg_index, thread),
1290                     {"direction": "send", "regex": r"^\$([0-9a-fA-F]+)#", "capture": {1: "p_response"}},
1291                     ], True)
1292                context = self.expect_gdbremote_sequence()
1293                self.assertIsNotNone(context)
1294
1295                # Set the next value to use for writing as the increment plus
1296                # current value.
1297                p_response = context.get("p_response")
1298                self.assertIsNotNone(p_response)
1299                next_value = lldbgdbserverutils.unpack_register_hex_unsigned(
1300                    endian, p_response)
1301
1302            # Set new value using P and thread suffix.
1303            self.reset_test_sequence()
1304            self.test_sequence.add_log_lines(
1305                [
1306                    "read packet: $P{0:x}={1};thread:{2:x}#00".format(
1307                        reg_index,
1308                        lldbgdbserverutils.pack_register_hex(
1309                            endian,
1310                            next_value,
1311                            byte_size=reg_byte_size),
1312                        thread),
1313                    "send packet: $OK#00",
1314                ],
1315                True)
1316            context = self.expect_gdbremote_sequence()
1317            self.assertIsNotNone(context)
1318
1319            # Save the value we set.
1320            expected_reg_values.append(next_value)
1321
1322            # Increment value for next thread to use (we want them all
1323            # different so we can verify they wrote to each thread correctly
1324            # next.)
1325            next_value += register_increment
1326
1327        # Revisit each thread and verify they have the expected value set for
1328        # the register we wrote.
1329        thread_index = 0
1330        for thread in threads:
1331            # Read pre-existing register value.
1332            self.reset_test_sequence()
1333            self.test_sequence.add_log_lines(
1334                ["read packet: $p{0:x};thread:{1:x}#00".format(reg_index, thread),
1335                 {"direction": "send", "regex": r"^\$([0-9a-fA-F]+)#", "capture": {1: "p_response"}},
1336                 ], True)
1337            context = self.expect_gdbremote_sequence()
1338            self.assertIsNotNone(context)
1339
1340            # Get the register value.
1341            p_response = context.get("p_response")
1342            self.assertIsNotNone(p_response)
1343            read_value = lldbgdbserverutils.unpack_register_hex_unsigned(
1344                endian, p_response)
1345
1346            # Make sure we read back what we wrote.
1347            self.assertEqual(read_value, expected_reg_values[thread_index])
1348            thread_index += 1
1349
1350    @skipIfWindows # No pty support to test any inferior output
1351    @add_test_categories(["llgs"])
1352    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1353    @skipIf(dwarf_version=['<', '3'])
1354    def test_launch_via_A(self):
1355        self.build()
1356        exe_path = self.getBuildArtifact("a.out")
1357        args = [exe_path, "stderr:arg1", "stderr:arg2", "stderr:arg3"]
1358        hex_args = [binascii.b2a_hex(x.encode()).decode() for x in args]
1359
1360        server = self.connect_to_debug_monitor()
1361        self.assertIsNotNone(server)
1362        self.do_handshake()
1363        # NB: strictly speaking we should use %x here but this packet
1364        # is deprecated, so no point in changing lldb-server's expectations
1365        self.test_sequence.add_log_lines(
1366            ["read packet: $A %d,0,%s,%d,1,%s,%d,2,%s,%d,3,%s#00" %
1367             tuple(itertools.chain.from_iterable(
1368                 [(len(x), x) for x in hex_args])),
1369             "send packet: $OK#00",
1370             "read packet: $c#00",
1371             "send packet: $W00#00"],
1372            True)
1373        context = self.expect_gdbremote_sequence()
1374        self.assertEqual(context["O_content"],
1375                         b'arg1\r\narg2\r\narg3\r\n')
1376
1377    @skipIfWindows # No pty support to test any inferior output
1378    @add_test_categories(["llgs"])
1379    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1380    @skipIf(dwarf_version=['<','3'])
1381    def test_launch_via_vRun(self):
1382        self.build()
1383        exe_path = self.getBuildArtifact("a.out")
1384        args = [exe_path, "stderr:arg1", "stderr:arg2", "stderr:arg3"]
1385        hex_args = [binascii.b2a_hex(x.encode()).decode() for x in args]
1386
1387        server = self.connect_to_debug_monitor()
1388        self.assertIsNotNone(server)
1389        self.do_handshake()
1390        self.test_sequence.add_log_lines(
1391            ["read packet: $vRun;%s;%s;%s;%s#00" % tuple(hex_args),
1392             {"direction": "send",
1393              "regex": r"^\$T([0-9a-fA-F]+)"},
1394             "read packet: $c#00",
1395             "send packet: $W00#00"],
1396            True)
1397        context = self.expect_gdbremote_sequence()
1398        self.assertEqual(context["O_content"],
1399                         b'arg1\r\narg2\r\narg3\r\n')
1400
1401    @add_test_categories(["llgs"])
1402    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1403    @skipIf(dwarf_version=['<', '3'])
1404    def test_launch_via_vRun_no_args(self):
1405        self.build()
1406        exe_path = self.getBuildArtifact("a.out")
1407        hex_path = binascii.b2a_hex(exe_path.encode()).decode()
1408
1409        server = self.connect_to_debug_monitor()
1410        self.assertIsNotNone(server)
1411        self.do_handshake()
1412        self.test_sequence.add_log_lines(
1413            ["read packet: $vRun;%s#00" % (hex_path,),
1414             {"direction": "send",
1415              "regex": r"^\$T([0-9a-fA-F]+)"},
1416             "read packet: $c#00",
1417             "send packet: $W00#00"],
1418            True)
1419        self.expect_gdbremote_sequence()
1420
1421    @skipIfWindows # No pty support to test any inferior output
1422    @add_test_categories(["llgs"])
1423    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1424    @skipIf(dwarf_version=['<', '3'])
1425    def test_QEnvironment(self):
1426        self.build()
1427        exe_path = self.getBuildArtifact("a.out")
1428        env = {"FOO": "test", "BAR": "a=z"}
1429        args = [exe_path, "print-env:FOO", "print-env:BAR"]
1430        hex_args = [binascii.b2a_hex(x.encode()).decode() for x in args]
1431
1432        server = self.connect_to_debug_monitor()
1433        self.assertIsNotNone(server)
1434        self.do_handshake()
1435
1436        for key, value in env.items():
1437            self.test_sequence.add_log_lines(
1438                ["read packet: $QEnvironment:%s=%s#00" % (key, value),
1439                 "send packet: $OK#00"],
1440                True)
1441        self.test_sequence.add_log_lines(
1442            ["read packet: $vRun;%s#00" % (";".join(hex_args),),
1443             {"direction": "send",
1444              "regex": r"^\$T([0-9a-fA-F]+)"},
1445             "read packet: $c#00",
1446             "send packet: $W00#00"],
1447            True)
1448        context = self.expect_gdbremote_sequence()
1449        self.assertEqual(context["O_content"], b"test\r\na=z\r\n")
1450
1451    @skipIfWindows # No pty support to test any inferior output
1452    @add_test_categories(["llgs"])
1453    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1454    @skipIf(dwarf_version=['<', '3'])
1455    def test_QEnvironmentHexEncoded(self):
1456        self.build()
1457        exe_path = self.getBuildArtifact("a.out")
1458        env = {"FOO": "test", "BAR": "a=z", "BAZ": "a*}#z"}
1459        args = [exe_path, "print-env:FOO", "print-env:BAR", "print-env:BAZ"]
1460        hex_args = [binascii.b2a_hex(x.encode()).decode() for x in args]
1461
1462        server = self.connect_to_debug_monitor()
1463        self.assertIsNotNone(server)
1464        self.do_handshake()
1465
1466        for key, value in env.items():
1467            hex_enc = binascii.b2a_hex(("%s=%s" % (key, value)).encode()).decode()
1468            self.test_sequence.add_log_lines(
1469                ["read packet: $QEnvironmentHexEncoded:%s#00" % (hex_enc,),
1470                 "send packet: $OK#00"],
1471                True)
1472        self.test_sequence.add_log_lines(
1473            ["read packet: $vRun;%s#00" % (";".join(hex_args),),
1474             {"direction": "send",
1475              "regex": r"^\$T([0-9a-fA-F]+)"},
1476             "read packet: $c#00",
1477             "send packet: $W00#00"],
1478            True)
1479        context = self.expect_gdbremote_sequence()
1480        self.assertEqual(context["O_content"], b"test\r\na=z\r\na*}#z\r\n")
1481
1482    @skipUnlessPlatform(oslist=["freebsd", "linux"])
1483    @add_test_categories(["llgs"])
1484    @skipIf(compiler="clang", compiler_version=['<', '14.0'])
1485    @skipIf(dwarf_version=['<', '3'])
1486    def test_qXfer_siginfo_read(self):
1487        self.build()
1488        self.set_inferior_startup_launch()
1489        procs = self.prep_debug_monitor_and_inferior(
1490                inferior_args=["thread:segfault", "thread:new", "sleep:10"])
1491        self.test_sequence.add_log_lines(["read packet: $c#63"], True)
1492        self.expect_gdbremote_sequence()
1493
1494        # Run until SIGSEGV comes in.
1495        self.reset_test_sequence()
1496        self.test_sequence.add_log_lines(
1497            [{"direction": "send",
1498              "regex": r"^\$T([0-9a-fA-F]{2})thread:([0-9a-fA-F]+);",
1499              "capture": {1: "signo", 2: "thread_id"},
1500              }], True)
1501
1502        # Figure out which thread crashed.
1503        context = self.expect_gdbremote_sequence()
1504        self.assertIsNotNone(context)
1505        self.assertEqual(int(context["signo"], 16),
1506                         lldbutil.get_signal_number('SIGSEGV'))
1507        crashing_thread = int(context["thread_id"], 16)
1508
1509        # Grab siginfo for the crashing thread.
1510        self.reset_test_sequence()
1511        self.add_process_info_collection_packets()
1512        self.test_sequence.add_log_lines(
1513            ["read packet: $Hg{:x}#00".format(crashing_thread),
1514             "send packet: $OK#00",
1515             "read packet: $qXfer:siginfo:read::0,80:#00",
1516             {"direction": "send",
1517              "regex": re.compile(r"^\$([^E])(.*)#[0-9a-fA-F]{2}$",
1518                                  re.MULTILINE | re.DOTALL),
1519              "capture": {1: "response_type", 2: "content_raw"},
1520              }], True)
1521        context = self.expect_gdbremote_sequence()
1522        self.assertIsNotNone(context)
1523
1524        # Ensure we end up with all data in one packet.
1525        self.assertEqual(context.get("response_type"), "l")
1526
1527        # Decode binary data.
1528        content_raw = context.get("content_raw")
1529        self.assertIsNotNone(content_raw)
1530        content = self.decode_gdbremote_binary(content_raw).encode("latin1")
1531
1532        # Decode siginfo_t.
1533        process_info = self.parse_process_info_response(context)
1534        pad = ""
1535        if process_info["ptrsize"] == "8":
1536            pad = "i"
1537        signo_idx = 0
1538        errno_idx = 1
1539        code_idx = 2
1540        addr_idx = -1
1541        SEGV_MAPERR = 1
1542        if process_info["ostype"] == "linux":
1543            # si_signo, si_errno, si_code, [pad], _sifields._sigfault.si_addr
1544            format_str = "iii{}P".format(pad)
1545        elif process_info["ostype"].startswith("freebsd"):
1546            # si_signo, si_errno, si_code, si_pid, si_uid, si_status, si_addr
1547            format_str = "iiiiiiP"
1548        elif process_info["ostype"].startswith("netbsd"):
1549            # _signo, _code, _errno, [pad], _reason._fault._addr
1550            format_str = "iii{}P".format(pad)
1551            errno_idx = 2
1552            code_idx = 1
1553        else:
1554            assert False, "unknown ostype"
1555
1556        decoder = struct.Struct(format_str)
1557        decoded = decoder.unpack(content[:decoder.size])
1558        self.assertEqual(decoded[signo_idx],
1559                         lldbutil.get_signal_number('SIGSEGV'))
1560        self.assertEqual(decoded[errno_idx], 0)  # si_errno
1561        self.assertEqual(decoded[code_idx], SEGV_MAPERR)  # si_code
1562        self.assertEqual(decoded[addr_idx], 0)  # si_addr
1563