================
@@ -0,0 +1,1702 @@
+# FIXME: remove when LLDB_MINIMUM_PYTHON_VERSION > 3.8
+from __future__ import annotations
+
+import base64
+import dataclasses
+import logging
+import os
+import unittest
+from dataclasses import dataclass
+from pathlib import Path
+from typing import (
+ Callable,
+ Iterable,
+ Iterator,
+ Literal,
+ Optional,
+ Sequence,
+ TypeVar,
+ cast,
+)
+
+from .dap_types import (
+ AttachArgs,
+ Breakpoint,
+ BreakpointEvent,
+ BreakpointLocationsArgs,
+ CompletionsArgs,
+ ConfigurationDoneArgs,
+ ContinueArgs,
+ DataBreakpoint,
+ DataBreakpointInfoArgs,
+ DisassembleArgs,
+ DisconnectArgs,
+ EmptyBodyResponse,
+ ErrorResponse,
+ EvaluateArgs,
+ EvaluateContext,
+ EvaluateResponse,
+ Event,
+ EventName,
+ ExceptionFilterOptions,
+ ExceptionInfoArgs,
+ ExceptionOptions,
+ ExitedEvent,
+ FunctionBreakpoint,
+ InitializeArgs,
+ InitializedEvent,
+ InstructionBreakpoint,
+ InvalidatedEvent,
+ LaunchArgs,
+ LocationsArgs,
+ MemoryEvent,
+ ModuleEvent,
+ ModuleReason,
+ ModulesArgs,
+ NextArgs,
+ OutputCategory,
+ OutputEvent,
+ ProcessEvent,
+ ReadMemoryArgs,
+ ReadMemoryResponse,
+ Response,
+ RestartArgs,
+ Scope,
+ ScopesArgs,
+ SetBreakpointsArgs,
+ SetDataBreakpointsArgs,
+ SetExceptionBreakpointsArgs,
+ SetFunctionBreakpointsArgs,
+ SetInstructionBreakpointsArgs,
+ SetVariableArgs,
+ SetVariableResponse,
+ Source,
+ SourceBreakpoint,
+ StackFrame,
+ StackFrameFormat,
+ StackTraceArgs,
+ StepInArgs,
+ StepOutArgs,
+ SteppingGranularity,
+ StoppedEvent,
+ StoppedReason,
+ TerminatedEvent,
+ ThreadsArgs,
+ ValueFormat,
+ Variable,
+ VariablePresentationHint,
+ VariablesArgs,
+ WriteMemoryArgs,
+)
+from .session import PendingResponse, Session
+from .utils import DebugAdapter, SubProcessSpawner
+
+T = TypeVar("T")
+
+
+class ThreadContext:
+ """Lazy view of a debug adapter thread.
+
+ Thread ids do not have a limited lifetime, so this context is long-lived.
+ It can be reused after continue and stepXXX requests.
+ """
+
+ def __init__(self, thread_id: int, session: DAPTestSession):
+ self._thread_id: int = thread_id
+ self._session: DAPTestSession = session
+
+ @property
+ def thread_id(self) -> int:
+ return self._thread_id
+
+ def step_in(
+ self,
+ *,
+ targetId: Optional[int] = None,
+ granularity: SteppingGranularity = "statement",
+ ):
+ return self._session.step_in(
+ threadId=self.thread_id, targetId=targetId, granularity=granularity
+ )
+
+ def step_over(self, *, granularity: SteppingGranularity = "statement"):
+ return self._session.step_over(threadId=self.thread_id,
granularity=granularity)
+
+ def step_out(self, *, granularity: SteppingGranularity = "statement"):
+ return self._session.step_out(threadId=self.thread_id,
granularity=granularity)
+
+ def top_frame(
+ self,
+ *,
+ format: Optional[StackFrameFormat] = None,
+ ) -> FrameContext:
+ return self.frames(levels=1, format=format)[0]
+
+ def frames(
+ self,
+ *,
+ startFrame: Optional[int] = None,
+ levels: Optional[int] = None,
+ format: Optional[StackFrameFormat] = None,
+ ) -> list[FrameContext]:
+ args = StackTraceArgs(
+ self._thread_id, startFrame=startFrame, levels=levels,
format=format
+ )
+ response = self._session.send_request(args).result()
+ generation = self._session._current_stop_generation()
+ return [
+ FrameContext(frame, self._session, generation)
+ for frame in response.body.stackFrames
+ ]
+
+
+class FrameContext:
+ """Lazy view of a stack frame. Valid only within its stop generation."""
+
+ def __init__(self, frame: StackFrame, session: DAPTestSession, generation:
int):
+ self._frame = frame
+ self._session = session
+ self._generation = generation
+ self._scopes: Optional[list[ScopeContext]] = None
+
+ @property
+ def frame(self) -> StackFrame:
+ self._session._check_stop_generation(self._generation, self)
+ return self._frame
+
+ @property
+ def id(self) -> int:
+ return self.frame.id
+
+ @property
+ def name(self) -> str:
+ return self.frame.name
+
+ def __dir__(self):
+ # Hide the property fields that may call 'ScopesRequest' from the
debugger.
+ # The python debugger will hang because it is waiting for a response
+ # when viewing the FrameContext.
+ hidden = {"locals", "globals", "registers", "scopes"}
+ return (attr for attr in super().__dir__() if attr not in hidden)
+
+ def source_and_line(self) -> tuple[str, int]:
+ frame = self.frame
+ assert frame.source is not None
+ assert frame.source.path is not None
+ assert frame.line is not None
+ return frame.source.path, frame.line
+
+ def scopes(self) -> list[ScopeContext]:
+ self._session._check_stop_generation(self._generation, self)
+ if self._scopes is None:
+ scope_args = ScopesArgs(frameId=self._frame.id)
+ response = self._session.send_request(scope_args).result()
+ self._scopes = [
+ ScopeContext(scope, self._session, self._generation)
+ for scope in response.body.scopes
+ ]
+ return self._scopes
+
+ def scope(self, name: str) -> ScopeContext:
+ scopes = self.scopes()
+ for scope in scopes:
+ if scope.scope.name == name:
+ return scope
+ scope_names = [scope.scope.name for scope in scopes]
+ self._session.test_case.fail(
+ f"scope '{name}' not in frame scopes: {scope_names}"
+ )
+
+ @property
+ def locals(self) -> ScopeContext:
+ return self.scope("Locals")
+
+ @property
+ def globals(self) -> ScopeContext:
+ return self.scope("Globals")
+
+ @property
+ def registers(self) -> ScopeContext:
+ return self.scope("Registers")
+
+ def evaluate(
+ self,
+ expression: str,
+ *,
+ context: Optional[EvaluateContext] = None,
+ format: Optional[ValueFormat] = None,
+ ):
+ """Evaluates `expression` in this frame's context."""
+ self._session._check_stop_generation(self._generation, self)
+ return self._session.evaluate(
+ expression, frameId=self._frame.id, context=context, format=format
+ )
+
+ def disassemble(self):
+ self._session._check_stop_generation(self._generation, self)
+
+ mem_ref = self._frame.instructionPointerReference
+ if mem_ref is None:
+ self._session.test_case.fail(
+ f"expects 'instructionPointerReference' for frame {self.frame}"
+ )
+ return self._session.disassemble(
+ mem_ref, instructionOffset=0, instructionCount=100
+ )
+
+
+class _VariableContainer:
+ """Shared dict-like behaviour for contexts that hold a variablesReference.
+
+ The optional `_value_format` is passed into every child-fetching
+ `variables` request in the container.
+ A child `VariableContext` inherits its parent's format,
+ so walking `locals.with_format(hex)["pt"]["x"]` keeps hex formatting all
+ the way down without the caller repeating it at each step.
+ """
+
+ _session: DAPTestSession
+ _generation: int
+ _value_format: Optional[ValueFormat] = None
+
+ def _fetch_variables(
+ self,
+ variables_reference: int,
+ *,
+ filter: Optional[Literal["indexed", "named"]] = None,
+ start: Optional[int] = None,
+ count: Optional[int] = None,
+ ) -> list[VariableContext]:
+ self._session._check_stop_generation(self._generation, self)
+ variables = self._session.get_variables(
+ variables_reference,
+ filter=filter,
+ start=start,
+ count=count,
+ format=self._value_format,
+ )
+ return [
+ VariableContext(var, self._session, self._generation,
self._value_format)
+ for var in variables
+ ]
+
+ def page(
+ self,
+ *,
+ filter: Optional[Literal["indexed", "named"]] = None,
+ start: Optional[int] = None,
+ count: Optional[int] = None,
+ ) -> list[VariableContext]:
+ """Fetch a subset of children with paging/filter arguments.
+
+ Inherits the container's value format.
+ """
+ return self._fetch_variables(
+ self._container_reference(),
+ filter=filter,
+ start=start,
+ count=count,
+ )
+
+ def set(
+ self, name: str, value, *, is_hex: bool = False
+ ) -> SetVariableResponse | ErrorResponse:
+ """Sends a `setVariable` request for a named child."""
+ self._session._check_stop_generation(self._generation, self)
+ return self._session.set_variable(
+ name, value, variablesReference=self._container_reference(),
is_hex=is_hex
+ )
+
+ def _container_reference(self) -> int:
+ raise NotImplementedError
+
+ def _by_name(self) -> dict[str, VariableContext]:
+ return {child.name: child for child in self._children()}
+
+ def _children(self) -> list[VariableContext]:
+ raise NotImplementedError
+
+ def __getitem__(self, name: str) -> VariableContext:
+ by_name = self._by_name()
+ try:
+ return by_name[name]
+ except KeyError:
+ self._session.test_case.fail(
+ f"'{name}' not found in {self}, has: {list(by_name)}"
+ )
+
+ def __contains__(self, name: object) -> bool:
+ return name in self._by_name()
+
+ def __iter__(self) -> Iterator[VariableContext]:
+ return iter(self._children())
+
+ def __len__(self) -> int:
+ return len(self._children())
+
+ def __str__(self) -> str:
+ return type(self).__name__
+
+
+class ScopeContext(_VariableContainer):
+ """Lazy view of a scope's variables. Valid only within its stop
generation."""
+
+ def __init__(
+ self,
+ scope: Scope,
+ session: DAPTestSession,
+ generation: int,
+ value_format: Optional[ValueFormat] = None,
+ ):
+ self._scope = scope
+ self._session = session
+ self._generation = generation
+ self._value_format = value_format
+
+ @property
+ def scope(self) -> Scope:
+ self._session._check_stop_generation(self._generation, self)
+ return self._scope
+
+ @property
+ def name(self) -> str:
+ return self.scope.name
+
+ @property
+ def variablesReference(self) -> int:
+ return self.scope.variablesReference
+
+ def variables(self) -> list[VariableContext]:
+ return self._fetch_variables(self._scope.variablesReference)
+
+ def with_format(self, *, is_hex: bool = False) -> ScopeContext:
+ """Return a new ScopeContext that applies the ValueFormat."""
+ value_format = ValueFormat(hex=True) if is_hex else None
+ return ScopeContext(self._scope, self._session, self._generation,
value_format)
+
+ def _container_reference(self) -> int:
+ return self._scope.variablesReference
+
+ def _children(self) -> list[VariableContext]:
+ return self.variables()
+
+ def __str__(self) -> str:
+ return f"scope '{self._scope.name}'"
+
+
+class VariableContext(_VariableContainer):
+ """Lazy view of a variable and (optionally) its children.
+
+ Valid only within its' stop generation.
+ """
+
+ def __init__(
+ self,
+ variable: Variable,
+ session: DAPTestSession,
+ generation: int,
+ value_format: Optional[ValueFormat] = None,
+ ):
+ self._variable = variable
+ self._session = session
+ self._generation = generation
+ self._value_format = value_format
+
+ @property
+ def variable(self) -> Variable:
+ self._session._check_stop_generation(self._generation, self)
+ return self._variable
+
+ @property
+ def name(self) -> str:
+ return self._variable.name
+
+ @property
+ def value(self) -> str:
+ return self._variable.value
+
+ @property
+ def value_as_int(self) -> int:
+ return self._variable.value_as_int
+
+ @property
+ def type(self) -> Optional[str]:
+ return self._variable.type
+
+ @property
+ def variablesReference(self) -> int:
+ return self._variable.variablesReference
+
+ @property
+ def memoryReference(self) -> Optional[str]:
+ return self._variable.memoryReference
+
+ @property
+ def indexedVariables(self) -> Optional[int]:
+ return self._variable.indexedVariables
+
+ @property
+ def namedVariables(self) -> Optional[int]:
+ return self._variable.namedVariables
+
+ @property
+ def has_children(self) -> bool:
+ return self._variable.variablesReference > 0
+
+ def children(self) -> list[VariableContext]:
+ if not self.has_children:
+ self._session.test_case.fail(
+ f"variable '{self._variable.name}' has no children"
+ )
+ return self._fetch_variables(self._variable.variablesReference)
+
+ def with_format(self, *, is_hex: bool = False) -> VariableContext:
+ """Return a new VariableContext that applies the ValueFormat"""
+ value_format = ValueFormat(hex=True) if is_hex else None
+ return VariableContext(
+ self._variable, self._session, self._generation, value_format
+ )
+
+ def _container_reference(self) -> int:
+ return self._variable.variablesReference
+
+ def _children(self) -> list[VariableContext]:
+ return self.children()
+
+ def __str__(self) -> str:
+ return f"variable '{self._variable.name}'"
+
+
+@dataclass(frozen=True)
+class CapturedOutput:
+ seen_texts: str
+ """The accumulated text until the terminator (included if it was an
OutputEvent)."""
+ event: Event
+ """The event that terminated the collection"""
+
+
+class _ConfigureContext:
+ """Handles the initial launch sequence handshake.
+
+ Orchestrates the full DAP initialization sequence:
+ On enter:
+ - Request and respond to the `Initialize` command.
+ - Send launch/attach request.
+ - Wait for InitializedEvent.
+
+ In between:
+ The test can set breakpoints or perform any check it needs do.
+
+ On exit:
+ 4. Set and verify the pending source and function breakpoints.
+ 5. Request and response to configurationDone.
+ 6. Wait for ProcessEvent and launch/attach response.
+
+ Example:
+
+ >>> session.configure(LaunchArgs(program="a.out")) as ctx:
+ ... session.resolve_function_breakpoints(["do_foo"])
+ >>> session.wait_for_breakpoint(after=ctx.process_event)
+ """
+
+ def __init__(
+ self,
+ session: "DAPTestSession",
+ config: LaunchArgs | AttachArgs,
+ ):
+ self._session = session
+ self._config = config
+ self._pending_request: Optional[PendingResponse[EmptyBodyResponse]] =
None
+
+ def __enter__(self) -> "_ConfigureContext":
+ session = self._session
+ session.test_case.assertFalse(
+ session._state.is_initialized, "session already started."
+ )
+ self.init_response =
session.initialize_sequence(session.initialize_args)
+ self._pending_request = session.send_request(self._config)
+ session.wait_for_event(InitializedEvent, after=self.init_response)
+ return self
+
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ if exc_type is not None:
+ return False
+
+ session = self._session
+ assert self.init_response is not None
+ assert self._pending_request is not None
+
+ session.verify_configuration_done()
+ self.process_event = session.wait_for_event(
+ ProcessEvent, after=self.init_response
+ )
+
+ self.launch_or_attach_response = self._pending_request.result()
+ return False
+
+
+@dataclass
+class _ExpectCommon:
+ """Shared fields used by both `ExpectVar` and `ExpectEval`.
+ Any attribute set to `None` will be skipped when checking.
+ """
+
+ type: Optional[str] = None
+ variables_reference: Optional[int] = None
+ named_variables: Optional[int] = None
+ indexed_variables: Optional[int] = None
+ read_only: bool = False
+
+ has_var_ref: Optional[bool] = None
+ has_mem_ref: Optional[bool] = None
+ has_loc_ref: Optional[bool] = None
+ has_indexed_variables: Optional[bool] = None
+
+ # Checks on the Expression's result or Variable's value.
+ startswith: Optional[str] = None
+ matches: Optional[str] = None # regex applied to .value/.result
+ # When set, fetch children via `variablesReference` and verify recursively.
+ children: Optional[dict[str, "ExpectVar"]] = None
+
+
+@dataclass
+class ExpectVar(_ExpectCommon):
+ """Typed expectation for a `Variable`.
+ Any attribute set to `None` will be skipped when checking.
+ """
+
+ value: Optional[str] = None
+ evaluate_name: Optional[str] = None
+ has_evaluate_name: Optional[bool] = None
+
+
+@dataclass
+class ExpectEval(_ExpectCommon):
+ """Typed expectation for an `EvaluateResponse.body`
+ Any attribute set to `None` will be skipped when checking.
+ """
+
+ result: Optional[str] = None
+
+
+class DAPTestSession(Session):
+ """A `Session` bound to a `unittest.TestCase`.
+
+ Adds repeating patterns for sending, receiving and verifying protocol
messages.
+ such as breakpoints, threads and evaluate.
+ """
+
+ def __init__(
+ self,
+ test_case: unittest.TestCase,
+ test_dir: Path,
+ adapter: DebugAdapter,
+ message_timeout: float,
+ process_spawner: SubProcessSpawner,
+ logger: logging.Logger,
+ ):
+ super().__init__(test_dir, adapter, message_timeout, process_spawner,
logger)
+ self.test_case = test_case
+
+ # The default features that lldb supports.
+ # When a test does not explicitly set initialize args this is used.
+ self._init_args = InitializeArgs(
+ adapterID="lldb-native",
+ clientID="vscode",
+ columnsStartAt1=True,
+ linesStartAt1=True,
+ locale="en-us",
+ pathFormat="path",
+ supportsRunInTerminalRequest=True,
+ supportsVariablePaging=True,
+ supportsVariableType=True,
+ supportsStartDebuggingRequest=True,
+ supportsProgressReporting=True,
+ supportsInvalidatedEvent=True,
+ supportsMemoryEvent=True,
+ )
+
+ def update_initialize_args(self, **kwargs):
+ self.test_case.assertFalse(
+ self._state.is_initialized,
+ "session already initialized cannot update initialize args.",
+ )
+
+ self._init_args = dataclasses.replace(self._init_args, **kwargs)
+
+ @property
+ def initialize_args(self):
+ return dataclasses.replace(self._init_args)
+
+ def launch(self, config: LaunchArgs) -> ProcessEvent:
+ """Drives the full launch handshake
+
+ (initialize -> launch -> configurationDone -> ProcessEvent).
+ """
+ with self.configure(config) as ctx:
+ pass
+ return ctx.process_event
+
+ def attach(self, config: AttachArgs) -> ProcessEvent:
+ """Drives the full attach handshake
+
+ (initialize -> attach -> configurationDone -> ProcessEvent).
+ """
+ with self.configure(config) as ctx:
+ pass
+ return ctx.process_event
+
+ def configure(self, config: LaunchArgs | AttachArgs) -> _ConfigureContext:
+ """Return a context that scopes the launch sequence.
+
+ process_event and launch_or_attach are only a valid after
+ leaving the context block.
+
+ Example:
+ >>> with session.configure(LaunchArgs(program)) as ctx:
+ ... session.set_source_breakpoints("main.cpp", [10, 25])
+ >>> process_event = ctx.process_event
+ >>> response = ctx.launch_or_attach_response
+ """
+ return _ConfigureContext(self, config)
+
+ def initialize_sequence(self, initialize_args: InitializeArgs):
+ init_response = self.send_request(initialize_args).result()
+ return init_response
+
+ def initialize_and_launch(self, args: LaunchArgs | AttachArgs):
+ self.initialize_sequence(self.initialize_args)
+ return self.send_request(args)
+
+ def configuration_done(self) -> PendingResponse[EmptyBodyResponse]:
+ # Wait for initialized event.
+ self.ensure_initialized()
+ # And then send configuration done.
+ return self.send_request(ConfigurationDoneArgs())
+
+ def verify_configuration_done(self, expected_success: bool = True):
+ response = self.configuration_done().result_or_error()
+ if expected_success:
+ self.test_case.assertEqual(
+ response.success, True, f"got error response: {response}."
+ )
+ self.test_case.assertIsInstance(response, EmptyBodyResponse)
+
+ # In VSCode, immediately following 'configurationDone', a
+ # 'threads' request is made to get the initial set of threads,
+ # specifically the main threads id and name.
+ # We issue the threads request to mimic this pattern and prevent
+ # tests that use threads to have the wrong result.
+ self.send_request(ThreadsArgs()).result()
+ else:
+ self.test_case.assertEqual(response.success, False)
+ self.test_case.assertIsInstance(response, ErrorResponse)
+ return response
+
+ def set_source_breakpoints(
+ self, source_path: str, breakpoints: list[int] | list[SourceBreakpoint]
+ ):
+ self.ensure_initialized()
+ # Convert the deprecated lines field to SourceBreakpoints.
+ s_breakpoints: list[SourceBreakpoint] = []
+ for bp in breakpoints:
+ if isinstance(bp, int):
+ s_breakpoints.append(SourceBreakpoint(bp))
+ elif isinstance(bp, SourceBreakpoint):
+ s_breakpoints.append(bp)
+ else:
+ self.test_case.fail(
+ "breakpoints must only contain ints or SourceBreakpoints."
+ f" got '{bp}' of type '{type(bp)}'."
+ )
+
+ source = Source.create(path=source_path)
+ bp_args = SetBreakpointsArgs(source, breakpoints=s_breakpoints)
+ return self.send_request(bp_args).result()
+
+ def set_assembly_breakpoints(
+ self,
+ source: Source | int,
+ breakpoints: list[int] | list[SourceBreakpoint],
+ ):
+ """Set breakpoints in an assembly source.
+
+ `source` can be either a `sourceReference` int (for a source produced
+ in the current session) or a full `Source` object (for replaying a
+ persisted assembly source across sessions).
+ """
+ self.ensure_initialized()
+ if isinstance(source, int):
+ source = Source(sourceReference=source)
+
+ s_breakpoints: list[SourceBreakpoint] = []
+ for bp in breakpoints:
+ if isinstance(bp, int):
+ s_breakpoints.append(SourceBreakpoint(bp))
+ elif isinstance(bp, SourceBreakpoint):
+ s_breakpoints.append(bp)
+ else:
+ self.test_case.fail(
+ "breakpoints must only contain ints or SourceBreakpoints."
+ f" got '{bp}' of type '{type(bp)}'."
+ )
+
+ bp_args = SetBreakpointsArgs(source=source, breakpoints=s_breakpoints)
+ return self.send_request(bp_args).result()
+
+ def set_function_breakpoints(
+ self,
+ function_names: list[str],
+ condition: Optional[str] = None,
+ hitCondition: Optional[str] = None,
+ ):
+ f_breakpoints = [
+ FunctionBreakpoint(name, condition=condition,
hitCondition=hitCondition)
+ for name in function_names
+ ]
+ response =
self.send_request(SetFunctionBreakpointsArgs(f_breakpoints)).result()
+ return response
+
+ def set_exception_breakpoints(
+ self,
+ *,
+ filters: list[str],
+ filterOptions: Optional[list[ExceptionFilterOptions]] = None,
+ exceptionOptions: Optional[list[ExceptionOptions]] = None,
+ ):
+ args = SetExceptionBreakpointsArgs(
+ filters=filters,
+ filterOptions=filterOptions,
+ exceptionOptions=exceptionOptions,
+ )
+ response = self.send_request(args).result()
+ return response
+
+ def set_variable(
+ self, name: str, value, *, variablesReference: int, is_hex: bool =
False
+ ) -> SetVariableResponse | ErrorResponse:
+ last_response = self.last_response()
+ handle = self.send_request(
+ SetVariableArgs(
+ variablesReference=variablesReference,
+ name=name,
+ value=str(value),
+ format=ValueFormat(hex=True) if is_hex else None,
+ )
+ )
+ response = handle.result_or_error()
+ if isinstance(response, SetVariableResponse):
+ invalidated_event = self.wait_for_invalidated(after=last_response)
+ invalidated_areas = invalidated_event.body.areas
+ self.test_case.assertEqual(["variables"], invalidated_areas)
+
+ memory_event = self.wait_for_memory(after=last_response)
+ self.test_case.assertEqual(
+ memory_event.body.memoryReference,
response.body.memoryReference
+ )
+ return response
+
+ @staticmethod
+ def breakpoints_to_ids(breakpoints: list[Breakpoint]):
+ ids: list[int] = []
+ for bp in breakpoints:
+ assert bp.id is not None, f"id is None for breakpoint: {bp}"
+ ids.append(bp.id)
+ return ids
+
+ def resolve_source_breakpoints(
+ self, source_path: str, breakpoints: list[int] | list[SourceBreakpoint]
+ ) -> list[int]:
+ last_response = self.last_response()
+ bp_response = self.set_source_breakpoints(source_path, breakpoints)
+ r_breakpoints = bp_response.body.breakpoints
+
+ all_verified = all(bp.verified for bp in r_breakpoints)
+ if not all_verified:
+ self.wait_until_all_breakpoints_verified(r_breakpoints,
after=last_response)
+
+ self.test_case.assertEqual(
+ len(breakpoints),
+ len(r_breakpoints),
+ "expect correct number of breakpoints.",
+ )
+ return self.breakpoints_to_ids(r_breakpoints)
+
+ def resolve_function_breakpoints(
+ self,
+ function_names: list[str],
+ condition: Optional[str] = None,
+ hitCondition: Optional[str] = None,
+ ) -> list[int]:
+ """Sets breakpoints by function name given an array of function names
+ and returns an array of strings containing the breakpoint IDs
+ ("1", "2") for each breakpoint that was set.
+ """
+ last_response = self.last_response()
+ response = self.set_function_breakpoints(
+ function_names, condition, hitCondition
+ )
+ breakpoints = response.body.breakpoints
+
+ all_verified = all(bp.verified for bp in breakpoints)
+ if not all_verified:
+ self.wait_until_all_breakpoints_verified(breakpoints,
after=last_response)
+
+ return self.breakpoints_to_ids(breakpoints)
+
+ def set_data_breakpoints(self, breakpoints: list[DataBreakpoint]):
+ args = SetDataBreakpointsArgs(breakpoints=breakpoints)
+ return self.send_request(args).result()
+
+ def set_instruction_breakpoints(self, memory_references: list[str]):
+ breakpoints = [InstructionBreakpoint(ref) for ref in memory_references]
+ return
self.send_request(SetInstructionBreakpointsArgs(breakpoints)).result()
+
+ def set_breakpoint_locations(
+ self,
+ file_path: str,
+ line: int,
+ column: Optional[int] = None,
+ endLine: Optional[int] = None,
+ endColumn: Optional[int] = None,
+ ):
+ _, name = os.path.split(file_path)
+ bp_loc_args = BreakpointLocationsArgs(
+ Source.create(name=name, path=file_path),
+ line=line,
+ column=column,
+ endLine=endLine,
+ endColumn=endColumn,
+ )
+ return self.send_request(bp_loc_args).result()
+
+ def data_breakpoint_info(self, name: str, variablesReference: int,
frameId: int):
+ info_args = DataBreakpointInfoArgs(
+ name=name, variablesReference=variablesReference, frameId=frameId
+ )
+ return self.send_request(info_args).result()
+
+ def data_breakpoint_info_as_address(self, address: str, size: int):
+ info_args = DataBreakpointInfoArgs(name=address, bytes=size,
asAddress=True)
+ return self.send_request(info_args).result()
+
+ def step_in(
+ self,
+ threadId: Optional[int] = None,
+ *,
+ targetId: Optional[int] = None,
+ granularity: SteppingGranularity = "statement",
+ ):
+ thread_id = threadId or self.stopped_thread_id
+ self.test_case.assertIsNotNone(thread_id)
+ stepin_args = StepInArgs(
+ threadId=thread_id, targetId=targetId, granularity=granularity
+ )
+ response = self.send_request(stepin_args).result()
+ stop_event = self.verify_stopped(StoppedReason.STEP, after=response)
+ return stop_event
+
+ def step_over(
+ self,
+ threadId: Optional[int] = None,
+ *,
+ granularity: SteppingGranularity = "statement",
+ ):
+ thread_id = threadId or self.stopped_thread_id
+ self.test_case.assertIsNotNone(thread_id)
+ next_args = NextArgs(threadId=thread_id, granularity=granularity)
+ response = self.send_request(next_args).result()
+ stop_event = self.verify_stopped(StoppedReason.STEP, after=response)
+ return stop_event
+
+ def step_out(
+ self,
+ threadId: Optional[int] = None,
+ *,
+ granularity: SteppingGranularity = "statement",
+ ):
+ thread_id = threadId or self.stopped_thread_id
+ self.test_case.assertIsNotNone(thread_id)
+ step_out_args = StepOutArgs(threadId=thread_id,
granularity=granularity)
+ response = self.send_request(step_out_args).result()
+
+ stop_event = self.verify_stopped(StoppedReason.STEP, after=response)
+ return stop_event
+
+ def wait_until_any_breakpoint_hit(
+ self, breakpoint_ids: list[int], *, after: Event | Response
+ ) -> StoppedEvent:
+ """Wait for the process to send `StoppedEvents` and verify we stopped
for
+ any breakpoint in breakpoint_ids the event or response.
+ """
+
+ self.test_case.assertGreater(len(breakpoint_ids), 0, "empty breakpoint
ids.")
+ is_ids_int = all(isinstance(id, int) for id in breakpoint_ids)
+ self.test_case.assertTrue(is_ids_int, "all breakpoint_ids must be
integers.")
+
+ breakpoint_stop_reasons = [
+ StoppedReason.BREAKPOINT,
+ StoppedReason.INSTRUCTION_BREAKPOINT,
+ StoppedReason.FUNCTION_BREAKPOINT,
+ StoppedReason.DATA_BREAKPOINT,
+ ]
+
+ def event_hit_id_in_breakpoint_ids(event: StoppedEvent):
+ hit_ids = event.body.hitBreakpointIds or []
+ for hit_id in hit_ids:
+ if hit_id in breakpoint_ids:
+ return True
+
+ return False
+
+ event = self.wait_for_stopped(
+ breakpoint_stop_reasons,
+ after=after,
+ until=event_hit_id_in_breakpoint_ids,
+ timeout_msg=f"waiting for any breakpoint id in {breakpoint_ids}
after seq {after.seq}.",
+ )
+
+ return event
+
+ def wait_until_all_breakpoints_verified(
+ self, breakpoints: list[int] | list[Breakpoint], *, after: Event |
Response
+ ):
+ """Wait for the process to send breakpoint events and verify we hit
+ all 'ids' in 'breakpoints' after the event or response.
+ """
+ self.test_case.assertTrue(len(breakpoints) > 0, "empty list of
breakpoints.")
+
+ id_to_verify: dict[int, bool] = {}
+ for bp in breakpoints:
+ if isinstance(bp, int):
+ id_to_verify[bp] = False
+ elif isinstance(bp, Breakpoint):
+ assert bp.id is not None
+ id_to_verify[bp.id] = bp.verified
+ else:
+ self.test_case.fail(
+ f"expected list of type 'Breakpoint' or 'int' got
'{breakpoints}'"
+ )
+
+ bp_ids = list(id_to_verify.keys())
+
+ def all_breakpoints_verified(evt: BreakpointEvent):
+ event_bp = evt.body.breakpoint
+ if event_bp.id is None:
+ return False
+
+ if event_bp.id not in bp_ids:
+ return False
+
+ id_to_verify[event_bp.id] = event_bp.verified
+ all_verified = all(verified for verified in id_to_verify.values())
+ return all_verified
+
+ timeout_msg = f"waiting for all breakpoint ids {bp_ids} to be verified"
+ last_breakpoint_event = self.wait_for_event(
+ BreakpointEvent,
+ after=after,
+ until=all_breakpoints_verified,
+ timeout_msg=timeout_msg,
+ )
+ return last_breakpoint_event
+
+ def wait_for_stopped_or_exited(
+ self,
+ *,
+ after: Event | Response,
+ until: Optional[Callable[[StoppedEvent | ExitedEvent], bool]] = None,
+ timeout_msg: Optional[str] = None,
+ ) -> StoppedEvent | ExitedEvent:
+ event = self.wait_for_any_event(
+ (StoppedEvent, ExitedEvent),
+ after=after,
+ until=until,
+ timeout_msg=timeout_msg,
+ )
+ return event
+
+ def wait_for_stopped(
+ self,
+ matching_any: Optional[Sequence[StoppedReason]] = None,
+ *,
+ after: Event | Response,
+ until: Optional[Callable[[StoppedEvent], bool]] = None,
+ timeout_msg: Optional[str] = None,
+ ) -> StoppedEvent:
+ """
+ Wait for a process to stop, optionally filtered by stop reason and
custom condition.
+
+ Blocks until a StoppedEvent is received after the specified event. If
matching_any
+ is provided, only stops with those reasons are accepted. The until
callback allows
+ additional condition checking. If an ExitedEvent is encountered,
wait_for terminates.
+
+ Args:
+ matching_any: Filter by specific stop reasons.
+ after: Event or Response to start waiting after.
+ until: Optional callback for additional filtering.
+ timeout_msg: Custom timeout error message.
+ """
+ if matching_any:
+ self.test_case.assertGreater(
+ len(matching_any), 0, "expected at least one stop reason."
+ )
+
+ def matches_any_reason_until(event: StoppedEvent | ExitedEvent):
+ # Break early for exited event.
+ # We cannot hit a stopped event after the process exited.
+ if isinstance(event, ExitedEvent):
+ return True
+
+ # Match any of the stopped reasons.
+ if matching_any and event.body.reason not in matching_any:
+ return False
+
+ if until:
+ return until(event)
+
+ return True
+
+ event = self.wait_for_stopped_or_exited(
+ after=after, until=matches_any_reason_until,
timeout_msg=timeout_msg
+ )
+
+ self.test_case.assertIsInstance(event, StoppedEvent, f"after seq:
{after.seq}")
+ self.test_case.assertEqual(event.event, EventName.STOPPED)
+ return cast(StoppedEvent, event)
+
+ def wait_for_exited(self, *, after: Event | Response) -> ExitedEvent:
+ """
+ Wait for a process to exit.
+
+ Blocks until an ExitedEvent is received following the given event or
response.
+ Raises an error if a StoppedEvent is encountered, as a stopped process
+ cannot subsequently exit.
+ """
+ event = self.wait_for_stopped_or_exited(after=after)
+ self.test_case.assertIsInstance(event, ExitedEvent)
+ self.test_case.assertEqual(event.event, "exited", "expected
ExitedEvent'")
+ return cast(ExitedEvent, event)
+
+ def verify_next_module_event(
+ self,
+ reason: Optional[ModuleReason] = None,
+ *,
+ after: Event | Response,
+ ):
+ event = self.wait_for_module_event(after=after)
+ event_body = event.body
+ if reason is not None:
+ msg = f"module event reason does not match, got {event_body}."
+ self.test_case.assertEqual(event_body.reason, reason, msg)
+ return event
+
+ def wait_for_module_event(
----------------
da-viper wrote:
updated.
https://github.com/llvm/llvm-project/pull/203978
_______________________________________________
lldb-commits mailing list
[email protected]
https://lists.llvm.org/cgi-bin/mailman/listinfo/lldb-commits
