[Lldb-commits] [lldb] [lldb-dap] Add a new lldb-dap test infrastructure (PR #203978)
Ebuka Ezike via lldb-commits
lldb-commits at lists.llvm.org
Fri Jun 26 08:57:00 PDT 2026
================
@@ -0,0 +1,2025 @@
+# NOTE: this module must not include `from __future__ import annotations`
+# as the `annotations` import changes some of the types hints to strings.
+# especially when you have a forward declared type reference.
+# see https://peps.python.org/pep-0649/#motivation-for-this-pep
+# https://peps.python.org/pep-0749/#rejected-alternatives
+#
+# This module may not depend on any other module.
+
+from contextlib import suppress
+import copy
+import dataclasses
+import enum
+import json
+import os
+import sys
+import typing
+from dataclasses import asdict, dataclass, field, is_dataclass
+from enum import Enum
+from functools import lru_cache
+from typing import (
+ Any,
+ ClassVar,
+ Dict,
+ List,
+ Literal,
+ Optional,
+ Protocol,
+ Set,
+ Tuple,
+ Type,
+ TypeVar,
+ Union,
+ cast,
+ runtime_checkable,
+)
+
+if sys.version_info >= (3, 11):
+ from enum import StrEnum
+else:
+
+ class StrEnum(str, Enum):
+ """Backport of StrEnum for Python < 3.11."""
+
+ def __str__(self) -> str:
+ return self.value
+
+ def __repr__(self) -> str:
+ return self.value
+
+ @staticmethod
+ def _generate_next_value_(name: str, start, count, last_values) -> str:
+ return name.lower()
+
+
+T = TypeVar("T")
+
+
+class DAPError(AssertionError):
+ """The base error for all DAP related errors
+
+ Inherits from assertion error because of unittests treats assertions outside of a test as a failure
+ instead of a test error. see https://docs.python.org/3/library/unittest.html#unittest.TestCase.setUp
+ """
+
+ @classmethod
+ def history_closed(cls, reason=None, last_event: Optional["Event"] = None):
+ suffix = f" (Reason: {reason})" if reason else ""
+ return cls(
+ f"EventHistory is closed{suffix}. Last recorded event: {last_event}."
+ )
+
+
+RawMessage = Dict[str, Any]
+"""Representation of a json protocol message """
+
+
+class MessageType(StrEnum):
+ REQUEST = "request"
+ RESPONSE = "response"
+ EVENT = "event"
+
+
+class EventName(StrEnum):
+ BREAKPOINT = "breakpoint"
+ CAPABILITIES = "capabilities"
+ CONTINUED = "continued"
+ EXITED = "exited"
+ INITIALIZED = "initialized"
+ INVALIDATED = "invalidated"
+ MEMORY = "memory"
+ MODULE = "module"
+ OUTPUT = "output"
+ PROCESS = "process"
+ PROGRESS_END = "progressEnd"
+ PROGRESS_START = "progressStart"
+ PROGRESS_UPDATE = "progressUpdate"
+ STOPPED = "stopped"
+ TERMINATED = "terminated"
+ THREAD = "thread"
+
+
+ at dataclass(frozen=True)
+class ProtocolMessage:
+ type: MessageType
+ seq: int
+
+ def to_dict(self):
+ return message_to_dict(self)
+
+ @classmethod
+ def from_json(cls: Type[T], json: RawMessage) -> T:
+ if not dataclasses.is_dataclass(cls):
+ raise ValueError(f"{cls.__name__} must be a dataclass")
+
+ return dict_to_message(cls, json)
+
+
+ at dataclass(frozen=True)
+class Request(ProtocolMessage):
+ command: str
+ arguments: Any
+
+ def __post_init__(self):
+ assert (
+ self.type == MessageType.REQUEST
+ ), f"expected request type to be 'request' got '{self.type}' in : {self}"
+
+
+ at dataclass(frozen=True)
+class Response(ProtocolMessage):
+ command: str
+ request_seq: int
+ success: bool
+
+ def __post_init__(self):
+ assert (
+ self.type == MessageType.RESPONSE
+ ), f"expected '{type(self).__name__}' to be of type response: {self}"
+
+ def __init_subclass__(cls, **kwargs):
+ super().__init_subclass__(**kwargs)
+ if not dataclasses.is_dataclass(cls):
+ raise TypeError(f"{cls.__name__} must be a dataclass")
+
+
+AnyResponse = TypeVar("AnyResponse", bound=Response)
+
+
+class RequestError(DAPError):
+ """Raised if a DAP request fails."""
+
+ def __init__(
+ self,
+ request: Union[Request, dict],
+ response: Optional[Union[Response, dict]] = 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
+
+
+ at dataclass(frozen=True)
+class Event(ProtocolMessage):
+ event: Union[EventName, str]
+ type: MessageType
+ __registry: ClassVar[Dict[str, Type]] = {}
+
+ def __init_subclass__(cls, *, event: str, **kwargs):
+ super().__init_subclass__(**kwargs)
+
+ assert event is not None
+ # Attach metadata (not a field of an event).
+ cls.__message_type__ = event
+
+ # Prevent duplicate types.
+ existing_event_class = Event.__registry.get(event)
+ if existing_event_class is not None:
+ raise Exception(
+ f"cannot register '{event}' event to class '{cls}' because it is already registered to '{existing_event_class}'"
+ )
+
+ Event.__registry[event] = cls
+
+ def __post_init__(self):
+ if self.type != MessageType.EVENT:
+ raise ValueError(f"event must have type of 'event': {self}.")
+
+ @classmethod
+ def from_json(cls: Type[T], json: dict) -> T:
+ event_type = json["event"]
+
+ if event_type not in Event.__registry:
+ raise ValueError(f"event type '{event_type}' is not registered.")
+
+ event_class = Event.__registry[event_type]
+
+ if cls not in (Event, event_class):
+ raise ValueError(
+ f"class: {cls.__name__!r} is not of the expected type {event_class.__name__!r}."
+ )
+
+ if not dataclasses.is_dataclass(event_class):
+ raise ValueError(f"{cls.__name__!r} must be a dataclass.")
+
+ return cast(T, dict_to_message(event_class, json))
+
+
+AnyEvent = TypeVar("AnyEvent", bound=Event)
+
+
+ at dataclass(frozen=True)
+class EmptyBodyResponse(Response):
+ body: None = field(init=False, default=None)
+
+
+def args_protocol(cls):
+ """Decorator to check a class conforms to the ArgsProtocol"""
+
+ required_fields = filter(lambda x: not x.startswith("_"), dir(ArgsProtocol))
+ for r_field in required_fields:
+ if not hasattr(cls, r_field):
+ raise AttributeError(
+ f"{cls.__name__} must define '{r_field}' to implement ArgProtocol"
+ )
+
+ command_name = getattr(cls, "command_")
+ if not issubclass(type(command_name), str):
+ raise TypeError(
+ f"the command_ type '{type(command_name)}' for class '{cls.__name__}' must be string like"
+ )
+
+ return cls
+
+
+def _message_to_dict_impl(obj: typing.Any, skip_none: bool = True) -> typing.Any:
+ if dataclasses.is_dataclass(obj):
+ fields = _get_dataclass_fields(type(obj))
+ visited: Dict[str, Any] = {}
+ for f in fields:
+ dict_name = f.metadata.get("alias", f.name)
+ if dict_name in visited or not hasattr(obj, f.name):
+ continue
+ name_attr = getattr(obj, f.name)
+ if skip_none and name_attr is None:
+ continue
+ visited[dict_name] = _message_to_dict_impl(name_attr, skip_none)
+ return visited
+
+ if isinstance(obj, (list, tuple)):
+ return type(obj)(_message_to_dict_impl(item, skip_none) for item in obj)
+
+ if isinstance(obj, dict):
+ result: Dict[str, Any] = {}
+ for key, value in obj.items():
+ result[str(key)] = _message_to_dict_impl(value, skip_none)
+ return result
+
+ # Test enum first as it can also be a subclass of other primitives.
+ if isinstance(obj, Enum):
+ return obj.value
+
+ if isinstance(obj, (bool, int, float, str, bytes, type, type(None))):
+ return obj
+
+ return copy.deepcopy(obj)
+
+
+def message_to_dict(args: Any) -> RawMessage:
+ """
+ Converts DAP types to dictionaries.
+ We always skip optional types in dataclasses during conversion.
+ """
+ if not is_dataclass(args):
+ raise TypeError(
+ f"expected a dataclass instance, got {type(args).__name__}: {args!r}"
+ )
+ result = _message_to_dict_impl(args)
+ assert isinstance(result, dict)
+ return result
+
+
+ at lru_cache
+def _get_dataclass_fields(cls: Type) -> Tuple[dataclasses.Field, ...]:
+ data_class_hints = typing.get_type_hints(cls)
+ result: List[dataclasses.Field] = []
+ for f in dataclasses.fields(cls): # noqa
+ # Ignores 'command_' and 'response_class_'.
+ if f.name.endswith("_"):
+ continue
+ f_copy = copy.copy(f)
+ f_copy.type = data_class_hints[f_copy.name]
+ result.append(f_copy)
+ return tuple(result)
+
+
+ at lru_cache
+def _prepare_dataclass_fields(cls: Type) -> Dict[str, dataclasses.Field]:
+ # Excludes init=False fields since they cannot be passed to __init__.
+ return {f.name: f for f in _get_dataclass_fields(cls) if f.init}
+
+
+def _get_compatible_union_types(data: Any, possible_types: List[Type]) -> List[Type]:
+ """Filters a list of candidate types to find those compatible with the given raw data.
+
+ This helps fail early on type mismatches during parsing. For example, it
+ ensures that a raw payload like `[10, 20]` is correctly matched to `List[int]`
+ rather than a primitive `int`.
+ """
+
+ def _is_type_compatible(candidate_type: Type, data: Any) -> bool:
+ """Helper function to check if a single type is compatible with the data."""
+ origin = typing.get_origin(candidate_type)
+
+ if dataclasses.is_dataclass(candidate_type) or origin is dict:
+ return isinstance(data, dict)
+
+ if origin in (list, tuple):
+ return isinstance(data, (list, tuple))
+
+ if candidate_type is bool:
+ return isinstance(data, bool)
+
+ if candidate_type in (int, float):
+ # Prevent implicit bool conversion as bool is a subclass of int.
+ return isinstance(data, (int, float)) and not isinstance(data, bool)
+
+ if candidate_type is str:
+ return isinstance(data, str)
+
+ if isinstance(candidate_type, type) and issubclass(candidate_type, enum.Enum):
+ if issubclass(candidate_type, enum.IntEnum):
+ return isinstance(data, int) and not isinstance(data, bool)
+ return isinstance(data, str)
+
+ # Fallback for unknown or other generic types
+ return True
+
+ result = [a_type for a_type in possible_types if _is_type_compatible(a_type, data)]
+ return result
+
+
+def _generic_to_message(cls: Optional[Type], data: Any, scope: List[str]) -> Any:
+ origin = typing.get_origin(cls)
+ args = typing.get_args(cls)
+ full_path = ".".join(scope)
+
+ if origin is Literal:
+ unique_literals: Set = set()
+
+ def flatten_literal(args: Any):
+ """'Literal' types can be nested e.g.
+ Literal[Literal["book"], Literal["pen"]] is the same as Literal["book", "pen"]
+ """
+ for val in args:
+ if typing.get_origin(val) is None:
+ unique_literals.add(val)
+ else:
+ flatten_literal(typing.get_args(val))
+
+ flatten_literal(args)
+
+ if data not in unique_literals:
+ raise ValueError(
+ f"expected one of {unique_literals!r} at {full_path}, got {data!r}"
+ )
+ return data
+
+ elif origin is Union:
+ none_type = type(None)
+ is_optional = none_type in args
+ candidate_args = [a for a in args if a is not none_type]
+
+ # Handle Optional.
+ # Optional is represented as Union[T, None].
+ if is_optional and len(candidate_args) == 1:
+ return _dict_to_message_impl(candidate_args[0], data, scope)
+
+ matches = _get_compatible_union_types(data, candidate_args)
+ if not matches:
+ raise TypeError(
+ f"no variant of {cls} is compatible with "
+ f"{type(data).__name__} at {full_path}: {data!r}"
+ )
+
+ # Only one match, try conversion.
+ if len(matches) == 1:
+ return _dict_to_message_impl(matches[0], data, scope)
+
+ for arg in matches:
+ with suppress(TypeError, ValueError, AttributeError, AssertionError):
+ return _dict_to_message_impl(arg, data, scope)
+
+ raise TypeError(
+ f"no variant of {cls} matched {type(data).__name__} at {full_path}: {data!r}"
+ )
+
+ elif origin is list:
+ if not isinstance(data, list):
+ raise TypeError(
+ f"expected list at {full_path}, got {type(data).__name__}: {data!r}"
+ )
+ list_type = args[0]
+ return [
+ _dict_to_message_impl(list_type, val, scope + [str(idx)])
+ for idx, val in enumerate(data)
+ ]
+
+ elif origin is tuple:
+ if not isinstance(data, list):
+ raise TypeError(
+ f"expected list for tuple at {full_path}, got {type(data).__name__}: {data!r}"
+ )
+ item_type = args[0]
+ return tuple(
+ _dict_to_message_impl(item_type, val, scope + [str(idx)])
+ for idx, val in enumerate(data)
+ )
+
+ elif origin is dict:
+ if not isinstance(data, dict):
+ raise TypeError(
+ f"expected dict at {full_path}, got {type(data).__name__}: {data!r}"
+ )
+ if len(args) == 0:
+ args = (str, Any)
+ key_type = args[0]
+ value_type = args[1]
+ return {
+ _dict_to_message_impl(key_type, key, scope): _dict_to_message_impl(
+ value_type, value, scope + [str(key)]
+ )
+ for key, value in data.items()
+ }
+
+ raise TypeError(f"unhandled generic type {cls} at {full_path}: {data!r}")
+
+
+def _dict_to_message_impl(
+ cls: Optional[Type], data: typing.Any, scope: List[str]
+) -> typing.Any:
+ """
+ Recursively deserializes a dictionary into a specified Python type or dataclass.
+
+ Args:
+ cls: The target Python type or dataclass to deserialize the data into.
+ data: The raw data (usually from a dictionary or JSON payload) to be converted.
+ scope: A list of keys representing the current depth in the nested data structure.
+ Used to provide error messages when validation fails.
+
+ Returns:
+ The deserialized data cast to the requested type or instantiated dataclass.
+
+ Raises:
+ TypeError: If the data type does not match the expected type and cannot be coerced.
+ ValueError: If a dataclass field has a required value that does not match the data.
+ """
+ if not cls:
+ return data
+
+ full_path = ".".join(scope)
+ if data is None:
+ # Only pass None through if the declared type permits it.
+ if cls is type(None):
+ return None
+ is_union = typing.get_origin(cls) is Union
+ if is_union and type(None) in typing.get_args(cls): # Is Optional
+ return None
+
+ raise TypeError(f"got None for non-optional type '{cls}' at {full_path}")
+
+ if cls in (bytes, bytearray, Any):
+ return data
+
+ if dataclasses.is_dataclass(cls):
+ if not isinstance(data, dict):
+ raise TypeError(
+ f"expected dict for '{cls.__name__}' at {full_path}, "
+ f"got {type(data).__name__}: {data!r}"
+ )
+ fields = _prepare_dataclass_fields(cls)
+ deserialized = {}
+ for key, f in fields.items():
----------------
da-viper wrote:
Updated.
https://github.com/llvm/llvm-project/pull/203978
More information about the lldb-commits
mailing list