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