================
@@ -0,0 +1,1699 @@
+# 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
----------------
da-viper wrote:
same as above
https://github.com/llvm/llvm-project/pull/203978
_______________________________________________
lldb-commits mailing list
[email protected]
https://lists.llvm.org/cgi-bin/mailman/listinfo/lldb-commits
