lldb/test/API/tools/lldb-server/attach-wait/TestGdbRemoteAttachWait.py

   1 import os
   2 from time import sleep
   3
   4 import gdbremote_testcase
   5 import lldbgdbserverutils
   6 from lldbsuite.test.decorators import *
   7 from lldbsuite.test.lldbtest import *
   8 from lldbsuite.test import lldbutil
   9
  10
  11 class TestGdbRemoteAttachWait(gdbremote_testcase.GdbRemoteTestCaseBase):
  12     def _set_up_inferior(self):
  13         self._exe_to_attach = "%s_%d" % (self.testMethodName, os.getpid())
  14         self.build(dictionary={"EXE": self._exe_to_attach, "CXX_SOURCES": "main.cpp"})
  15
  16         if self.getPlatform() != "windows":
  17             # Use a shim to ensure that the process is ready to be attached from
  18             # the get-go.
  19             self._exe_to_run = "shim"
  20             self._run_args = [self.getBuildArtifact(self._exe_to_attach)]
  21             self.build(dictionary={"EXE": self._exe_to_run, "CXX_SOURCES": "shim.cpp"})
  22         else:
  23             self._exe_to_run = self._exe_to_attach
  24             self._run_args = []
  25
  26     def _launch_inferior(self, args):
  27         inferior = self.spawnSubprocess(self.getBuildArtifact(self._exe_to_run), args)
  28         self.assertIsNotNone(inferior)
  29         self.assertTrue(inferior.pid > 0)
  30         self.assertTrue(lldbgdbserverutils.process_is_running(inferior.pid, True))
  31         return inferior
  32
  33     def _launch_and_wait_for_init(self):
  34         sync_file_path = lldbutil.append_to_process_working_directory(
  35             self, "process_ready"
  36         )
  37         inferior = self._launch_inferior(self._run_args + [sync_file_path])
  38         lldbutil.wait_for_file_on_target(self, sync_file_path)
  39         return inferior
  40
  41     def _attach_packet(self, packet_type):
  42         return "read packet: ${};{}#00".format(
  43             packet_type,
  44             lldbgdbserverutils.gdbremote_hex_encode_string(self._exe_to_attach),
  45         )
  46
  47     @skipIfWindows  # This test is flaky on Windows
  48     def test_attach_with_vAttachWait(self):
  49         self._set_up_inferior()
  50
  51         self.set_inferior_startup_attach_manually()
  52         server = self.connect_to_debug_monitor()
  53         self.do_handshake()
  54
  55         # Launch the first inferior (we shouldn't attach to this one).
  56         self._launch_and_wait_for_init()
  57
  58         self.test_sequence.add_log_lines([self._attach_packet("vAttachWait")], True)
  59         # Run the stream until attachWait.
  60         context = self.expect_gdbremote_sequence()
  61         self.assertIsNotNone(context)
  62
  63         # Sleep so we're sure that the inferior is launched after we ask for the attach.
  64         sleep(1)
  65
  66         # Launch the second inferior (we SHOULD attach to this one).
  67         inferior_to_attach = self._launch_inferior(self._run_args)
  68
  69         # Make sure the attach succeeded.
  70         self.test_sequence.add_log_lines(
  71             [
  72                 {
  73                     "direction": "send",
  74                     "regex": r"^\$T([0-9a-fA-F]{2})[^#]*#[0-9a-fA-F]{2}$",
  75                     "capture": {1: "stop_signal_hex"},
  76                 },
  77             ],
  78             True,
  79         )
  80         self.add_process_info_collection_packets()
  81
  82         # Run the stream sending the response..
  83         context = self.expect_gdbremote_sequence()
  84         self.assertIsNotNone(context)
  85
  86         # Gather process info response.
  87         process_info = self.parse_process_info_response(context)
  88         self.assertIsNotNone(process_info)
  89
  90         # Ensure the process id matches what we expected.
  91         pid_text = process_info.get("pid", None)
  92         self.assertIsNotNone(pid_text)
  93         reported_pid = int(pid_text, base=16)
  94         self.assertEqual(reported_pid, inferior_to_attach.pid)
  95
  96     @skipIfWindows  # This test is flaky on Windows
  97     def test_launch_before_attach_with_vAttachOrWait(self):
  98         self._set_up_inferior()
  99
 100         self.set_inferior_startup_attach_manually()
 101         server = self.connect_to_debug_monitor()
 102         self.do_handshake()
 103
 104         inferior = self._launch_and_wait_for_init()
 105
 106         # Add attach packets.
 107         self.test_sequence.add_log_lines(
 108             [
 109                 # Do the attach.
 110                 self._attach_packet("vAttachOrWait"),
 111                 # Expect a stop notification from the attach.
 112                 {
 113                     "direction": "send",
 114                     "regex": r"^\$T([0-9a-fA-F]{2})[^#]*#[0-9a-fA-F]{2}$",
 115                     "capture": {1: "stop_signal_hex"},
 116                 },
 117             ],
 118             True,
 119         )
 120         self.add_process_info_collection_packets()
 121
 122         # Run the stream
 123         context = self.expect_gdbremote_sequence()
 124         self.assertIsNotNone(context)
 125
 126         # Gather process info response
 127         process_info = self.parse_process_info_response(context)
 128         self.assertIsNotNone(process_info)
 129
 130         # Ensure the process id matches what we expected.
 131         pid_text = process_info.get("pid", None)
 132         self.assertIsNotNone(pid_text)
 133         reported_pid = int(pid_text, base=16)
 134         self.assertEqual(reported_pid, inferior.pid)
 135
 136     @skipIfWindows  # This test is flaky on Windows
 137     def test_launch_after_attach_with_vAttachOrWait(self):
 138         self._set_up_inferior()
 139
 140         self.set_inferior_startup_attach_manually()
 141         server = self.connect_to_debug_monitor()
 142         self.do_handshake()
 143
 144         self.test_sequence.add_log_lines([self._attach_packet("vAttachOrWait")], True)
 145         # Run the stream until attachWait.
 146         context = self.expect_gdbremote_sequence()
 147         self.assertIsNotNone(context)
 148
 149         # Sleep so we're sure that the inferior is launched after we ask for the attach.
 150         sleep(1)
 151
 152         # Launch the inferior.
 153         inferior = self._launch_inferior(self._run_args)
 154
 155         # Make sure the attach succeeded.
 156         self.test_sequence.add_log_lines(
 157             [
 158                 {
 159                     "direction": "send",
 160                     "regex": r"^\$T([0-9a-fA-F]{2})[^#]*#[0-9a-fA-F]{2}$",
 161                     "capture": {1: "stop_signal_hex"},
 162                 },
 163             ],
 164             True,
 165         )
 166         self.add_process_info_collection_packets()
 167
 168         # Run the stream sending the response..
 169         context = self.expect_gdbremote_sequence()
 170         self.assertIsNotNone(context)
 171
 172         # Gather process info response.
 173         process_info = self.parse_process_info_response(context)
 174         self.assertIsNotNone(process_info)
 175
 176         # Ensure the process id matches what we expected.
 177         pid_text = process_info.get("pid", None)
 178         self.assertIsNotNone(pid_text)
 179         reported_pid = int(pid_text, base=16)
 180         self.assertEqual(reported_pid, inferior.pid)