dd59a99cbf
Updates the lldb python test suite to ensure we call dumpSessionInfo()
in the test result's stopTest() method. This will ensure that we get the
session info dumped for all tests, even those that don't have an
explicit call to dumpSessionInfo() in the test case.
Additionally, I updated the lldb-dap test case to mark the '-dap.log' as
a log file, which will be recorded in the test output on failure.
Here is an example test run with a failure:
```
PASS: LLDB (build/bin/clang-arm64) :: test_step (TestDAP_step.TestDAP_step)
FAIL: LLDB (build/bin/clang-arm64) :: test_step_over_inlined_function (TestDAP_step.TestDAP_step)
Log Files:
- build/lldb-test-build.noindex/tools/lldb-dap/step/TestDAP_step/Failure.log
- build/lldb-test-build.noindex/tools/lldb-dap/step/TestDAP_step/Failure-dap.log
======================================================================
FAIL: test_step_over_inlined_function (TestDAP_step.TestDAP_step)
Test stepping over when the program counter is in another file.
----------------------------------------------------------------------
Traceback (most recent call last):
File "llvm-project/lldb/test/API/tools/lldb-dap/step/TestDAP_step.py", line 113, in test_step_over_inlined_function
self.assertFalse(
AssertionError: True is not false : expect path ending with 'main.cpp'.
Config=arm64-build/bin/clang
----------------------------------------------------------------------
Ran 2 tests in 4.849s
```
2139 lines
74 KiB
Python
2139 lines
74 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# FIXME: remove when LLDB_MINIMUM_PYTHON_VERSION > 3.8
|
|
from __future__ import annotations
|
|
|
|
from typing import Protocol
|
|
import argparse
|
|
import dataclasses
|
|
import enum
|
|
import json
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from typing import (
|
|
Any,
|
|
BinaryIO,
|
|
Callable,
|
|
Final,
|
|
cast,
|
|
Dict,
|
|
List,
|
|
Literal,
|
|
Optional,
|
|
Tuple,
|
|
TypedDict,
|
|
Union,
|
|
)
|
|
|
|
|
|
# set timeout based on whether ASAN was enabled or not. Increase
|
|
# timeout by a factor of 10 if ASAN is enabled.
|
|
DEFAULT_TIMEOUT: Final[float] = 50.0 * (10.0 if ("ASAN_OPTIONS" in os.environ) else 1.0)
|
|
|
|
|
|
# A quiet period between events, used to determine if we're done receiving
|
|
# events in a given window, otherwise 'wait_for_stopped' would need to wait
|
|
# until the DEFAULT_TIMEOUT occurs, slows down tests significantly.
|
|
EVENT_QUIET_PERIOD = 0.25 * (20.0 if ("ASAN_OPTIONS" in os.environ) else 1.0)
|
|
|
|
|
|
# See lldbtest.Base.spawnSubprocess, which should help ensure any processes
|
|
# created by the DAP client are terminated correctly when the test ends.
|
|
class SpawnHelperCallback(Protocol):
|
|
def __call__(
|
|
self, executable: str, args: List[str], extra_env: List[str], **kwargs
|
|
) -> subprocess.Popen:
|
|
...
|
|
|
|
|
|
## DAP type references
|
|
|
|
|
|
class Event(TypedDict):
|
|
type: Literal["event"]
|
|
seq: int
|
|
event: str
|
|
body: Any
|
|
|
|
|
|
class Request(TypedDict):
|
|
type: Literal["request"]
|
|
seq: int
|
|
command: str
|
|
arguments: Any
|
|
|
|
|
|
class Response(TypedDict):
|
|
type: Literal["response"]
|
|
seq: int
|
|
request_seq: int
|
|
success: bool
|
|
command: str
|
|
message: Optional[str]
|
|
body: Any
|
|
|
|
|
|
ProtocolMessage = Union[Event, Request, Response]
|
|
|
|
|
|
class Source(TypedDict, total=False):
|
|
name: str
|
|
path: str
|
|
sourceReference: int
|
|
|
|
@staticmethod
|
|
def build(
|
|
*,
|
|
name: Optional[str] = None,
|
|
path: Optional[str] = None,
|
|
source_reference: Optional[int] = None,
|
|
) -> "Source":
|
|
"""Builds a source from the given name, path or source_reference."""
|
|
if not name and not path and not source_reference:
|
|
raise ValueError(
|
|
"Source.build requires either name, path, or source_reference"
|
|
)
|
|
|
|
s = Source()
|
|
if name:
|
|
s["name"] = name
|
|
if path:
|
|
if not name:
|
|
s["name"] = os.path.basename(path)
|
|
s["path"] = path
|
|
if source_reference is not None:
|
|
s["sourceReference"] = source_reference
|
|
return s
|
|
|
|
|
|
class Breakpoint(TypedDict, total=False):
|
|
id: int
|
|
verified: bool
|
|
source: Source
|
|
|
|
@staticmethod
|
|
def is_verified(src: "Breakpoint") -> bool:
|
|
return src.get("verified", False)
|
|
|
|
|
|
class NotSupportedError(KeyError):
|
|
"""Raised if a feature is not supported due to its capabilities."""
|
|
|
|
|
|
class RequestError(Exception):
|
|
"""Raised if a DAP request fails."""
|
|
|
|
def __init__(self, request: Request, response: Optional[Response] = None):
|
|
super().__init__()
|
|
self.request = request
|
|
self.response = response
|
|
|
|
def __str__(self) -> str:
|
|
desc = f"request failed request={self.request!r}"
|
|
if self.response:
|
|
desc += f" response={self.response!r}"
|
|
return desc
|
|
|
|
|
|
class ReplayMods(TypedDict, total=False):
|
|
"""Fields that can be overwritten in requests during a replay."""
|
|
|
|
frameId: Optional[int]
|
|
threadId: Optional[int]
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class Log:
|
|
class Dir(enum.Enum):
|
|
SENT = 1
|
|
RECV = 2
|
|
|
|
@property
|
|
def requests(self) -> List[Tuple[Dir, Request]]:
|
|
"""All requests in the log, in order."""
|
|
return [m for m in self.messages if m[1]["type"] == "request"]
|
|
|
|
@property
|
|
def events(self) -> List[Tuple[Dir, Event]]:
|
|
"""All events in the log, in order."""
|
|
return [m for m in self.messages if m[1]["type"] == "event"]
|
|
|
|
@property
|
|
def responses(self) -> List[Tuple[Dir, Response]]:
|
|
"""All responses in the log, in order."""
|
|
return [m for m in self.messages if m[1]["type"] == "response"]
|
|
|
|
messages: List[Tuple[Dir, ProtocolMessage]] = dataclasses.field(
|
|
default_factory=list
|
|
)
|
|
|
|
@classmethod
|
|
def load(cls, file: pathlib.Path) -> "Log":
|
|
"""Load the file and parse any log messages. Returns (sent, recv)."""
|
|
sent_pattern = re.compile(r"\d+\.\d+ \(.+\) --> ")
|
|
recv_pattern = re.compile(r"\d+\.\d+ \(.+\) <-- ")
|
|
|
|
log = Log()
|
|
with open(file, "r") as f:
|
|
for line in f:
|
|
if sent_pattern.match(line):
|
|
packet = line.split("--> ", maxsplit=1)[1]
|
|
log.messages.append((Log.Dir.SENT, json.loads(packet)))
|
|
elif recv_pattern.match(line):
|
|
packet = line.split("<-- ", maxsplit=1)[1]
|
|
log.messages.append((Log.Dir.RECV, json.loads(packet)))
|
|
return log
|
|
|
|
|
|
class DebugCommunication(object):
|
|
@property
|
|
def is_stopped(self) -> bool:
|
|
"""Returns True if the debuggee is stopped, otherwise False."""
|
|
return len(self.thread_stop_reasons) > 0 or self.exit_status is not None
|
|
|
|
def __init__(
|
|
self,
|
|
recv: BinaryIO,
|
|
send: BinaryIO,
|
|
init_commands: Optional[List[str]] = None,
|
|
spawn_helper: Optional[SpawnHelperCallback] = None,
|
|
):
|
|
self._log = Log()
|
|
self.send = send
|
|
self.recv = recv
|
|
self.spawn_helper = spawn_helper
|
|
|
|
# Packets that have been received and processed but have not yet been
|
|
# requested by a test case.
|
|
self._pending_packets: List[Optional[ProtocolMessage]] = []
|
|
# Received packets that have not yet been processed.
|
|
self._recv_packets: List[Optional[ProtocolMessage]] = []
|
|
# Used as a mutex for _recv_packets and for notify when _recv_packets
|
|
# changes.
|
|
self._recv_condition = threading.Condition()
|
|
self._recv_thread = threading.Thread(target=self._read_packet_thread)
|
|
|
|
# session state
|
|
self.init_commands = init_commands if init_commands else []
|
|
self.exit_status: Optional[int] = None
|
|
self.capabilities: Dict = {}
|
|
self.initialized: bool = False
|
|
self.process_event_body: Optional[Dict] = None
|
|
self.configuration_done_sent = False
|
|
self.launch_or_attach_sent = False
|
|
self.terminated: bool = False
|
|
self.events: List[Event] = []
|
|
self.progress_events: List[Event] = []
|
|
self.reverse_requests: List[Request] = []
|
|
self.sequence: int = 1
|
|
self.output: Dict[str, str] = {}
|
|
self.reverse_process: Optional[subprocess.Popen] = None
|
|
|
|
# debuggee state
|
|
self.threads: Optional[dict] = None
|
|
self.stopped_thread: Optional[dict] = None
|
|
self.thread_stacks: Optional[Dict[int, List[dict]]]
|
|
self.thread_stop_reasons: Dict[str, Any] = {}
|
|
self.focused_tid: Optional[int] = None
|
|
self.frame_scopes: Dict[str, Any] = {}
|
|
# keyed by breakpoint id
|
|
self.resolved_breakpoints: dict[int, Breakpoint] = {}
|
|
|
|
# Modifiers used when replaying a log file.
|
|
self._mod = ReplayMods()
|
|
|
|
# trigger enqueue thread
|
|
self._recv_thread.start()
|
|
self.initialized_event = None
|
|
|
|
@classmethod
|
|
def encode_content(cls, s: str) -> bytes:
|
|
return ("Content-Length: %u\r\n\r\n%s" % (len(s), s)).encode("utf-8")
|
|
|
|
@classmethod
|
|
def validate_response(cls, request: Request, response: Response) -> None:
|
|
if request["command"] != response["command"]:
|
|
raise ValueError(
|
|
f"command mismatch in response {request['command']} != {response['command']}"
|
|
)
|
|
if request["seq"] != response["request_seq"]:
|
|
raise ValueError(
|
|
f"seq mismatch in response {request['seq']} != {response['request_seq']}"
|
|
)
|
|
|
|
def _read_packet(self) -> Optional[ProtocolMessage]:
|
|
"""Decode a JSON packet that starts with the content length and is
|
|
followed by the JSON bytes. Returns None on EOF.
|
|
"""
|
|
line = self.recv.readline().decode("utf-8")
|
|
if len(line) == 0:
|
|
return None # EOF.
|
|
|
|
# Watch for line that starts with the prefix
|
|
prefix = "Content-Length: "
|
|
if line.startswith(prefix):
|
|
# Decode length of JSON bytes
|
|
length = int(line[len(prefix) :])
|
|
# Skip empty line
|
|
separator = self.recv.readline().decode()
|
|
if separator != "":
|
|
Exception("malformed DAP content header, unexpected line: " + separator)
|
|
# Read JSON bytes
|
|
json_str = self.recv.read(length).decode()
|
|
# Decode the JSON bytes into a python dictionary
|
|
return json.loads(json_str)
|
|
|
|
raise Exception("unexpected malformed message from lldb-dap: " + line)
|
|
|
|
def _read_packet_thread(self):
|
|
while True:
|
|
packet = self._read_packet()
|
|
# `packet` will be `None` on EOF. We want to pass it down to
|
|
# handle_recv_packet anyway so the main thread can handle unexpected
|
|
# termination of lldb-dap and stop waiting for new packets.
|
|
if not self._handle_recv_packet(packet):
|
|
break
|
|
|
|
def get_modules(
|
|
self, start_module: Optional[int] = None, module_count: Optional[int] = None
|
|
) -> Dict:
|
|
resp = self.request_modules(start_module, module_count)
|
|
if not resp["success"]:
|
|
raise ValueError(f"request_modules failed: {resp!r}")
|
|
modules = {}
|
|
module_list = resp["body"]["modules"]
|
|
for module in module_list:
|
|
modules[module["name"]] = module
|
|
return modules
|
|
|
|
def get_output(self, category: str, clear=True) -> str:
|
|
output = ""
|
|
if category in self.output:
|
|
output = self.output.get(category, "")
|
|
if clear:
|
|
del self.output[category]
|
|
return output
|
|
|
|
def collect_output(
|
|
self,
|
|
category: str,
|
|
pattern: Optional[str] = None,
|
|
clear=True,
|
|
) -> str:
|
|
"""Collect output from 'output' events.
|
|
Args:
|
|
category: The category to collect.
|
|
pattern:
|
|
Optional, if set, return once this pattern is detected in the
|
|
collected output.
|
|
Returns:
|
|
The collected output.
|
|
"""
|
|
deadline = time.monotonic() + DEFAULT_TIMEOUT
|
|
output = self.get_output(category, clear)
|
|
while deadline >= time.monotonic() and (
|
|
pattern is None or pattern not in output
|
|
):
|
|
event = self.wait_for_event(["output"])
|
|
if not event: # Timeout or EOF
|
|
break
|
|
output += self.get_output(category, clear=clear)
|
|
return output
|
|
|
|
def _handle_recv_packet(self, packet: Optional[ProtocolMessage]) -> bool:
|
|
"""Handles an incoming packet.
|
|
|
|
Called by the read thread that is waiting for all incoming packets
|
|
to store the incoming packet in "self._recv_packets" in a thread safe
|
|
way. This function will then signal the "self._recv_condition" to
|
|
indicate a new packet is available.
|
|
|
|
Args:
|
|
packet: A new packet to store.
|
|
|
|
Returns:
|
|
True if the caller should keep calling this function for more
|
|
packets.
|
|
"""
|
|
with self._recv_condition:
|
|
self._recv_packets.append(packet)
|
|
self._recv_condition.notify()
|
|
# packet is None on EOF
|
|
return packet is not None and not (
|
|
packet["type"] == "response" and packet["command"] == "disconnect"
|
|
)
|
|
|
|
def _recv_packet(
|
|
self,
|
|
*,
|
|
predicate: Optional[Callable[[ProtocolMessage], bool]] = None,
|
|
timeout: float = DEFAULT_TIMEOUT,
|
|
) -> ProtocolMessage:
|
|
"""Processes received packets from the adapter.
|
|
Updates the DebugCommunication stateful properties based on the received
|
|
packets in the order they are received.
|
|
|
|
NOTE: The only time the session state properties should be updated is
|
|
during this call to ensure consistency during tests.
|
|
|
|
Args:
|
|
predicate:
|
|
Optional, if specified, returns the first packet that matches
|
|
the given predicate.
|
|
timeout:
|
|
Processes packets until either the timeout occurs or the
|
|
predicate matches a packet, whichever occurs first.
|
|
|
|
Returns:
|
|
The first matching packet for the given predicate.
|
|
|
|
Raises:
|
|
TimeoutError: Timeout while waiting for predicate.
|
|
EOFError: End of stream detected.
|
|
"""
|
|
assert (
|
|
threading.current_thread() != self._recv_thread
|
|
), "Must not be called from the _recv_thread"
|
|
|
|
def process_until_match():
|
|
self._process_recv_packets()
|
|
for i, packet in enumerate(self._pending_packets):
|
|
if packet is None:
|
|
# We need to return a truthy value to break out of the
|
|
# wait_for, use `EOFError` as an indicator of EOF.
|
|
return EOFError()
|
|
if predicate and predicate(packet):
|
|
self._pending_packets.pop(i)
|
|
return packet
|
|
return False
|
|
|
|
with self._recv_condition:
|
|
packet = self._recv_condition.wait_for(process_until_match, timeout)
|
|
if isinstance(packet, EOFError):
|
|
raise EOFError
|
|
if not packet:
|
|
raise TimeoutError
|
|
return cast(ProtocolMessage, packet)
|
|
|
|
def _process_recv_packets(self) -> None:
|
|
"""Process received packets, updating the session state."""
|
|
with self._recv_condition:
|
|
for packet in self._recv_packets:
|
|
if packet and ("seq" not in packet or packet["seq"] == 0):
|
|
raise ValueError(
|
|
f"received a malformed packet, expected 'seq != 0' for {packet!r}"
|
|
)
|
|
if packet:
|
|
self._log.messages.append((Log.Dir.RECV, packet))
|
|
# Handle events that may modify any stateful properties of
|
|
# the DAP session.
|
|
if packet and packet["type"] == "event":
|
|
self._handle_event(packet)
|
|
elif packet and packet["type"] == "request":
|
|
# Handle reverse requests and keep processing.
|
|
self._handle_reverse_request(packet)
|
|
# Move the packet to the pending queue.
|
|
self._pending_packets.append(packet)
|
|
self._recv_packets.clear()
|
|
|
|
def _handle_event(self, packet: Event) -> None:
|
|
"""Handle any events that modify debug session state we track."""
|
|
event = packet["event"]
|
|
body: Optional[Dict] = packet.get("body", None)
|
|
|
|
if event == "output" and body:
|
|
# Store any output we receive so clients can retrieve it later.
|
|
category = body["category"]
|
|
output = body["output"]
|
|
if category in self.output:
|
|
self.output[category] += output
|
|
else:
|
|
self.output[category] = output
|
|
elif event == "initialized":
|
|
self.initialized = True
|
|
self.initialized_event = packet
|
|
elif event == "process":
|
|
# When a new process is attached or launched, remember the
|
|
# details that are available in the body of the event
|
|
self.process_event_body = body
|
|
elif event == "exited" and body:
|
|
# Process exited, mark the status to indicate the process is not
|
|
# alive.
|
|
self.exit_status = body["exitCode"]
|
|
elif event == "continued" and body:
|
|
# When the process continues, clear the known threads and
|
|
# thread_stop_reasons.
|
|
all_threads_continued = body.get("allThreadsContinued", True)
|
|
tid = body["threadId"]
|
|
if tid in self.thread_stop_reasons:
|
|
del self.thread_stop_reasons[tid]
|
|
self._process_continued(all_threads_continued)
|
|
elif event == "stopped" and body:
|
|
# Each thread that stops with a reason will send a
|
|
# 'stopped' event. We need to remember the thread stop
|
|
# reasons since the 'threads' command doesn't return
|
|
# that information.
|
|
self._process_stopped()
|
|
tid = body["threadId"]
|
|
self.thread_stop_reasons[tid] = body
|
|
if "preserveFocusHint" not in body or not body["preserveFocusHint"]:
|
|
self.focused_tid = tid
|
|
elif event.startswith("progress"):
|
|
# Progress events come in as 'progressStart', 'progressUpdate',
|
|
# and 'progressEnd' events. Keep these around in case test
|
|
# cases want to verify them.
|
|
self.progress_events.append(packet)
|
|
elif event == "breakpoint" and body:
|
|
# Breakpoint events are sent when a breakpoint is resolved
|
|
self._update_verified_breakpoints([body["breakpoint"]])
|
|
elif event == "capabilities" and body:
|
|
# Update the capabilities with new ones from the event.
|
|
self.capabilities.update(body["capabilities"])
|
|
|
|
def _handle_reverse_request(self, request: Request) -> None:
|
|
if request in self.reverse_requests:
|
|
return
|
|
self.reverse_requests.append(request)
|
|
arguments = request.get("arguments")
|
|
if request["command"] == "runInTerminal" and arguments is not None:
|
|
assert self.spawn_helper is not None, "Not configured to spawn subprocesses"
|
|
[exe, *args] = arguments["args"]
|
|
env = [f"{k}={v}" for k, v in arguments.get("env", {}).items()]
|
|
self.reverse_process = self.spawn_helper(
|
|
exe, args, env, stdout=subprocess.PIPE, stderr=subprocess.PIPE
|
|
)
|
|
body = {"processId": self.reverse_process.pid}
|
|
self.send_packet(
|
|
{
|
|
"type": "response",
|
|
"seq": 0,
|
|
"request_seq": request["seq"],
|
|
"success": True,
|
|
"message": None,
|
|
"command": "runInTerminal",
|
|
"body": body,
|
|
}
|
|
)
|
|
elif request["command"] == "startDebugging":
|
|
self.send_packet(
|
|
{
|
|
"type": "response",
|
|
"seq": 0,
|
|
"request_seq": request["seq"],
|
|
"success": True,
|
|
"message": None,
|
|
"command": "startDebugging",
|
|
"body": {},
|
|
}
|
|
)
|
|
else:
|
|
desc = 'unknown reverse request "%s"' % (request["command"])
|
|
raise ValueError(desc)
|
|
|
|
def _process_continued(self, all_threads_continued: bool):
|
|
self.frame_scopes = {}
|
|
if all_threads_continued:
|
|
self.thread_stop_reasons = {}
|
|
self.focused_tid = None
|
|
|
|
def _update_verified_breakpoints(self, breakpoints: List[Breakpoint]):
|
|
for bp in breakpoints:
|
|
# If no id is set, we cannot correlate the given breakpoint across
|
|
# requests, ignore it.
|
|
if "id" not in bp:
|
|
continue
|
|
|
|
self.resolved_breakpoints[bp["id"]] = bp
|
|
|
|
def send_packet(self, packet: ProtocolMessage) -> int:
|
|
"""Takes a dictionary representation of a DAP request and send the request to the debug adapter.
|
|
|
|
Returns the seq number of the request.
|
|
"""
|
|
# Set the seq for requests.
|
|
if packet["type"] == "request":
|
|
packet["seq"] = self.sequence
|
|
self.sequence += 1
|
|
else:
|
|
packet["seq"] = 0
|
|
|
|
# Encode our command dictionary as a JSON string
|
|
json_str = json.dumps(packet, separators=(",", ":"))
|
|
|
|
length = len(json_str)
|
|
if length > 0:
|
|
# Send the encoded JSON packet and flush the 'send' file
|
|
self.send.write(self.encode_content(json_str))
|
|
self.send.flush()
|
|
|
|
self._log.messages.append((Log.Dir.SENT, packet))
|
|
|
|
return packet["seq"]
|
|
|
|
def _send_recv(self, request: Request) -> Response:
|
|
"""Send a command python dictionary as JSON and receive the JSON
|
|
response. Validates that the response is the correct sequence and
|
|
command in the reply. Any events that are received are added to the
|
|
events list in this object"""
|
|
response = None
|
|
try:
|
|
seq = self.send_packet(request)
|
|
response = self.receive_response(seq)
|
|
self.validate_response(request, response)
|
|
return response
|
|
except Exception as exc:
|
|
# Add additional context for the request and response.
|
|
raise RequestError(request, response) from exc
|
|
|
|
def receive_response(self, seq: int) -> Response:
|
|
"""Waits for a response with the associated request_sec."""
|
|
|
|
def predicate(p: ProtocolMessage):
|
|
return p["type"] == "response" and p["request_seq"] == seq
|
|
|
|
return cast(Response, self._recv_packet(predicate=predicate))
|
|
|
|
def wait_for_event(self, filter: List[str]) -> Event:
|
|
"""Wait for the first event that matches the filter."""
|
|
|
|
def predicate(p: ProtocolMessage):
|
|
return p["type"] == "event" and p["event"] in filter
|
|
|
|
return cast(Event, self._recv_packet(predicate=predicate))
|
|
|
|
def collect_events(self, filter: List[str]) -> List[Event]:
|
|
"""Wait for the first event that matches the filter."""
|
|
events = []
|
|
|
|
def predicate(p: ProtocolMessage):
|
|
return p["type"] == "event" and p["event"] in filter
|
|
|
|
event = cast(Event, self._recv_packet(predicate=predicate))
|
|
while event:
|
|
events.append(event)
|
|
try:
|
|
event = cast(
|
|
Event,
|
|
self._recv_packet(predicate=predicate, timeout=EVENT_QUIET_PERIOD),
|
|
)
|
|
except TimeoutError:
|
|
break
|
|
return events
|
|
|
|
def wait_for_initialized(self) -> None:
|
|
"""Wait for the debugger to become initialized."""
|
|
if self.initialized:
|
|
return
|
|
self.wait_for_event(["initialized"])
|
|
|
|
def ensure_initialized(self) -> None:
|
|
"""Validates that we can wait for initialized."""
|
|
if self.initialized:
|
|
return
|
|
assert self.launch_or_attach_sent, "launch or attach request not yet sent"
|
|
assert (
|
|
not self.configuration_done_sent
|
|
), "configuration done has already been sent, 'initialized' should have already occurred"
|
|
self.wait_for_initialized()
|
|
|
|
def wait_for_stopped(self) -> List[Event]:
|
|
"""Wait for the next 'stopped' event to occur, coalescing all stopped events within a given quiet period."""
|
|
return self.collect_events(["stopped", "exited"])
|
|
|
|
def wait_for_module_events(self) -> List[Event]:
|
|
"""Wait for the next 'module' event to occur, coalescing all module events within a given quiet period."""
|
|
return self.collect_events(["module"])
|
|
|
|
def wait_for_breakpoint_events(self) -> List[Event]:
|
|
"""wait for the next 'breakpoint' event to occur, coalescing all breakpoint events within a given quiet period."""
|
|
return self.collect_events(["breakpoint"])
|
|
|
|
def wait_for_breakpoints_to_be_verified(self, breakpoint_ids: List[int]):
|
|
"""Wait for all breakpoints to be verified. Return all unverified breakpoints."""
|
|
while any(id not in self.resolved_breakpoints for id in breakpoint_ids):
|
|
breakpoint_event = self.wait_for_event(["breakpoint"])
|
|
if breakpoint_event is None:
|
|
break
|
|
|
|
return [
|
|
id
|
|
for id in breakpoint_ids
|
|
if (
|
|
id not in self.resolved_breakpoints
|
|
or not Breakpoint.is_verified(self.resolved_breakpoints[id])
|
|
)
|
|
]
|
|
|
|
def wait_for_exited(self):
|
|
event_dict = self.wait_for_event(["exited"])
|
|
if event_dict is None:
|
|
raise ValueError("didn't get exited event")
|
|
return event_dict
|
|
|
|
def wait_for_terminated(self):
|
|
event_dict = self.wait_for_event(["terminated"])
|
|
if event_dict is None:
|
|
raise ValueError("didn't get terminated event")
|
|
return event_dict
|
|
|
|
def wait_for_invalidated(self):
|
|
event_dict = self.wait_for_event(["invalidated"])
|
|
if event_dict is None:
|
|
raise ValueError("didn't get invalidated event")
|
|
return event_dict
|
|
|
|
def wait_for_memory(self):
|
|
event_dict = self.wait_for_event(["memory"])
|
|
if event_dict is None:
|
|
raise ValueError("didn't get memory event")
|
|
return event_dict
|
|
|
|
def get_capability(self, key: str):
|
|
"""Get a value for the given key if it there is a key/value pair in
|
|
the capabilities reported by the adapter.
|
|
"""
|
|
if key in self.capabilities:
|
|
return self.capabilities[key]
|
|
raise NotSupportedError(key)
|
|
|
|
def get_threads(self):
|
|
if self.threads is None:
|
|
self.request_threads()
|
|
return self.threads
|
|
|
|
def get_thread_id(self, threadIndex=0):
|
|
"""Utility function to get the first thread ID in the thread list.
|
|
If the thread list is empty, then fetch the threads.
|
|
"""
|
|
if self.threads is None:
|
|
self.request_threads()
|
|
if self.threads and threadIndex < len(self.threads):
|
|
return self.threads[threadIndex]["id"]
|
|
return None
|
|
|
|
def get_stackFrame(self, frameIndex=0, threadId=None):
|
|
"""Get a single "StackFrame" object from a "stackTrace" request and
|
|
return the "StackFrame" as a python dictionary, or None on failure
|
|
"""
|
|
if threadId is None:
|
|
threadId = self.get_thread_id()
|
|
if threadId is None:
|
|
print("invalid threadId")
|
|
return None
|
|
response = self.request_stackTrace(threadId, startFrame=frameIndex, levels=1)
|
|
if response:
|
|
return response["body"]["stackFrames"][0]
|
|
print("invalid response")
|
|
return None
|
|
|
|
def get_completions(self, text, frameId=None):
|
|
if frameId is None:
|
|
stackFrame = self.get_stackFrame()
|
|
frameId = stackFrame["id"]
|
|
response = self.request_completions(text, frameId)
|
|
return response["body"]["targets"]
|
|
|
|
def get_scope_variables(self, scope_name, frameIndex=0, threadId=None, is_hex=None):
|
|
stackFrame = self.get_stackFrame(frameIndex=frameIndex, threadId=threadId)
|
|
if stackFrame is None:
|
|
return []
|
|
frameId = stackFrame["id"]
|
|
if frameId in self.frame_scopes:
|
|
frame_scopes = self.frame_scopes[frameId]
|
|
else:
|
|
scopes_response = self.request_scopes(frameId)
|
|
frame_scopes = scopes_response["body"]["scopes"]
|
|
self.frame_scopes[frameId] = frame_scopes
|
|
for scope in frame_scopes:
|
|
if scope["name"] == scope_name:
|
|
varRef = scope["variablesReference"]
|
|
variables_response = self.request_variables(varRef, is_hex=is_hex)
|
|
if not variables_response["success"]:
|
|
return variables_response
|
|
if variables_response and "body" in variables_response:
|
|
body = variables_response["body"]
|
|
if "variables" in body:
|
|
vars = body["variables"]
|
|
return vars
|
|
return []
|
|
|
|
def get_global_variables(self, frameIndex=0, threadId=None):
|
|
return self.get_scope_variables(
|
|
"Globals", frameIndex=frameIndex, threadId=threadId
|
|
)
|
|
|
|
def get_local_variables(self, frameIndex=0, threadId=None, is_hex=None):
|
|
return self.get_scope_variables(
|
|
"Locals", frameIndex=frameIndex, threadId=threadId, is_hex=is_hex
|
|
)
|
|
|
|
def get_registers(self, frameIndex=0, threadId=None):
|
|
return self.get_scope_variables(
|
|
"Registers", frameIndex=frameIndex, threadId=threadId
|
|
)
|
|
|
|
def get_local_variable(self, name, frameIndex=0, threadId=None, is_hex=None):
|
|
locals = self.get_local_variables(
|
|
frameIndex=frameIndex, threadId=threadId, is_hex=is_hex
|
|
)
|
|
for local in locals:
|
|
if "name" in local and local["name"] == name:
|
|
return local
|
|
return None
|
|
|
|
def get_local_variable_value(self, name, frameIndex=0, threadId=None, is_hex=None):
|
|
variable = self.get_local_variable(
|
|
name, frameIndex=frameIndex, threadId=threadId, is_hex=is_hex
|
|
)
|
|
if variable and "value" in variable:
|
|
return variable["value"]
|
|
return None
|
|
|
|
def get_local_variable_child(
|
|
self, name, child_name, frameIndex=0, threadId=None, is_hex=None
|
|
):
|
|
local = self.get_local_variable(name, frameIndex, threadId)
|
|
if local["variablesReference"] == 0:
|
|
return None
|
|
children = self.request_variables(local["variablesReference"], is_hex=is_hex)[
|
|
"body"
|
|
]["variables"]
|
|
for child in children:
|
|
if child["name"] == child_name:
|
|
return child
|
|
return None
|
|
|
|
def _preconditions(self, req: Request) -> None:
|
|
"""Validate any preconditions for the given command, potentially waiting
|
|
for the debuggee to be in a specific state.
|
|
"""
|
|
if req["command"] == "threads":
|
|
logging.debug("Waiting on precondition: stopped")
|
|
self._recv_packet(predicate=lambda _: self.is_stopped)
|
|
|
|
# Apply any modifications to arguments.
|
|
args = req["arguments"]
|
|
if "threadId" in args and "threadId" in self._mod:
|
|
args["threadId"] = self._mod["threadId"]
|
|
if "frameId" in args and "frameId" in self._mod:
|
|
args["frameId"] = self._mod["frameId"]
|
|
|
|
def _postconditions(self, resp: Response) -> None:
|
|
"""Validate any postconditions for the given response, potentially
|
|
waiting for the debuggee to be in a specific state.
|
|
"""
|
|
if resp["command"] == "launch":
|
|
logging.debug("Waiting on postcondition: initialized")
|
|
self._recv_packet(predicate=lambda _: self.initialized)
|
|
elif resp["command"] == "configurationDone":
|
|
logging.debug("Waiting on postcondition: process")
|
|
self._recv_packet(predicate=lambda _: self.process_event_body is not None)
|
|
|
|
# Store some modifications related to replayed requests.
|
|
if resp["command"] == "threads":
|
|
self._mod["threadId"] = resp["body"]["threads"][0]["id"]
|
|
if resp["command"] in ["continue", "next", "stepIn", "stepOut", "pause"]:
|
|
self._mod.clear()
|
|
self._recv_packet(predicate=lambda _: self.is_stopped)
|
|
if resp["command"] == "stackTrace" and not self._mod.get("frameId", None):
|
|
self._mod["frameId"] = next(
|
|
(frame["id"] for frame in resp["body"]["stackFrames"]), None
|
|
)
|
|
|
|
def replay(self, file: pathlib.Path) -> None:
|
|
"""Replay a log file."""
|
|
log = Log.load(file)
|
|
responses = {
|
|
r["request_seq"]: r for (dir, r) in log.responses if dir == Log.Dir.RECV
|
|
}
|
|
for dir, packet in log.messages:
|
|
if dir != Log.Dir.SENT or packet["type"] != "request":
|
|
continue
|
|
req = packet
|
|
want = responses[req["seq"]]
|
|
|
|
self._preconditions(req)
|
|
|
|
logging.info("Sending req %r", req)
|
|
got = self._send_recv(req)
|
|
logging.info("Received resp %r", got)
|
|
|
|
assert (
|
|
got["command"] == want["command"] == req["command"]
|
|
), f"got {got} want {want} for req {req}"
|
|
assert (
|
|
got["success"] == want["success"]
|
|
), f"got {got} want {want} for req {req}"
|
|
|
|
self._postconditions(got)
|
|
|
|
def request_attach(
|
|
self,
|
|
*,
|
|
program: Optional[str] = None,
|
|
pid: Optional[int] = None,
|
|
session: Optional[dict[str, int]] = None,
|
|
waitFor=False,
|
|
initCommands: Optional[list[str]] = None,
|
|
preRunCommands: Optional[list[str]] = None,
|
|
attachCommands: Optional[list[str]] = None,
|
|
postRunCommands: Optional[list[str]] = None,
|
|
stopCommands: Optional[list[str]] = None,
|
|
exitCommands: Optional[list[str]] = None,
|
|
terminateCommands: Optional[list[str]] = None,
|
|
coreFile: Optional[str] = None,
|
|
stopOnEntry=False,
|
|
sourceMap: Optional[Union[list[tuple[str, str]], dict[str, str]]] = None,
|
|
gdbRemotePort: Optional[int] = None,
|
|
gdbRemoteHostname: Optional[str] = None,
|
|
) -> int:
|
|
args_dict: Dict[str, Any] = {}
|
|
if pid is not None:
|
|
args_dict["pid"] = pid
|
|
if program is not None:
|
|
args_dict["program"] = program
|
|
if session is not None:
|
|
args_dict["session"] = session
|
|
if waitFor:
|
|
args_dict["waitFor"] = waitFor
|
|
args_dict["initCommands"] = self.init_commands
|
|
if initCommands:
|
|
args_dict["initCommands"].extend(initCommands)
|
|
if preRunCommands:
|
|
args_dict["preRunCommands"] = preRunCommands
|
|
if stopCommands:
|
|
args_dict["stopCommands"] = stopCommands
|
|
if exitCommands:
|
|
args_dict["exitCommands"] = exitCommands
|
|
if terminateCommands:
|
|
args_dict["terminateCommands"] = terminateCommands
|
|
if attachCommands:
|
|
args_dict["attachCommands"] = attachCommands
|
|
if coreFile:
|
|
args_dict["coreFile"] = coreFile
|
|
if stopOnEntry:
|
|
args_dict["stopOnEntry"] = stopOnEntry
|
|
if postRunCommands:
|
|
args_dict["postRunCommands"] = postRunCommands
|
|
if sourceMap:
|
|
args_dict["sourceMap"] = sourceMap
|
|
if gdbRemotePort is not None:
|
|
args_dict["gdb-remote-port"] = gdbRemotePort
|
|
if gdbRemoteHostname is not None:
|
|
args_dict["gdb-remote-hostname"] = gdbRemoteHostname
|
|
command_dict: Request = {
|
|
"seq": 0,
|
|
"command": "attach",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
self.launch_or_attach_sent = True
|
|
return self.send_packet(command_dict)
|
|
|
|
def request_breakpointLocations(
|
|
self, file_path, line, end_line=None, column=None, end_column=None
|
|
):
|
|
(dir, base) = os.path.split(file_path)
|
|
source_dict = {"name": base, "path": file_path}
|
|
args_dict = {}
|
|
args_dict["source"] = source_dict
|
|
if line is not None:
|
|
args_dict["line"] = line
|
|
if end_line is not None:
|
|
args_dict["endLine"] = end_line
|
|
if column is not None:
|
|
args_dict["column"] = column
|
|
if end_column is not None:
|
|
args_dict["endColumn"] = end_column
|
|
command_dict = {
|
|
"command": "breakpointLocations",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_configurationDone(self):
|
|
command_dict = {
|
|
"command": "configurationDone",
|
|
"type": "request",
|
|
"arguments": {},
|
|
}
|
|
response = self._send_recv(command_dict)
|
|
self.configuration_done_sent = True
|
|
if response and response["success"]:
|
|
stopped_on_entry = self.is_stopped
|
|
threads_response = self.request_threads()
|
|
if not stopped_on_entry:
|
|
# Drop the initial cached threads if we did not stop-on-entry.
|
|
# 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 but in our
|
|
# tests we don't want to cache the result unless the process is
|
|
# actually stopped.
|
|
self.threads = None
|
|
return response
|
|
|
|
def _process_stopped(self):
|
|
self.threads = None
|
|
self.frame_scopes = {}
|
|
|
|
def request_continue(self, threadId=None, singleThread=False):
|
|
if self.exit_status is not None:
|
|
raise ValueError("request_continue called after process exited")
|
|
# If we have launched or attached, then the first continue is done by
|
|
# sending the 'configurationDone' request
|
|
if not self.configuration_done_sent:
|
|
return self.request_configurationDone()
|
|
args_dict = {}
|
|
if threadId is None:
|
|
threadId = self.get_thread_id()
|
|
if threadId:
|
|
args_dict["threadId"] = threadId
|
|
if singleThread:
|
|
args_dict["singleThread"] = True
|
|
command_dict = {
|
|
"command": "continue",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
response = self._send_recv(command_dict)
|
|
if response["success"]:
|
|
self._process_continued(response["body"]["allThreadsContinued"])
|
|
# Caller must still call wait_for_stopped.
|
|
return response
|
|
|
|
def request_restart(self, restartArguments=None):
|
|
if self.exit_status is not None:
|
|
raise ValueError("request_restart called after process exited")
|
|
self.get_capability("supportsRestartRequest")
|
|
command_dict = {
|
|
"command": "restart",
|
|
"type": "request",
|
|
}
|
|
if restartArguments:
|
|
command_dict["arguments"] = restartArguments
|
|
|
|
response = self._send_recv(command_dict)
|
|
# Caller must still call wait_for_stopped.
|
|
return response
|
|
|
|
def request_disconnect(self, terminateDebuggee=None):
|
|
args_dict = {}
|
|
if terminateDebuggee is not None:
|
|
if terminateDebuggee:
|
|
args_dict["terminateDebuggee"] = True
|
|
else:
|
|
args_dict["terminateDebuggee"] = False
|
|
command_dict = {
|
|
"command": "disconnect",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_disassemble(
|
|
self,
|
|
memoryReference,
|
|
instructionOffset=-50,
|
|
instructionCount=200,
|
|
resolveSymbols=True,
|
|
):
|
|
args_dict = {
|
|
"memoryReference": memoryReference,
|
|
"instructionOffset": instructionOffset,
|
|
"instructionCount": instructionCount,
|
|
"resolveSymbols": resolveSymbols,
|
|
}
|
|
command_dict = {
|
|
"command": "disassemble",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)["body"]["instructions"]
|
|
|
|
def request_readMemory(self, memoryReference, offset, count):
|
|
args_dict = {
|
|
"memoryReference": memoryReference,
|
|
"offset": offset,
|
|
"count": count,
|
|
}
|
|
command_dict = {
|
|
"command": "readMemory",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_writeMemory(self, memoryReference, data, offset=0, allowPartial=False):
|
|
args_dict = {
|
|
"memoryReference": memoryReference,
|
|
"data": data,
|
|
}
|
|
|
|
if offset:
|
|
args_dict["offset"] = offset
|
|
if allowPartial:
|
|
args_dict["allowPartial"] = allowPartial
|
|
|
|
command_dict = {
|
|
"command": "writeMemory",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_evaluate(
|
|
self,
|
|
expression,
|
|
frameIndex: Optional[int] = 0,
|
|
threadId=None,
|
|
context=None,
|
|
is_hex: Optional[bool] = None,
|
|
) -> Response:
|
|
args_dict = {
|
|
"expression": expression,
|
|
}
|
|
|
|
if frameIndex is not None:
|
|
if threadId is None:
|
|
threadId = self.get_thread_id()
|
|
stackFrame = self.get_stackFrame(frameIndex=frameIndex, threadId=threadId)
|
|
|
|
if stackFrame is None:
|
|
raise ValueError("invalid frameIndex")
|
|
args_dict["frameId"] = stackFrame["id"]
|
|
|
|
if context:
|
|
args_dict["context"] = context
|
|
if is_hex is not None:
|
|
args_dict["format"] = {"hex": is_hex}
|
|
command_dict = {
|
|
"command": "evaluate",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_exceptionInfo(self, threadId=None):
|
|
if threadId is None:
|
|
threadId = self.get_thread_id()
|
|
args_dict = {"threadId": threadId}
|
|
command_dict = {
|
|
"command": "exceptionInfo",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_initialize(
|
|
self, client_features: Optional[dict[str, bool]] = None, sourceInitFile=False
|
|
):
|
|
command_dict = {
|
|
"command": "initialize",
|
|
"type": "request",
|
|
"arguments": {
|
|
"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,
|
|
"$__lldb_sourceInitFile": sourceInitFile,
|
|
},
|
|
}
|
|
|
|
if client_features is not None:
|
|
arguments = command_dict["arguments"]
|
|
# replace the default client features.
|
|
for key, value in client_features.items():
|
|
arguments[key] = value
|
|
|
|
response = self._send_recv(command_dict)
|
|
if response:
|
|
if "body" in response:
|
|
self.capabilities.update(response.get("body", {}))
|
|
return response
|
|
|
|
def request_launch(
|
|
self,
|
|
program: str,
|
|
*,
|
|
args: Optional[list[str]] = None,
|
|
cwd: Optional[str] = None,
|
|
env: Optional[dict[str, str]] = None,
|
|
stopOnEntry=False,
|
|
disableASLR=False,
|
|
disableSTDIO=False,
|
|
shellExpandArguments=False,
|
|
console: Optional[str] = None,
|
|
stdio: Optional[list[str]] = None,
|
|
enableAutoVariableSummaries=False,
|
|
displayExtendedBacktrace=False,
|
|
enableSyntheticChildDebugging=False,
|
|
initCommands: Optional[list[str]] = None,
|
|
preRunCommands: Optional[list[str]] = None,
|
|
launchCommands: Optional[list[str]] = None,
|
|
postRunCommands: Optional[list[str]] = None,
|
|
stopCommands: Optional[list[str]] = None,
|
|
exitCommands: Optional[list[str]] = None,
|
|
terminateCommands: Optional[list[str]] = None,
|
|
sourceMap: Optional[Union[list[tuple[str, str]], dict[str, str]]] = None,
|
|
sourcePath: Optional[str] = None,
|
|
debuggerRoot: Optional[str] = None,
|
|
commandEscapePrefix: Optional[str] = None,
|
|
customFrameFormat: Optional[str] = None,
|
|
customThreadFormat: Optional[str] = None,
|
|
) -> int:
|
|
args_dict: Dict[str, Any] = {"program": program}
|
|
if args:
|
|
args_dict["args"] = args
|
|
if cwd:
|
|
args_dict["cwd"] = cwd
|
|
if env:
|
|
args_dict["env"] = env
|
|
if stopOnEntry:
|
|
args_dict["stopOnEntry"] = stopOnEntry
|
|
if disableSTDIO:
|
|
args_dict["disableSTDIO"] = disableSTDIO
|
|
if shellExpandArguments:
|
|
args_dict["shellExpandArguments"] = shellExpandArguments
|
|
args_dict["initCommands"] = self.init_commands
|
|
if initCommands:
|
|
args_dict["initCommands"].extend(initCommands)
|
|
if preRunCommands:
|
|
args_dict["preRunCommands"] = preRunCommands
|
|
if stopCommands:
|
|
args_dict["stopCommands"] = stopCommands
|
|
if exitCommands:
|
|
args_dict["exitCommands"] = exitCommands
|
|
if terminateCommands:
|
|
args_dict["terminateCommands"] = terminateCommands
|
|
if sourcePath:
|
|
args_dict["sourcePath"] = sourcePath
|
|
if debuggerRoot:
|
|
args_dict["debuggerRoot"] = debuggerRoot
|
|
if launchCommands:
|
|
args_dict["launchCommands"] = launchCommands
|
|
if sourceMap:
|
|
args_dict["sourceMap"] = sourceMap
|
|
if console:
|
|
args_dict["console"] = console
|
|
if stdio:
|
|
args_dict["stdio"] = stdio
|
|
if postRunCommands:
|
|
args_dict["postRunCommands"] = postRunCommands
|
|
if customFrameFormat:
|
|
args_dict["customFrameFormat"] = customFrameFormat
|
|
if customThreadFormat:
|
|
args_dict["customThreadFormat"] = customThreadFormat
|
|
|
|
args_dict["disableASLR"] = disableASLR
|
|
args_dict["enableAutoVariableSummaries"] = enableAutoVariableSummaries
|
|
args_dict["enableSyntheticChildDebugging"] = enableSyntheticChildDebugging
|
|
args_dict["displayExtendedBacktrace"] = displayExtendedBacktrace
|
|
if commandEscapePrefix is not None:
|
|
args_dict["commandEscapePrefix"] = commandEscapePrefix
|
|
command_dict: Request = {
|
|
"seq": 0,
|
|
"command": "launch",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
self.launch_or_attach_sent = True
|
|
return self.send_packet(command_dict)
|
|
|
|
def request_next(self, threadId, granularity="statement"):
|
|
if self.exit_status is not None:
|
|
raise ValueError("request_continue called after process exited")
|
|
args_dict = {"threadId": threadId, "granularity": granularity}
|
|
command_dict = {"command": "next", "type": "request", "arguments": args_dict}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_stepIn(self, threadId, targetId, granularity="statement"):
|
|
if self.exit_status is not None:
|
|
raise ValueError("request_stepIn called after process exited")
|
|
if threadId is None:
|
|
threadId = self.get_thread_id()
|
|
args_dict = {
|
|
"threadId": threadId,
|
|
"targetId": targetId,
|
|
"granularity": granularity,
|
|
}
|
|
command_dict = {"command": "stepIn", "type": "request", "arguments": args_dict}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_stepInTargets(self, frameId):
|
|
if self.exit_status is not None:
|
|
raise ValueError("request_stepInTargets called after process exited")
|
|
self.get_capability("supportsStepInTargetsRequest")
|
|
args_dict = {"frameId": frameId}
|
|
command_dict = {
|
|
"command": "stepInTargets",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_stepOut(self, threadId):
|
|
if self.exit_status is not None:
|
|
raise ValueError("request_stepOut called after process exited")
|
|
args_dict = {"threadId": threadId}
|
|
command_dict = {"command": "stepOut", "type": "request", "arguments": args_dict}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_pause(self, threadId=None):
|
|
if self.exit_status is not None:
|
|
raise ValueError("request_pause called after process exited")
|
|
if threadId is None:
|
|
threadId = self.get_thread_id()
|
|
args_dict = {"threadId": threadId}
|
|
command_dict = {"command": "pause", "type": "request", "arguments": args_dict}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_scopes(self, frameId):
|
|
args_dict = {"frameId": frameId}
|
|
command_dict = {"command": "scopes", "type": "request", "arguments": args_dict}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_setBreakpoints(self, source: Source, line_array, data=None):
|
|
"""data is array of parameters for breakpoints in line_array.
|
|
Each parameter object is 1:1 mapping with entries in line_entry.
|
|
It contains optional location/hitCondition/logMessage parameters.
|
|
"""
|
|
self.ensure_initialized()
|
|
args_dict = {
|
|
"source": source,
|
|
"sourceModified": False,
|
|
}
|
|
if line_array is not None:
|
|
args_dict["lines"] = line_array
|
|
breakpoints = []
|
|
for i, line in enumerate(line_array):
|
|
breakpoint_data = None
|
|
if data is not None and i < len(data):
|
|
breakpoint_data = data[i]
|
|
bp = {"line": line}
|
|
if breakpoint_data is not None:
|
|
if breakpoint_data.get("condition"):
|
|
bp["condition"] = breakpoint_data["condition"]
|
|
if breakpoint_data.get("hitCondition"):
|
|
bp["hitCondition"] = breakpoint_data["hitCondition"]
|
|
if breakpoint_data.get("logMessage"):
|
|
bp["logMessage"] = breakpoint_data["logMessage"]
|
|
if breakpoint_data.get("column"):
|
|
bp["column"] = breakpoint_data["column"]
|
|
breakpoints.append(bp)
|
|
args_dict["breakpoints"] = breakpoints
|
|
|
|
command_dict = {
|
|
"command": "setBreakpoints",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
response = self._send_recv(command_dict)
|
|
if response["success"]:
|
|
self._update_verified_breakpoints(response["body"]["breakpoints"])
|
|
return response
|
|
|
|
def request_setExceptionBreakpoints(
|
|
self, *, filters: list[str] = [], filter_options: list[dict] = []
|
|
):
|
|
self.ensure_initialized()
|
|
args_dict = {"filters": filters}
|
|
if filter_options:
|
|
args_dict["filterOptions"] = filter_options
|
|
command_dict = {
|
|
"command": "setExceptionBreakpoints",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_setFunctionBreakpoints(self, names, condition=None, hitCondition=None):
|
|
self.ensure_initialized()
|
|
breakpoints = []
|
|
for name in names:
|
|
bp = {"name": name}
|
|
if condition is not None:
|
|
bp["condition"] = condition
|
|
if hitCondition is not None:
|
|
bp["hitCondition"] = hitCondition
|
|
breakpoints.append(bp)
|
|
args_dict = {"breakpoints": breakpoints}
|
|
command_dict = {
|
|
"command": "setFunctionBreakpoints",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
response = self._send_recv(command_dict)
|
|
if response["success"]:
|
|
self._update_verified_breakpoints(response["body"]["breakpoints"])
|
|
return response
|
|
|
|
def request_dataBreakpointInfo(
|
|
self, variablesReference, name, size=None, frameIndex=0, threadId=None
|
|
):
|
|
stackFrame = self.get_stackFrame(frameIndex=frameIndex, threadId=threadId)
|
|
if stackFrame is None:
|
|
return []
|
|
args_dict = {"name": name}
|
|
if size is None:
|
|
args_dict["variablesReference"] = variablesReference
|
|
args_dict["frameId"] = stackFrame["id"]
|
|
else:
|
|
args_dict["asAddress"] = True
|
|
args_dict["bytes"] = size
|
|
command_dict = {
|
|
"command": "dataBreakpointInfo",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_setDataBreakpoint(self, dataBreakpoints):
|
|
"""dataBreakpoints is a list of dictionary with following fields:
|
|
{
|
|
dataId: (address in hex)/(size in bytes)
|
|
accessType: read/write/readWrite
|
|
[condition]: string
|
|
[hitCondition]: string
|
|
}
|
|
"""
|
|
self.ensure_initialized()
|
|
args_dict = {"breakpoints": dataBreakpoints}
|
|
command_dict = {
|
|
"command": "setDataBreakpoints",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_compileUnits(self, moduleId):
|
|
args_dict = {"moduleId": moduleId}
|
|
command_dict = {
|
|
"command": "compileUnits",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
response = self._send_recv(command_dict)
|
|
return response
|
|
|
|
def request_completions(self, text, frameId=None):
|
|
def code_units(input: str) -> int:
|
|
utf16_bytes = input.encode("utf-16-le")
|
|
# one UTF16 codeunit = 2 bytes.
|
|
return len(utf16_bytes) // 2
|
|
|
|
args_dict = {"text": text, "column": code_units(text) + 1}
|
|
if frameId:
|
|
args_dict["frameId"] = frameId
|
|
command_dict = {
|
|
"command": "completions",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_modules(
|
|
self,
|
|
start_module: Optional[int] = None,
|
|
module_count: Optional[int] = None,
|
|
):
|
|
args_dict = {}
|
|
|
|
if start_module is not None:
|
|
args_dict["startModule"] = start_module
|
|
if module_count is not None:
|
|
args_dict["moduleCount"] = module_count
|
|
|
|
return self._send_recv(
|
|
{"command": "modules", "type": "request", "arguments": args_dict}
|
|
)
|
|
|
|
def request_moduleSymbols(
|
|
self,
|
|
moduleId: str = "",
|
|
moduleName: str = "",
|
|
startIndex: int = 0,
|
|
count: int = 0,
|
|
):
|
|
command_dict = {
|
|
"command": "__lldb_moduleSymbols",
|
|
"type": "request",
|
|
"arguments": {
|
|
"moduleId": moduleId,
|
|
"moduleName": moduleName,
|
|
"startIndex": startIndex,
|
|
"count": count,
|
|
},
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_stackTrace(
|
|
self, threadId=None, startFrame=None, levels=None, format=None, dump=False
|
|
):
|
|
if threadId is None:
|
|
threadId = self.get_thread_id()
|
|
args_dict = {"threadId": threadId}
|
|
if startFrame is not None:
|
|
args_dict["startFrame"] = startFrame
|
|
if levels is not None:
|
|
args_dict["levels"] = levels
|
|
if format is not None:
|
|
args_dict["format"] = format
|
|
command_dict = {
|
|
"command": "stackTrace",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
response = self._send_recv(command_dict)
|
|
if dump:
|
|
for idx, frame in enumerate(response["body"]["stackFrames"]):
|
|
name = frame["name"]
|
|
if "line" in frame and "source" in frame:
|
|
source = frame["source"]
|
|
if "sourceReference" not in source:
|
|
if "name" in source:
|
|
source_name = source["name"]
|
|
line = frame["line"]
|
|
print("[%3u] %s @ %s:%u" % (idx, name, source_name, line))
|
|
continue
|
|
print("[%3u] %s" % (idx, name))
|
|
return response
|
|
|
|
def request_source(
|
|
self, *, source: Optional[Source] = None, sourceReference: Optional[int] = None
|
|
):
|
|
"""Request a source from a 'Source' reference."""
|
|
if source is None and sourceReference is None:
|
|
raise ValueError("request_source requires either source or sourceReference")
|
|
elif source is not None:
|
|
sourceReference = source["sourceReference"]
|
|
elif sourceReference is not None:
|
|
source = {"sourceReference": sourceReference}
|
|
else:
|
|
raise ValueError(
|
|
"request_source requires either source or sourceReference not both"
|
|
)
|
|
command_dict = {
|
|
"command": "source",
|
|
"type": "request",
|
|
"arguments": {
|
|
"source": source,
|
|
# legacy version of the request
|
|
"sourceReference": sourceReference,
|
|
},
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_threads(self):
|
|
"""Request a list of all threads and combine any information from any
|
|
"stopped" events since those contain more information about why a
|
|
thread actually stopped. Returns an array of thread dictionaries
|
|
with information about all threads"""
|
|
command_dict = {"command": "threads", "type": "request", "arguments": {}}
|
|
response = self._send_recv(command_dict)
|
|
if not response["success"]:
|
|
self.threads = None
|
|
return response
|
|
body = response["body"]
|
|
# Fill in "self.threads" correctly so that clients that call
|
|
# self.get_threads() or self.get_thread_id(...) can get information
|
|
# on threads when the process is stopped.
|
|
if "threads" in body:
|
|
self.threads = body["threads"]
|
|
for thread in self.threads:
|
|
# Copy the thread dictionary so we can add key/value pairs to
|
|
# it without affecting the original info from the "threads"
|
|
# command.
|
|
tid = thread["id"]
|
|
if tid in self.thread_stop_reasons:
|
|
thread_stop_info = self.thread_stop_reasons[tid]
|
|
copy_keys = ["reason", "description", "text", "hitBreakpointIds"]
|
|
for key in copy_keys:
|
|
if key in thread_stop_info:
|
|
thread[key] = thread_stop_info[key]
|
|
else:
|
|
self.threads = None
|
|
return response
|
|
|
|
def request_variables(
|
|
self, variablesReference, start=None, count=None, is_hex: Optional[bool] = None
|
|
):
|
|
args_dict = {"variablesReference": variablesReference}
|
|
if start is not None:
|
|
args_dict["start"] = start
|
|
if count is not None:
|
|
args_dict["count"] = count
|
|
if is_hex is not None:
|
|
args_dict["format"] = {"hex": is_hex}
|
|
command_dict = {
|
|
"command": "variables",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_setVariable(self, containingVarRef, name, value, id=None, is_hex=None):
|
|
args_dict = {
|
|
"variablesReference": containingVarRef,
|
|
"name": name,
|
|
"value": str(value),
|
|
}
|
|
if id is not None:
|
|
args_dict["id"] = id
|
|
if is_hex is not None:
|
|
args_dict["format"] = {"hex": is_hex}
|
|
command_dict = {
|
|
"command": "setVariable",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_locations(self, locationReference):
|
|
args_dict = {
|
|
"locationReference": locationReference,
|
|
}
|
|
command_dict = {
|
|
"command": "locations",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_testGetTargetBreakpoints(self):
|
|
"""A request packet used in the LLDB test suite to get all currently
|
|
set breakpoint infos for all breakpoints currently set in the
|
|
target.
|
|
"""
|
|
command_dict = {
|
|
"command": "_testGetTargetBreakpoints",
|
|
"type": "request",
|
|
"arguments": {},
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def request_custom(self, command: str, arguments: Optional[dict[str, Any]] = None):
|
|
command_dict = {
|
|
"command": command,
|
|
"type": "request",
|
|
"arguments": {} if arguments is None else arguments,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
def terminate(self):
|
|
self.send.close()
|
|
if self._recv_thread.is_alive():
|
|
self._recv_thread.join()
|
|
|
|
def request_setInstructionBreakpoints(self, memory_reference=[]):
|
|
breakpoints = []
|
|
for i in memory_reference:
|
|
args_dict = {
|
|
"instructionReference": i,
|
|
}
|
|
breakpoints.append(args_dict)
|
|
args_dict = {"breakpoints": breakpoints}
|
|
command_dict = {
|
|
"command": "setInstructionBreakpoints",
|
|
"type": "request",
|
|
"arguments": args_dict,
|
|
}
|
|
return self._send_recv(command_dict)
|
|
|
|
|
|
class DebugAdapterServer(DebugCommunication):
|
|
def __init__(
|
|
self,
|
|
*,
|
|
executable: Optional[str] = None,
|
|
connection: Optional[str] = None,
|
|
connection_timeout: Optional[stintr] = None,
|
|
init_commands: Optional[list[str]] = None,
|
|
log_file: Optional[str] = None,
|
|
env: Optional[Dict[str, str]] = None,
|
|
additional_args: Optional[List[str]] = None,
|
|
spawn_helper: Optional[SpawnHelperCallback] = None,
|
|
):
|
|
self.process = None
|
|
self.connection = None
|
|
if executable is not None:
|
|
process, connection = DebugAdapterServer.launch(
|
|
executable=executable,
|
|
connection=connection,
|
|
connection_timeout=connection_timeout,
|
|
env=env,
|
|
log_file=log_file,
|
|
additional_args=additional_args,
|
|
)
|
|
self.process = process
|
|
self.connection = connection
|
|
|
|
if connection is not None:
|
|
scheme, address = connection.split("://")
|
|
if scheme == "unix-connect": # unix-connect:///path
|
|
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
s.connect(address)
|
|
elif scheme == "connection": # connection://[host]:port
|
|
host, port = address.rsplit(":", 1)
|
|
# create_connection with try both ipv4 and ipv6.
|
|
s = socket.create_connection((host.strip("[]"), int(port)))
|
|
else:
|
|
raise ValueError("invalid connection: {}".format(connection))
|
|
super().__init__(
|
|
s.makefile("rb"),
|
|
s.makefile("wb"),
|
|
init_commands,
|
|
spawn_helper,
|
|
)
|
|
self.connection = connection
|
|
else:
|
|
super().__init__(
|
|
self.process.stdout,
|
|
self.process.stdin,
|
|
init_commands,
|
|
spawn_helper,
|
|
)
|
|
|
|
@classmethod
|
|
def launch(
|
|
cls,
|
|
*,
|
|
executable: str,
|
|
env: Optional[Dict[str, str]] = None,
|
|
log_file: Optional[str] = None,
|
|
connection: Optional[str] = None,
|
|
connection_timeout: Optional[int] = None,
|
|
additional_args: Optional[List[str]] = None,
|
|
) -> tuple[subprocess.Popen, Optional[str]]:
|
|
adapter_env = os.environ.copy()
|
|
if env:
|
|
adapter_env.update(env)
|
|
|
|
if log_file:
|
|
adapter_env["LLDBDAP_LOG"] = log_file
|
|
args = [executable]
|
|
|
|
# Add additional arguments first (like --no-lldbinit)
|
|
if additional_args:
|
|
args.extend(additional_args)
|
|
|
|
if connection is not None:
|
|
args.append("--connection")
|
|
args.append(connection)
|
|
|
|
if connection_timeout is not None:
|
|
args.append("--connection-timeout")
|
|
args.append(str(connection_timeout))
|
|
|
|
process = subprocess.Popen(
|
|
args,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=sys.stderr,
|
|
env=adapter_env,
|
|
)
|
|
|
|
if connection is None:
|
|
return (process, None)
|
|
|
|
# lldb-dap will print the listening address once the listener is
|
|
# made to stdout. The listener is formatted like
|
|
# `connection://host:port` or `unix-connection:///path`.
|
|
expected_prefix = "Listening for: "
|
|
out = process.stdout.readline().decode()
|
|
if not out.startswith(expected_prefix):
|
|
process.kill()
|
|
raise ValueError(
|
|
"lldb-dap failed to print listening address, expected '{}', got '{}'".format(
|
|
expected_prefix, out
|
|
)
|
|
)
|
|
|
|
# FIXME: use `str.removeprefix` when LLDB_MINIMUM_PYTHON_VERSION > 3.8
|
|
if out.startswith(expected_prefix):
|
|
out = out[len(expected_prefix) :]
|
|
|
|
# If the listener expanded into multiple addresses, use the first.
|
|
connection = out.rstrip("\r\n").split(",", 1)[0]
|
|
|
|
return (process, connection)
|
|
|
|
def get_pid(self) -> int:
|
|
if self.process:
|
|
return self.process.pid
|
|
return -1
|
|
|
|
def terminate(self):
|
|
try:
|
|
if self.process is not None:
|
|
process = self.process
|
|
self.process = None
|
|
try:
|
|
# When we close stdin it should signal the lldb-dap that no
|
|
# new messages will arrive and it should shutdown on its
|
|
# own.
|
|
process.stdin.close()
|
|
process.wait(timeout=DEFAULT_TIMEOUT)
|
|
except subprocess.TimeoutExpired:
|
|
process.kill()
|
|
process.wait()
|
|
if process.returncode != 0:
|
|
raise DebugAdapterProcessError(process.returncode)
|
|
finally:
|
|
super(DebugAdapterServer, self).terminate()
|
|
|
|
|
|
class DebugAdapterError(Exception):
|
|
pass
|
|
|
|
|
|
class DebugAdapterProcessError(DebugAdapterError):
|
|
"""Raised when the lldb-dap process exits with a non-zero exit status."""
|
|
|
|
def __init__(self, returncode):
|
|
self.returncode = returncode
|
|
|
|
def __str__(self):
|
|
if self.returncode and self.returncode < 0:
|
|
try:
|
|
return f"lldb-dap died with {signal.Signals(-self.returncode).name}."
|
|
except ValueError:
|
|
return f"lldb-dap died with unknown signal {-self.returncode}."
|
|
else:
|
|
return f"lldb-dap returned non-zero exit status {self.returncode}."
|
|
|
|
|
|
def attach_options_specified(opts):
|
|
if opts.pid is not None:
|
|
return True
|
|
if opts.wait_for:
|
|
return True
|
|
if opts.attach:
|
|
return True
|
|
if opts.attach_command:
|
|
return True
|
|
return False
|
|
|
|
|
|
def run_adapter(dbg: DebugCommunication, opts: argparse.Namespace) -> None:
|
|
dbg.request_initialize(sourceInitFile=opts.source_init_file)
|
|
|
|
source_to_lines: Dict[str, List[int]] = {}
|
|
for sbp in cast(List[str], opts.source_bp):
|
|
if ":" not in sbp:
|
|
print(f"error: invalid source with line {sbp!r}", file=sys.stderr)
|
|
continue
|
|
path, line = sbp.split(":")
|
|
if path in source_to_lines:
|
|
source_to_lines[path].append(int(line))
|
|
else:
|
|
source_to_lines[path] = [int(line)]
|
|
for source in source_to_lines:
|
|
dbg.request_setBreakpoints(Source.build(path=source), source_to_lines[source])
|
|
if opts.function_bp:
|
|
dbg.request_setFunctionBreakpoints(opts.function_bp)
|
|
|
|
dbg.request_configurationDone()
|
|
|
|
if attach_options_specified(opts):
|
|
response = dbg.request_attach(
|
|
program=opts.program,
|
|
pid=opts.pid,
|
|
waitFor=opts.wait_for,
|
|
attachCommands=opts.attach_command,
|
|
initCommands=opts.init_command,
|
|
preRunCommands=opts.pre_run_command,
|
|
stopCommands=opts.stop_command,
|
|
terminateCommands=opts.terminate_command,
|
|
exitCommands=opts.exit_command,
|
|
)
|
|
else:
|
|
response = dbg.request_launch(
|
|
opts.program,
|
|
args=opts.args,
|
|
env=opts.env,
|
|
cwd=opts.working_dir,
|
|
debuggerRoot=opts.debugger_root,
|
|
sourceMap=opts.source_map,
|
|
initCommands=opts.init_command,
|
|
preRunCommands=opts.pre_run_command,
|
|
stopCommands=opts.stop_command,
|
|
exitCommands=opts.exit_command,
|
|
terminateCommands=opts.terminate_command,
|
|
)
|
|
|
|
if response["success"]:
|
|
dbg.wait_for_stopped()
|
|
else:
|
|
print("failed to launch/attach: ", response, file=sys.stderr)
|
|
dbg.request_disconnect(terminateDebuggee=True)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(
|
|
prog="dap_server.py",
|
|
description=(
|
|
"A testing framework for the Visual Studio Code Debug Adapter protocol"
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--adapter",
|
|
help=(
|
|
"The path to the command line program that implements the Debug Adapter protocol."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--adapter-arg",
|
|
action="append",
|
|
default=[],
|
|
help="Additional args to pass to the debug adapter.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--program",
|
|
help="The path to the program to debug.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--working-dir",
|
|
help="Set the working directory for the process we launch.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--source-map",
|
|
nargs=2,
|
|
action="extend",
|
|
metavar=("PREFIX", "REPLACEMENT"),
|
|
help=(
|
|
"Source path remappings apply substitutions to the paths of source "
|
|
"files, typically needed to debug from a different host than the "
|
|
"one that built the target."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--debugger-root",
|
|
help=(
|
|
"Set the working directory for lldb-dap for any object files "
|
|
"with relative paths in the Mach-o debug map."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-r",
|
|
"--replay",
|
|
help=(
|
|
"Specify a file containing a packet log to replay with the "
|
|
"current debug adapter."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-g",
|
|
"--debug",
|
|
action="store_true",
|
|
help="Pause waiting for a debugger to attach to the debug adapter.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--source-init-file",
|
|
action="store_true",
|
|
help="Whether lldb-dap should source .lldbinit file or not.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--connection",
|
|
dest="connection",
|
|
help=(
|
|
"Communicate with the debug adapter over specified connection "
|
|
"instead of launching the debug adapter directly."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--pid",
|
|
type=int,
|
|
dest="pid",
|
|
help="The process ID to attach to.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--attach",
|
|
action="store_true",
|
|
help=(
|
|
"Specify this option to attach to a process by name. The "
|
|
"process name is the basename of the executable specified with "
|
|
"the --program option."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-f",
|
|
"--function-bp",
|
|
action="append",
|
|
default=[],
|
|
metavar="FUNCTION",
|
|
help=(
|
|
"Specify the name of a function to break at. Can be specified more "
|
|
"than once."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-s",
|
|
"--source-bp",
|
|
action="append",
|
|
default=[],
|
|
metavar="SOURCE:LINE",
|
|
help="Specify source breakpoints to set. Can be specified more than once.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--attach-command",
|
|
action="append",
|
|
default=[],
|
|
help=(
|
|
"Specify a LLDB command that will attach to a process. "
|
|
"Can be specified more than once."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--init-command",
|
|
action="append",
|
|
default=[],
|
|
help=(
|
|
"Specify a LLDB command that will be executed before the target "
|
|
"is created. Can be specified more than once."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--pre-run-command",
|
|
action="append",
|
|
default=[],
|
|
help=(
|
|
"Specify a LLDB command that will be executed after the target "
|
|
"has been created. Can be specified more than once."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--stop-command",
|
|
action="append",
|
|
default=[],
|
|
help=(
|
|
"Specify a LLDB command that will be executed each time the"
|
|
"process stops. Can be specified more than once."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--exit-command",
|
|
action="append",
|
|
default=[],
|
|
help=(
|
|
"Specify a LLDB command that will be executed when the process "
|
|
"exits. Can be specified more than once."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--terminate-command",
|
|
action="append",
|
|
default=[],
|
|
help=(
|
|
"Specify a LLDB command that will be executed when the debugging "
|
|
"session is terminated. Can be specified more than once."
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"--env",
|
|
action="append",
|
|
default=[],
|
|
metavar="NAME=VALUE",
|
|
help="Specify environment variables to pass to the launched process. Can be specified more than once.",
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-w",
|
|
"--wait-for",
|
|
action="store_true",
|
|
help=(
|
|
"Wait for the next process to be launched whose name matches "
|
|
"the basename of the program specified with the --program "
|
|
"option"
|
|
),
|
|
)
|
|
|
|
parser.add_argument(
|
|
"-v", "--verbose", help="Verbosity level.", action="count", default=0
|
|
)
|
|
|
|
parser.add_argument(
|
|
"args",
|
|
nargs="*",
|
|
help="A list containing all the arguments to be passed to the executable when it is run.",
|
|
)
|
|
|
|
opts = parser.parse_args()
|
|
|
|
logging.basicConfig(
|
|
format="%(asctime)s - %(levelname)s - %(message)s",
|
|
level=(
|
|
logging.DEBUG
|
|
if opts.verbose > 1
|
|
else logging.INFO
|
|
if opts.verbose > 0
|
|
else logging.WARNING
|
|
),
|
|
)
|
|
|
|
if opts.adapter is None and opts.connection is None:
|
|
print(
|
|
"error: must either specify a path to a Debug Protocol Adapter "
|
|
"executable using the --adapter option, or using the --connection "
|
|
"option",
|
|
file=sys.stderr,
|
|
)
|
|
return
|
|
dbg = DebugAdapterServer(
|
|
executable=opts.adapter,
|
|
connection=opts.connection,
|
|
additional_args=opts.adapter_arg,
|
|
)
|
|
if opts.debug:
|
|
input(f"Waiting for debugger to attach pid '{dbg.get_pid()}'")
|
|
try:
|
|
if opts.replay:
|
|
dbg.replay(opts.replay)
|
|
else:
|
|
run_adapter(dbg, opts)
|
|
finally:
|
|
dbg.terminate()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|