================ @@ -0,0 +1,2022 @@ +# 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, + 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" + + +@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) + + +@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}" + + +@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 + + +@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, check_class: bool = True) -> T: + event_type = json["event"] + + if event_type not in Event.__registry: ---------------- DrSergei wrote:
Looks like current protocol is too strict. I think it will be problem in tests for [custom events](https://github.com/llvm/llvm-project/tree/main/lldb/tools/lldb-dap#lldb-dap-send-event) and custom requests https://github.com/llvm/llvm-project/pull/203978 _______________________________________________ lldb-commits mailing list [email protected] https://lists.llvm.org/cgi-bin/mailman/listinfo/lldb-commits
