refactor(debugger): enhance type safety by introducing protocols and type definitions for log messages and attributes

Author: d-biehl

packages/debugger/src/robotcode/debugger/debugger.py

@@ -23,6 +23,7 @@
     MutableMapping,
     NamedTuple,
     Optional,
+    Protocol,
     Sequence,
     Set,
     Tuple,
@@ -106,6 +107,40 @@ def __repr__(self) -> str:
 MAX_VARIABLE_ITEMS_DISPLAY = 500  # Maximum items to display in variable view
 
 
+# Type definitions for better type safety
+EvaluationResult = Union[Any, Exception]
+KeywordCallable = Callable[[], EvaluationResult]
+AttributeDict = Dict[str, Any]
+
+
+class ExceptionInformation(TypedDict, total=False):
+    text: Optional[str]
+    description: str
+    status: str
+
+
+class LogMessage(TypedDict, total=False):
+    level: str
+    message: str
+    timestamp: str
+    html: Optional[str]
+
+
+class RobotContextProtocol(Protocol):
+    """Protocol for Robot Framework execution context."""
+
+    variables: Any
+    namespace: Any
+
+
+class KeywordHandlerProtocol(Protocol):
+    """Protocol for Robot Framework keyword handlers."""
+
+    name: str
+    args: Any  # Robot version dependent type
+    arguments: Any  # Robot version dependent type
+
+
 class DebugRepr(reprlib.Repr):
     def __init__(self) -> None:
         super().__init__()
@@ -259,12 +294,14 @@ class PathMapping(NamedTuple):
     remote_root: Optional[str]
 
 
-if get_robot_version() < (7, 0):
+class DebugLoggerBase:
+    def __init__(self) -> None:
+        self.steps: List[Any] = []
+
 
-    class DebugLogger:
-        def __init__(self) -> None:
-            self.steps: List[Any] = []
+if get_robot_version() < (7, 0):
 
+    class DebugLogger(DebugLoggerBase):
         def start_keyword(self, kw: Any) -> None:
             self.steps.append(kw)
 
@@ -275,10 +312,7 @@ def end_keyword(self, kw: Any) -> None:
     from robot import result, running
     from robot.output.loggerapi import LoggerApi
 
-    class DebugLogger(LoggerApi):  # type: ignore[no-redef]
-        def __init__(self) -> None:
-            self.steps: List[Any] = []
-
+    class DebugLogger(DebugLoggerBase, LoggerApi):  # type: ignore[no-redef]
         def start_try(self, data: "running.Try", result: "result.Try") -> None:
             self.steps.append(data)
 
@@ -295,12 +329,6 @@ def end_keyword(self, data: running.Keyword, result: result.Keyword) -> None:
 breakpoint_id_manager = IdManager()
 
 
-class ExceptionInformation(TypedDict):
-    text: Optional[str]
-    description: str
-    status: str
-
-
 class _DebuggerInstanceDescriptor:
     """Descriptor that forwards all attribute access to the singleton instance."""
 
@@ -395,8 +423,8 @@ def __init__(self) -> None:
         self.attached = False
         self.path_mappings: List[PathMapping] = []
 
-        self._keyword_to_evaluate: Optional[Callable[..., Any]] = None
-        self._evaluated_keyword_result: Any = None
+        self._keyword_to_evaluate: Optional[KeywordCallable] = None
+        self._evaluated_keyword_result: Optional[EvaluationResult] = None
         self._evaluate_keyword_event = threading.Event()
         self._evaluate_keyword_event.set()
         self._after_evaluate_keyword_event = threading.Event()
@@ -843,7 +871,7 @@ def wait_for_running(self) -> None:
                 break
             self._current_exception = None
 
-    def start_output_group(self, name: str, attributes: Dict[str, Any], type: Optional[str] = None) -> None:
+    def start_output_group(self, name: str, attributes: AttributeDict, type: Optional[str] = None) -> None:
         if self.group_output:
             source = attributes.get("source")
             line_no = attributes.get("lineno")
@@ -862,7 +890,7 @@ def start_output_group(self, name: str, attributes: Dict[str, Any], type: Option
                 ),
             )
 
-    def end_output_group(self, name: str, attributes: Dict[str, Any], type: Optional[str] = None) -> None:
+    def end_output_group(self, name: str, attributes: AttributeDict, type: Optional[str] = None) -> None:
         if self.group_output:
             source = attributes.get("source")
             line_no = attributes.get("lineno")
@@ -889,7 +917,7 @@ def add_stackframe_entry(
         line: Optional[int],
         column: Optional[int] = None,
         *,
-        handler: Any = None,
+        handler: Optional[KeywordHandlerProtocol] = None,
         libname: Optional[str] = None,
         kwname: Optional[str] = None,
         longname: Optional[str] = None,
@@ -943,7 +971,7 @@ def remove_stackframe_entry(
         line: Optional[int],
         column: Optional[int] = None,
         *,
-        handler: Any = None,
+        handler: Optional[KeywordHandlerProtocol] = None,
     ) -> None:
         self.full_stack_frames.popleft()
 
@@ -961,7 +989,7 @@ def remove_stackframe_entry(
             if self.stack_frames:
                 self.stack_frames[0].stack_frames.popleft()
 
-    def start_suite(self, name: str, attributes: Dict[str, Any]) -> None:
+    def start_suite(self, name: str, attributes: AttributeDict) -> None:
         if self.state == State.CallKeyword:
             return
 
@@ -1009,7 +1037,7 @@ def start_suite(self, name: str, attributes: Dict[str, Any]) -> None:
 
                 self.wait_for_running()
 
-    def end_suite(self, name: str, attributes: Dict[str, Any]) -> None:
+    def end_suite(self, name: str, attributes: AttributeDict) -> None:
         if self.state == State.CallKeyword:
             return
 
@@ -1031,7 +1059,7 @@ def end_suite(self, name: str, attributes: Dict[str, Any]) -> None:
 
         self.remove_stackframe_entry(name, type, source, line_no)
 
-    def start_test(self, name: str, attributes: Dict[str, Any]) -> None:
+    def start_test(self, name: str, attributes: AttributeDict) -> None:
         if self.state == State.CallKeyword:
             return
 
@@ -1058,7 +1086,7 @@ def start_test(self, name: str, attributes: Dict[str, Any]) -> None:
 
             self.wait_for_running()
 
-    def end_test(self, name: str, attributes: Dict[str, Any]) -> None:
+    def end_test(self, name: str, attributes: AttributeDict) -> None:
         if self.state == State.CallKeyword:
             return
 
@@ -1092,7 +1120,7 @@ def get_current_keyword_handler(self, name: str) -> UserKeywordHandler:
         def get_current_keyword_handler(self, name: str) -> UserKeywordHandler:
             return EXECUTION_CONTEXTS.current.namespace.get_runner(name)._handler
 
-    def start_keyword(self, name: str, attributes: Dict[str, Any]) -> None:
+    def start_keyword(self, name: str, attributes: AttributeDict) -> None:
         if self.state == State.CallKeyword:
             return
 
@@ -1107,7 +1135,7 @@ def start_keyword(self, name: str, attributes: Dict[str, Any]) -> None:
         libname = attributes.get("libname")
         kwname = attributes.get("kwname")
 
-        handler: Any = None
+        handler: Optional[KeywordHandlerProtocol] = None
         if type in ["KEYWORD", "SETUP", "TEARDOWN"]:
             try:
                 handler = self.get_current_keyword_handler(name)
@@ -1222,7 +1250,7 @@ def is_not_caugthed_by_except(self, message: Optional[str]) -> bool:
                                 return False
         return True
 
-    def end_keyword(self, name: str, attributes: Dict[str, Any]) -> None:
+    def end_keyword(self, name: str, attributes: AttributeDict) -> None:
         if self.state == State.CallKeyword:
             return
 
@@ -1253,7 +1281,7 @@ def end_keyword(self, name: str, attributes: Dict[str, Any]) -> None:
         type = attributes.get("type", "KEYWORD")
         kwname = attributes.get("kwname")
 
-        handler: Any = None
+        handler: Optional[KeywordHandlerProtocol] = None
         if type in ["KEYWORD", "SETUP", "TEARDOWN"]:
             try:
                 handler = self.get_current_keyword_handler(name)
@@ -1378,7 +1406,7 @@ def yield_stack() -> Iterator[StackFrame]:
         "DEBUG": "\u001b[38;5;8m",
     }
 
-    def log_message(self, message: Dict[str, Any]) -> None:
+    def log_message(self, message: LogMessage) -> None:
         level = message["level"]
         msg = message["message"]
 
@@ -1438,7 +1466,7 @@ def _build_output(self, level: str, msg: str, timestamp: str) -> str:
             + f"{msg}\n"
         )
 
-    def message(self, message: Dict[str, Any]) -> None:
+    def message(self, message: LogMessage) -> None:
         level = message["level"]
         current_frame = self.full_stack_frames[0] if self.full_stack_frames else None
 
@@ -1905,7 +1933,7 @@ def _create_evaluate_result(self, value: Any) -> EvaluateResult:
 
         return EvaluateResult(result=repr(value), type=repr(type(value)))
 
-    def run_in_robot_thread(self, kw: Callable[[], Any]) -> Any:
+    def run_in_robot_thread(self, kw: KeywordCallable) -> EvaluationResult:
         with self.condition:
             self._keyword_to_evaluate = kw
             self._evaluated_keyword_result = None

packages/debugger/src/robotcode/debugger/listeners.py

@@ -9,7 +9,7 @@
 from robotcode.core.utils.path import normalized_path
 
 from .dap_types import Event, Model
-from .debugger import Debugger
+from .debugger import Debugger, LogMessage
 from .mixins import SyncedEventBody
 
 
@@ -220,7 +220,7 @@ def end_keyword(self, name: str, attributes: Dict[str, Any]) -> None:
 
     RE_FILE_LINE_MATCHER = re.compile(r".+\sin\sfile\s'(?P<file>.*)'\son\sline\s(?P<line>\d+):(?P<message>.*)")
 
-    def log_message(self, message: Dict[str, Any]) -> None:
+    def log_message(self, message: LogMessage) -> None:
         if message["level"] in ["FAIL", "ERROR", "WARN"]:
             current_frame = Debugger.instance.full_stack_frames[0] if Debugger.instance.full_stack_frames else None
 
@@ -273,7 +273,7 @@ def log_message(self, message: Dict[str, Any]) -> None:
 
         Debugger.instance.log_message(message)
 
-    def message(self, message: Dict[str, Any]) -> None:
+    def message(self, message: LogMessage) -> None:
         if message["level"] in ["FAIL", "ERROR", "WARN"]:
             current_frame = Debugger.instance.full_stack_frames[0] if Debugger.instance.full_stack_frames else None