603 lines · python
1import os2import time3from typing import Optional, Callable, Any, List, Union4import uuid5 6import dap_server7from dap_server import Source8from lldbsuite.test.decorators import skipIf9from lldbsuite.test.lldbtest import *10from lldbsuite.test import lldbplatformutil11import lldbgdbserverutils12import base6413 14 15# DAP tests as a whole have been flakey on the Windows on Arm bot. See:16# https://github.com/llvm/llvm-project/issues/13766017@skipIf(oslist=["windows"], archs=["aarch64"])18class DAPTestCaseBase(TestBase):19 # set timeout based on whether ASAN was enabled or not. Increase20 # timeout by a factor of 10 if ASAN is enabled.21 DEFAULT_TIMEOUT = dap_server.DEFAULT_TIMEOUT22 NO_DEBUG_INFO_TESTCASE = True23 24 def create_debug_adapter(25 self,26 lldbDAPEnv: Optional[dict[str, str]] = None,27 connection: Optional[str] = None,28 additional_args: Optional[list[str]] = None,29 ):30 """Create the Visual Studio Code debug adapter"""31 self.assertTrue(32 is_exe(self.lldbDAPExec), "lldb-dap must exist and be executable"33 )34 log_file_path = self.getBuildArtifact("dap.txt")35 self.dap_server = dap_server.DebugAdapterServer(36 executable=self.lldbDAPExec,37 connection=connection,38 init_commands=self.setUpCommands(),39 log_file=log_file_path,40 env=lldbDAPEnv,41 additional_args=additional_args or [],42 spawn_helper=self.spawnSubprocess,43 )44 45 def build_and_create_debug_adapter(46 self,47 lldbDAPEnv: Optional[dict[str, str]] = None,48 dictionary: Optional[dict] = None,49 additional_args: Optional[list[str]] = None,50 ):51 self.build(dictionary=dictionary)52 self.create_debug_adapter(lldbDAPEnv, additional_args=additional_args)53 54 def build_and_create_debug_adapter_for_attach(self):55 """Variant of build_and_create_debug_adapter that builds a uniquely56 named binary."""57 unique_name = str(uuid.uuid4())58 self.build_and_create_debug_adapter(dictionary={"EXE": unique_name})59 return self.getBuildArtifact(unique_name)60 61 def set_source_breakpoints(62 self, source_path, lines, data=None, wait_for_resolve=True63 ):64 """Sets source breakpoints and returns an array of strings containing65 the breakpoint IDs ("1", "2") for each breakpoint that was set.66 Parameter data is array of data objects for breakpoints.67 Each object in data is 1:1 mapping with the entry in lines.68 It contains optional location/hitCondition/logMessage parameters.69 """70 return self.set_source_breakpoints_from_source(71 Source(path=source_path), lines, data, wait_for_resolve72 )73 74 def set_source_breakpoints_assembly(75 self, source_reference, lines, data=None, wait_for_resolve=True76 ):77 return self.set_source_breakpoints_from_source(78 Source.build(source_reference=source_reference),79 lines,80 data,81 wait_for_resolve,82 )83 84 def set_source_breakpoints_from_source(85 self, source: Source, lines, data=None, wait_for_resolve=True86 ):87 response = self.dap_server.request_setBreakpoints(88 source,89 lines,90 data,91 )92 if response is None:93 return []94 breakpoints = response["body"]["breakpoints"]95 breakpoint_ids = []96 for breakpoint in breakpoints:97 breakpoint_ids.append("%i" % (breakpoint["id"]))98 if wait_for_resolve:99 self.wait_for_breakpoints_to_resolve(breakpoint_ids)100 return breakpoint_ids101 102 def set_function_breakpoints(103 self, functions, condition=None, hitCondition=None, wait_for_resolve=True104 ):105 """Sets breakpoints by function name given an array of function names106 and returns an array of strings containing the breakpoint IDs107 ("1", "2") for each breakpoint that was set.108 """109 response = self.dap_server.request_setFunctionBreakpoints(110 functions, condition=condition, hitCondition=hitCondition111 )112 if response is None:113 return []114 breakpoints = response["body"]["breakpoints"]115 breakpoint_ids = []116 for breakpoint in breakpoints:117 breakpoint_ids.append("%i" % (breakpoint["id"]))118 if wait_for_resolve:119 self.wait_for_breakpoints_to_resolve(breakpoint_ids)120 return breakpoint_ids121 122 def wait_for_breakpoints_to_resolve(self, breakpoint_ids: list[str]):123 unresolved_breakpoints = self.dap_server.wait_for_breakpoints_to_be_verified(124 breakpoint_ids125 )126 self.assertEqual(127 len(unresolved_breakpoints),128 0,129 f"Expected to resolve all breakpoints. Unresolved breakpoint ids: {unresolved_breakpoints}",130 )131 132 def wait_until(133 self,134 predicate: Callable[[], bool],135 delay: float = 0.5,136 ) -> bool:137 """Repeatedly run the predicate until either the predicate returns True138 or a timeout has occurred."""139 deadline = time.monotonic() + self.DEFAULT_TIMEOUT140 while deadline > time.monotonic():141 if predicate():142 return True143 time.sleep(delay)144 return False145 146 def assertCapabilityIsSet(self, key: str, msg: Optional[str] = None) -> None:147 """Assert that given capability is set in the client."""148 self.assertIn(key, self.dap_server.capabilities, msg)149 self.assertEqual(self.dap_server.capabilities[key], True, msg)150 151 def assertCapabilityIsNotSet(self, key: str, msg: Optional[str] = None) -> None:152 """Assert that given capability is not set in the client."""153 if key in self.dap_server.capabilities:154 self.assertEqual(self.dap_server.capabilities[key], False, msg)155 156 def verify_breakpoint_hit(self, breakpoint_ids: List[Union[int, str]]):157 """Wait for the process we are debugging to stop, and verify we hit158 any breakpoint location in the "breakpoint_ids" array.159 "breakpoint_ids" should be a list of breakpoint ID strings160 (["1", "2"]). The return value from self.set_source_breakpoints()161 or self.set_function_breakpoints() can be passed to this function"""162 stopped_events = self.dap_server.wait_for_stopped()163 normalized_bp_ids = [str(b) for b in breakpoint_ids]164 for stopped_event in stopped_events:165 if "body" in stopped_event:166 body = stopped_event["body"]167 if "reason" not in body:168 continue169 if (170 body["reason"] != "breakpoint"171 and body["reason"] != "instruction breakpoint"172 and body["reason"] != "data breakpoint"173 ):174 continue175 if "hitBreakpointIds" not in body:176 continue177 hit_breakpoint_ids = body["hitBreakpointIds"]178 for bp in hit_breakpoint_ids:179 if str(bp) in normalized_bp_ids:180 return181 self.assertTrue(182 False,183 f"breakpoint not hit, wanted breakpoint_ids {breakpoint_ids} in stopped_events {stopped_events}",184 )185 186 def verify_all_breakpoints_hit(self, breakpoint_ids):187 """Wait for the process we are debugging to stop, and verify we hit188 all of the breakpoint locations in the "breakpoint_ids" array.189 "breakpoint_ids" should be a list of int breakpoint IDs ([1, 2])."""190 stopped_events = self.dap_server.wait_for_stopped()191 for stopped_event in stopped_events:192 if "body" in stopped_event:193 body = stopped_event["body"]194 if "reason" not in body:195 continue196 if (197 body["reason"] != "breakpoint"198 and body["reason"] != "instruction breakpoint"199 ):200 continue201 if "hitBreakpointIds" not in body:202 continue203 hit_bps = body["hitBreakpointIds"]204 if all(breakpoint_id in hit_bps for breakpoint_id in breakpoint_ids):205 return206 self.assertTrue(False, f"breakpoints not hit, stopped_events={stopped_events}")207 208 def verify_stop_exception_info(self, expected_description):209 """Wait for the process we are debugging to stop, and verify the stop210 reason is 'exception' and that the description matches211 'expected_description'212 """213 stopped_events = self.dap_server.wait_for_stopped()214 for stopped_event in stopped_events:215 if "body" in stopped_event:216 body = stopped_event["body"]217 if "reason" not in body:218 continue219 if body["reason"] != "exception":220 continue221 if "description" not in body:222 continue223 description = body["description"]224 if expected_description == description:225 return True226 return False227 228 def verify_stop_on_entry(self) -> None:229 """Waits for the process to be stopped and then verifies at least one230 thread has the stop reason 'entry'."""231 self.dap_server.wait_for_stopped()232 self.assertIn(233 "entry",234 (t["reason"] for t in self.dap_server.thread_stop_reasons.values()),235 "Expected at least one thread to report stop reason 'entry' in {self.dap_server.thread_stop_reasons}",236 )237 238 def verify_commands(self, flavor: str, output: str, commands: list[str]):239 self.assertTrue(output and len(output) > 0, "expect console output")240 lines = output.splitlines()241 prefix = "(lldb) "242 for cmd in commands:243 found = False244 for line in lines:245 if len(cmd) > 0 and (cmd[0] == "!" or cmd[0] == "?"):246 cmd = cmd[1:]247 if line.startswith(prefix) and cmd in line:248 found = True249 break250 self.assertTrue(251 found,252 f"Command '{flavor}' - '{cmd}' not found in output: {output}",253 )254 255 def verify_invalidated_event(self, expected_areas):256 event = self.dap_server.invalidated_event257 self.dap_server.invalidated_event = None258 self.assertIsNotNone(event)259 areas = event["body"].get("areas", [])260 self.assertEqual(set(expected_areas), set(areas))261 262 def verify_memory_event(self, memoryReference):263 if memoryReference is None:264 self.assertIsNone(self.dap_server.memory_event)265 event = self.dap_server.memory_event266 self.dap_server.memory_event = None267 self.assertIsNotNone(event)268 self.assertEqual(memoryReference, event["body"].get("memoryReference"))269 270 def get_dict_value(self, d: dict, key_path: list[str]) -> Any:271 """Verify each key in the key_path array is in contained in each272 dictionary within "d". Assert if any key isn't in the273 corresponding dictionary. This is handy for grabbing values from VS274 Code response dictionary like getting275 response['body']['stackFrames']276 """277 value = d278 for key in key_path:279 if key in value:280 value = value[key]281 else:282 self.assertTrue(283 key in value,284 'key "%s" from key_path "%s" not in "%s"' % (key, key_path, d),285 )286 return value287 288 def get_stackFrames_and_totalFramesCount(289 self, threadId=None, startFrame=None, levels=None, format=None, dump=False290 ):291 response = self.dap_server.request_stackTrace(292 threadId=threadId,293 startFrame=startFrame,294 levels=levels,295 format=format,296 dump=dump,297 )298 if response:299 stackFrames = self.get_dict_value(response, ["body", "stackFrames"])300 totalFrames = self.get_dict_value(response, ["body", "totalFrames"])301 self.assertTrue(302 totalFrames > 0,303 "verify totalFrames count is provided by extension that supports "304 "async frames loading",305 )306 return (stackFrames, totalFrames)307 return (None, 0)308 309 def get_stackFrames(310 self, threadId=None, startFrame=None, levels=None, format=None, dump=False311 ):312 (stackFrames, totalFrames) = self.get_stackFrames_and_totalFramesCount(313 threadId=threadId,314 startFrame=startFrame,315 levels=levels,316 format=format,317 dump=dump,318 )319 return stackFrames320 321 def get_exceptionInfo(self, threadId=None):322 response = self.dap_server.request_exceptionInfo(threadId=threadId)323 return self.get_dict_value(response, ["body"])324 325 def get_source_and_line(self, threadId=None, frameIndex=0):326 stackFrames = self.get_stackFrames(327 threadId=threadId, startFrame=frameIndex, levels=1328 )329 if stackFrames is not None:330 stackFrame = stackFrames[0]331 ["source", "path"]332 if "source" in stackFrame:333 source = stackFrame["source"]334 if "path" in source:335 if "line" in stackFrame:336 return (source["path"], stackFrame["line"])337 return ("", 0)338 339 def get_stdout(self):340 return self.dap_server.get_output("stdout")341 342 def get_console(self):343 return self.dap_server.get_output("console")344 345 def get_important(self):346 return self.dap_server.get_output("important")347 348 def collect_stdout(self, pattern: Optional[str] = None) -> str:349 return self.dap_server.collect_output("stdout", pattern=pattern)350 351 def collect_console(self, pattern: Optional[str] = None) -> str:352 return self.dap_server.collect_output("console", pattern=pattern)353 354 def collect_important(self, pattern: Optional[str] = None) -> str:355 return self.dap_server.collect_output("important", pattern=pattern)356 357 def get_local_as_int(self, name, threadId=None):358 value = self.dap_server.get_local_variable_value(name, threadId=threadId)359 # 'value' may have the variable value and summary.360 # Extract the variable value since summary can have nonnumeric characters.361 value = value.split(" ")[0]362 if value.startswith("0x"):363 return int(value, 16)364 elif value.startswith("0"):365 return int(value, 8)366 else:367 return int(value)368 369 def set_variable(self, varRef, name, value, id=None):370 """Set a variable."""371 response = self.dap_server.request_setVariable(varRef, name, str(value), id=id)372 if response["success"]:373 self.verify_invalidated_event(["variables"])374 self.verify_memory_event(response["body"].get("memoryReference"))375 return response376 377 def set_local(self, name, value, id=None):378 """Set a top level local variable only."""379 return self.set_variable(1, name, str(value), id=id)380 381 def set_global(self, name, value, id=None):382 """Set a top level global variable only."""383 return self.set_variable(2, name, str(value), id=id)384 385 def stepIn(386 self,387 threadId=None,388 targetId=None,389 waitForStop=True,390 granularity="statement",391 ):392 response = self.dap_server.request_stepIn(393 threadId=threadId, targetId=targetId, granularity=granularity394 )395 self.assertTrue(response["success"])396 if waitForStop:397 return self.dap_server.wait_for_stopped()398 return None399 400 def stepOver(401 self,402 threadId=None,403 waitForStop=True,404 granularity="statement",405 ):406 response = self.dap_server.request_next(407 threadId=threadId, granularity=granularity408 )409 self.assertTrue(410 response["success"], f"next request failed: response {response}"411 )412 if waitForStop:413 return self.dap_server.wait_for_stopped()414 return None415 416 def stepOut(self, threadId=None, waitForStop=True):417 self.dap_server.request_stepOut(threadId=threadId)418 if waitForStop:419 return self.dap_server.wait_for_stopped()420 return None421 422 def do_continue(self): # `continue` is a keyword.423 resp = self.dap_server.request_continue()424 self.assertTrue(resp["success"], f"continue request failed: {resp}")425 426 def continue_to_next_stop(self):427 self.do_continue()428 return self.dap_server.wait_for_stopped()429 430 def continue_to_breakpoint(self, breakpoint_id: str):431 self.continue_to_breakpoints((breakpoint_id))432 433 def continue_to_breakpoints(self, breakpoint_ids):434 self.do_continue()435 self.verify_breakpoint_hit(breakpoint_ids)436 437 def continue_to_exception_breakpoint(self, filter_label):438 self.do_continue()439 self.assertTrue(440 self.verify_stop_exception_info(filter_label),441 'verify we got "%s"' % (filter_label),442 )443 444 def continue_to_exit(self, exitCode=0):445 self.do_continue()446 stopped_events = self.dap_server.wait_for_stopped()447 self.assertEqual(448 len(stopped_events), 1, "stopped_events = {}".format(stopped_events)449 )450 self.assertEqual(451 stopped_events[0]["event"], "exited", "make sure program ran to completion"452 )453 self.assertEqual(454 stopped_events[0]["body"]["exitCode"],455 exitCode,456 "exitCode == %i" % (exitCode),457 )458 459 def disassemble(self, threadId=None, frameIndex=None):460 stackFrames = self.get_stackFrames(461 threadId=threadId, startFrame=frameIndex, levels=1462 )463 self.assertIsNotNone(stackFrames)464 memoryReference = stackFrames[0]["instructionPointerReference"]465 self.assertIsNotNone(memoryReference)466 467 instructions = self.dap_server.request_disassemble(468 memoryReference=memoryReference469 )470 disassembled_instructions = {}471 for inst in instructions:472 disassembled_instructions[inst["address"]] = inst473 474 return disassembled_instructions, disassembled_instructions[memoryReference]475 476 def _build_error_message(self, base_message, response):477 """Build a detailed error message from a DAP response.478 Extracts error information from various possible locations in the response structure.479 """480 error_msg = base_message481 if response:482 if "message" in response:483 error_msg += " (%s)" % response["message"]484 elif "body" in response and "error" in response["body"]:485 if "format" in response["body"]["error"]:486 error_msg += " (%s)" % response["body"]["error"]["format"]487 else:488 error_msg += " (error in body)"489 else:490 error_msg += " (no error details available)"491 else:492 error_msg += " (no response)"493 return error_msg494 495 def attach(496 self,497 *,498 disconnectAutomatically=True,499 sourceInitFile=False,500 expectFailure=False,501 **kwargs,502 ):503 """Build the default Makefile target, create the DAP debug adapter,504 and attach to the process.505 """506 507 # Make sure we disconnect and terminate the DAP debug adapter even508 # if we throw an exception during the test case.509 def cleanup():510 if disconnectAutomatically:511 self.dap_server.request_disconnect(terminateDebuggee=True)512 self.dap_server.terminate()513 514 # Execute the cleanup function during test case tear down.515 self.addTearDownHook(cleanup)516 # Initialize and launch the program517 self.dap_server.request_initialize(sourceInitFile)518 response = self.dap_server.request_attach(**kwargs)519 if expectFailure:520 return response521 if not (response and response["success"]):522 error_msg = self._build_error_message("attach failed", response)523 self.assertTrue(response and response["success"], error_msg)524 525 def launch(526 self,527 program=None,528 *,529 sourceInitFile=False,530 disconnectAutomatically=True,531 expectFailure=False,532 **kwargs,533 ):534 """Sending launch request to dap"""535 536 # Make sure we disconnect and terminate the DAP debug adapter,537 # if we throw an exception during the test case538 def cleanup():539 if disconnectAutomatically:540 self.dap_server.request_disconnect(terminateDebuggee=True)541 self.dap_server.terminate()542 543 # Execute the cleanup function during test case tear down.544 self.addTearDownHook(cleanup)545 546 # Initialize and launch the program547 self.dap_server.request_initialize(sourceInitFile)548 response = self.dap_server.request_launch(program, **kwargs)549 if expectFailure:550 return response551 if not (response and response["success"]):552 error_msg = self._build_error_message("launch failed", response)553 self.assertTrue(response and response["success"], error_msg)554 555 def build_and_launch(556 self,557 program,558 *,559 lldbDAPEnv: Optional[dict[str, str]] = None,560 **kwargs,561 ):562 """Build the default Makefile target, create the DAP debug adapter,563 and launch the process.564 """565 self.build_and_create_debug_adapter(lldbDAPEnv)566 self.assertTrue(os.path.exists(program), "executable must exist")567 568 return self.launch(program, **kwargs)569 570 def getBuiltinDebugServerTool(self):571 # Tries to find simulation/lldb-server/gdbserver tool path.572 server_tool = None573 if lldbplatformutil.getPlatform() == "linux":574 server_tool = lldbgdbserverutils.get_lldb_server_exe()575 if server_tool is None:576 self.dap_server.request_disconnect(terminateDebuggee=True)577 self.assertIsNotNone(server_tool, "lldb-server not found.")578 elif lldbplatformutil.getPlatform() == "macosx":579 server_tool = lldbgdbserverutils.get_debugserver_exe()580 if server_tool is None:581 self.dap_server.request_disconnect(terminateDebuggee=True)582 self.assertIsNotNone(server_tool, "debugserver not found.")583 return server_tool584 585 def writeMemory(self, memoryReference, data=None, offset=0, allowPartial=False):586 # This function accepts data in decimal and hexadecimal format,587 # converts it to a Base64 string, and send it to the DAP,588 # which expects Base64 encoded data.589 encodedData = (590 ""591 if data is None592 else base64.b64encode(593 # (bit_length + 7 (rounding up to nearest byte) ) //8 = converts to bytes.594 data.to_bytes((data.bit_length() + 7) // 8, "little")595 ).decode()596 )597 response = self.dap_server.request_writeMemory(598 memoryReference, encodedData, offset=offset, allowPartial=allowPartial599 )600 if response["success"]:601 self.verify_invalidated_event(["all"])602 return response603