typing

Author: sadym-chromium

tools/webdriver/webdriver/bidi/protocol.py

@@ -10,13 +10,13 @@ class BidiValue(ABC):
     protocol_value: Dict[str, Any]
     type: str
 
-    def __init__(self, protocol_value):
+    def __init__(self, protocol_value: Dict[str, Any]):
         assert isinstance(protocol_value, dict)
         assert isinstance(protocol_value["type"], str)
         self.type = protocol_value["type"]
         self.protocol_value = protocol_value
 
-    def to_classic_protocol_value(self) -> Dict:
+    def to_classic_protocol_value(self) -> Dict[str, Any]:
         """Convert the BiDi value to the classic protocol value. Required for compatibility of the values sent over BiDi
         transport with the classic actions."""
         raise NotImplementedError("No conversion to the classic protocol value is implemented.")
@@ -30,7 +30,7 @@ def __init__(self, protocol_value: Dict[str, Any]):
         assert self.type == "node"
         self.shared_id = self.protocol_value["sharedId"]
 
-    def to_classic_protocol_value(self) -> Dict:
+    def to_classic_protocol_value(self) -> Dict[str, Any]:
         return {WebElement.identifier: self.shared_id}
 
 
@@ -43,7 +43,7 @@ def __init__(self, protocol_value: Dict[str, Any]):
         self.browsing_context = self.protocol_value["value"]["context"]
 
 
-def bidi_deserialize(bidi_value: Union[str, int, Dict]):
+def bidi_deserialize(bidi_value: Union[str, int, Dict[str, Any]]) -> Any:
     """
     Deserialize the BiDi primitive values, lists and objects to the Python value, keeping non-common data types
     in BiDi format.
@@ -82,15 +82,15 @@ def bidi_deserialize(bidi_value: Union[str, int, Dict]):
         return int(bidi_value["value"])
     # script.RemoteValue https://w3c.github.io/webdriver-bidi/#type-script-RemoteValue
     if bidi_value["type"] == "array":
-        result = []
+        list_result = []
         for item in bidi_value["value"]:
-            result.append(bidi_deserialize(item))
-        return result
+            list_result.append(bidi_deserialize(item))
+        return list_result
     if bidi_value["type"] == "object":
-        result = {}
+        dict_result = {}
         for item in bidi_value["value"]:
-            result[bidi_deserialize(item[0])] = bidi_deserialize(item[1])
-        return result
+            dict_result[bidi_deserialize(item[0])] = bidi_deserialize(item[1])
+        return dict_result
     if bidi_value["type"] == "node":
         return BidiNode(bidi_value)
     if bidi_value["type"] == "window":

tools/wptrunner/wptrunner/executors/asyncactions.py

@@ -1,12 +1,14 @@
-from typing import List, Optional, TypedDict, Union
+from typing import Any, List, Mapping, Optional, TypedDict, Union
 from webdriver.bidi.protocol import BidiWindow
+from .protocol import BidiEventsProtocolPart
+from mozlog.structuredlog import StructuredLogger
 
 
 class BrowsingContextArgument(str):
     """Represent a browsing context argument passed from testdriver. It can be either a browsing context id, or a BiDi
     serialized window. In the latter case, the value is extracted from the serialized object."""
 
-    def __new__(cls, context: Union[str, BidiWindow]):
+    def __new__(cls, context: Union[str, BidiWindow]) -> BrowsingContextArgument:
         if isinstance(context, str):
             _context_id = context
         elif isinstance(context, BidiWindow):
@@ -29,18 +31,18 @@ class Payload(TypedDict):
         events: List[str]
         contexts: Optional[List[BrowsingContextArgument]]
 
-    def __init__(self, logger, protocol):
+    def __init__(self, logger: StructuredLogger, protocol: Any):
         self.logger = logger
-        self.protocol = protocol
+        self.bidi_events_protocol: BidiEventsProtocolPart = protocol.bidi_events
 
-    async def __call__(self, payload: Payload):
+    async def __call__(self, payload: Payload) -> Mapping[str, Any]:
         events = payload["events"]
         contexts = None
         if payload["contexts"] is not None:
             contexts = []
             for browsing_context_argument in payload["contexts"]:
-                contexts.append(BrowsingContextArgument(browsing_context_argument))
-        return await self.protocol.bidi_events.subscribe(events, contexts)
+                contexts.append(str(BrowsingContextArgument(browsing_context_argument)))
+        return await self.bidi_events_protocol.subscribe(events, contexts)
 
 
 async_actions = [BidiSessionSubscribeAction]

tools/wptrunner/wptrunner/executors/protocol.py

@@ -205,7 +205,7 @@ def close_old_windows(self, url_protocol):
         pass
 
     @abstractmethod
-    def get_test_window(self, window_id, parent) -> str:
+    def get_test_window(self, window_id: str, parent: str) -> str:
         """Get the window handle dorresponding to the window containing the
         currently active test.
 
@@ -331,7 +331,7 @@ class BidiBrowsingContextProtocolPart(ProtocolPart):
     async def handle_user_prompt(self,
                                  context: str,
                                  accept: Optional[bool] = None,
-                                 user_text: Optional[str] = None):
+                                 user_text: Optional[str] = None) -> None:
         """
         Allows closing an open prompt.
         :param context: The context of the prompt.
@@ -389,7 +389,7 @@ async def call_function(
             function_declaration: str,
             target: Target,
             arguments: Optional[List[Mapping[str, Any]]] = None
-    ):
+    ) -> Mapping[str, Any]:
         """
         Executes the provided script in the given target in asynchronous mode.