| import os | |
| from time import sleep | |
| import gdbremote_testcase | |
| import lldbgdbserverutils | |
| from lldbsuite.test.decorators import * | |
| from lldbsuite.test.lldbtest import * | |
| from lldbsuite.test import lldbutil | |
| @skipIfMTE # MTE security transition shims restrict socket operations. | |
| class TestGdbRemoteAttachWait(gdbremote_testcase.GdbRemoteTestCaseBase): | |
| SHARED_BUILD_TESTCASE = False | |
| def _set_up_inferior(self): | |
| self._exe_to_attach = "%s_%d" % (self.testMethodName, os.getpid()) | |
| self.build(dictionary={"EXE": self._exe_to_attach, "CXX_SOURCES": "main.cpp"}) | |
| if self.getPlatform() != "windows": | |
| # Use a shim to ensure that the process is ready to be attached from | |
| # the get-go. | |
| self._exe_to_run = "shim" | |
| self._exe_to_attach = lldbutil.install_to_target( | |
| self, self.getBuildArtifact(self._exe_to_attach) | |
| ) | |
| self._run_args = [self._exe_to_attach] | |
| self.build(dictionary={"EXE": self._exe_to_run, "CXX_SOURCES": "shim.cpp"}) | |
| else: | |
| self._exe_to_run = self._exe_to_attach | |
| self._run_args = [] | |
| def _launch_inferior(self, args): | |
| inferior = self.spawnSubprocess(self.getBuildArtifact(self._exe_to_run), args) | |
| self.assertIsNotNone(inferior) | |
| self.assertGreater(inferior.pid, 0) | |
| self.assertTrue(lldbgdbserverutils.process_is_running(inferior.pid, True)) | |
| return inferior | |
| def _launch_and_wait_for_init(self): | |
| sync_file_path = lldbutil.append_to_process_working_directory( | |
| self, "process_ready" | |
| ) | |
| inferior = self._launch_inferior(self._run_args + [sync_file_path]) | |
| lldbutil.wait_for_file_on_target(self, sync_file_path) | |
| return inferior | |
| def _attach_packet(self, packet_type): | |
| return "read packet: ${};{}#00".format( | |
| packet_type, | |
| lldbgdbserverutils.gdbremote_hex_encode_string(self._exe_to_attach), | |
| ) | |
| def test_attach_with_vAttachWait(self): | |
| self._set_up_inferior() | |
| self.set_inferior_startup_attach_manually() | |
| server = self.connect_to_debug_monitor() | |
| self.do_handshake() | |
| # Launch the first inferior (we shouldn't attach to this one). | |
| self._launch_and_wait_for_init() | |
| self.test_sequence.add_log_lines([self._attach_packet("vAttachWait")], True) | |
| # Run the stream until attachWait. | |
| context = self.expect_gdbremote_sequence() | |
| self.assertIsNotNone(context) | |
| # Sleep so we're sure that the inferior is launched after we ask for the attach. | |
| sleep(1) | |
| # Launch the second inferior (we SHOULD attach to this one). | |
| inferior_to_attach = self._launch_inferior(self._run_args) | |
| # Make sure the attach succeeded. | |
| self.test_sequence.add_log_lines( | |
| [ | |
| { | |
| "direction": "send", | |
| "regex": r"^\$T([0-9a-fA-F]{2})[^#]*#[0-9a-fA-F]{2}$", | |
| "capture": {1: "stop_signal_hex"}, | |
| }, | |
| ], | |
| True, | |
| ) | |
| self.add_process_info_collection_packets() | |
| # Run the stream sending the response.. | |
| context = self.expect_gdbremote_sequence() | |
| self.assertIsNotNone(context) | |
| # Gather process info response. | |
| process_info = self.parse_process_info_response(context) | |
| self.assertIsNotNone(process_info) | |
| # Ensure the process id matches what we expected. | |
| pid_text = process_info.get("pid", None) | |
| self.assertIsNotNone(pid_text) | |
| reported_pid = int(pid_text, base=16) | |
| self.assertEqual(reported_pid, inferior_to_attach.pid) | |
| def test_launch_before_attach_with_vAttachOrWait(self): | |
| self._set_up_inferior() | |
| self.set_inferior_startup_attach_manually() | |
| server = self.connect_to_debug_monitor() | |
| self.do_handshake() | |
| inferior = self._launch_and_wait_for_init() | |
| # Add attach packets. | |
| self.test_sequence.add_log_lines( | |
| [ | |
| # Do the attach. | |
| self._attach_packet("vAttachOrWait"), | |
| # Expect a stop notification from the attach. | |
| { | |
| "direction": "send", | |
| "regex": r"^\$T([0-9a-fA-F]{2})[^#]*#[0-9a-fA-F]{2}$", | |
| "capture": {1: "stop_signal_hex"}, | |
| }, | |
| ], | |
| True, | |
| ) | |
| self.add_process_info_collection_packets() | |
| # Run the stream | |
| context = self.expect_gdbremote_sequence() | |
| self.assertIsNotNone(context) | |
| # Gather process info response | |
| process_info = self.parse_process_info_response(context) | |
| self.assertIsNotNone(process_info) | |
| # Ensure the process id matches what we expected. | |
| pid_text = process_info.get("pid", None) | |
| self.assertIsNotNone(pid_text) | |
| reported_pid = int(pid_text, base=16) | |
| self.assertEqual(reported_pid, inferior.pid) | |
| def test_launch_after_attach_with_vAttachOrWait(self): | |
| self._set_up_inferior() | |
| self.set_inferior_startup_attach_manually() | |
| server = self.connect_to_debug_monitor() | |
| self.do_handshake() | |
| self.test_sequence.add_log_lines([self._attach_packet("vAttachOrWait")], True) | |
| # Run the stream until attachWait. | |
| context = self.expect_gdbremote_sequence() | |
| self.assertIsNotNone(context) | |
| # Sleep so we're sure that the inferior is launched after we ask for the attach. | |
| sleep(1) | |
| # Launch the inferior. | |
| inferior = self._launch_inferior(self._run_args) | |
| # Make sure the attach succeeded. | |
| self.test_sequence.add_log_lines( | |
| [ | |
| { | |
| "direction": "send", | |
| "regex": r"^\$T([0-9a-fA-F]{2})[^#]*#[0-9a-fA-F]{2}$", | |
| "capture": {1: "stop_signal_hex"}, | |
| }, | |
| ], | |
| True, | |
| ) | |
| self.add_process_info_collection_packets() | |
| # Run the stream sending the response.. | |
| context = self.expect_gdbremote_sequence() | |
| self.assertIsNotNone(context) | |
| # Gather process info response. | |
| process_info = self.parse_process_info_response(context) | |
| self.assertIsNotNone(process_info) | |
| # Ensure the process id matches what we expected. | |
| pid_text = process_info.get("pid", None) | |
| self.assertIsNotNone(pid_text) | |
| reported_pid = int(pid_text, base=16) | |
| self.assertEqual(reported_pid, inferior.pid) |