================
@@ -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"
+
+
+@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) -> T:
+ event_type = json["event"]
+
+ if event_type not in Event.__registry:
+ raise ValueError(f"event type '{event_type}' is not registered.")
----------------
da-viper wrote:
It does support custom/unknown events. we just have to register it in the test.
```py
@dataclass(frozen=True)
class CustomEvent(Event, event="my-custom-event"): # registered here.
@dataclass
class Body:
key: int
arr: List[bool]
body: Optional[Body] = None
```
see migrated
[TestDAP_sendEvent.py](https://github.com/da-viper/lldb_dap_test/blob/master/tests/TestDAP_sendEvent.py)
https://github.com/llvm/llvm-project/pull/203978
_______________________________________________
lldb-commits mailing list
[email protected]
https://lists.llvm.org/cgi-bin/mailman/listinfo/lldb-commits
