python: add typing support with mypy

.pre-commit-config.yaml

@@ -52,6 +52,11 @@ repos:
     rev: 5.12.0
     hooks:
       - id: isort
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.2.0
+    hooks:
+      - id: mypy
+        additional_dependencies: [types-requests]
   - repo: https://github.com/asottile/pyupgrade
     rev: v3.3.1
     hooks:

examples/_helpers.py

@@ -65,11 +65,11 @@ async def get_websocket():
         return await websockets.connect(ws_url)
 
 
-async def send_JSON_command(command, websocket):
+async def send_JSON_command(command: dict, websocket):
     await websocket.send(json.dumps(command))
 
 
-async def read_JSON_message(websocket):
+async def read_JSON_message(websocket) -> dict:
     return json.loads(await websocket.recv())
 
 

tests/conftest.py

@@ -60,7 +60,7 @@ async def create_context():
 
 
 @pytest_asyncio.fixture
-async def default_realm(websocket, context_id):
+async def default_realm(websocket, context_id: str):
     """Return the default realm for the given browsing context."""
     result = await execute_command(
         websocket, {
@@ -78,7 +78,7 @@ async def default_realm(websocket, context_id):
 
 
 @pytest_asyncio.fixture
-async def sandbox_realm(websocket, context_id):
+async def sandbox_realm(websocket, context_id: str):
     """Return a sandbox realm for the given browsing context."""
     result = await execute_command(
         websocket, {
@@ -150,7 +150,7 @@ def html_iframe_same_origin(html, iframe, url_same_origin):
 
 
 @pytest_asyncio.fixture
-async def iframe_id(websocket, context_id, html_iframe_same_origin, html):
+async def iframe_id(websocket, context_id: str, html_iframe_same_origin, html):
     """Navigate to a page with an iframe of the same origin, and return the
     iframe browser context id."""
     await goto_url(websocket, context_id, html_iframe_same_origin)

tests/requirements.txt

@@ -1,4 +1,5 @@
 anys==0.2.1
+mypy==1.2.0
 pytest==7.2.2
 pytest-asyncio==0.21.0
 pytest-icdiff==0.6

tests/test_helpers.py

@@ -22,7 +22,7 @@
 _command_counter = itertools.count(1)
 
 
-def get_next_command_id():
+def get_next_command_id() -> int:
     """
     >>> x = get_next_command_id()
     >>> y = get_next_command_id()
@@ -31,12 +31,15 @@ def get_next_command_id():
     return next(_command_counter)
 
 
-async def subscribe(websocket, event_names, context_ids=None, channel=None):
+async def subscribe(websocket,
+                    event_names: str | list[str],
+                    context_ids: str | list[str] | None = None,
+                    channel: str | None = None):
     if isinstance(event_names, str):
         event_names = [event_names]
     if isinstance(context_ids, str):
         context_ids = [context_ids]
-    command = {
+    command: dict = {
         "method": "session.subscribe",
         "params": {
             "events": event_names
@@ -51,18 +54,18 @@ async def subscribe(websocket, event_names, context_ids=None, channel=None):
     await execute_command(websocket, command)
 
 
-async def send_JSON_command(websocket, command):
+async def send_JSON_command(websocket, command: dict) -> int:
     if "id" not in command:
         command["id"] = get_next_command_id()
     await websocket.send(json.dumps(command))
     return command["id"]
 
 
-async def read_JSON_message(websocket):
+async def read_JSON_message(websocket) -> dict:
     return json.loads(await websocket.recv())
 
 
-async def set_html_content(websocket, context_id, html_content):
+async def set_html_content(websocket, context_id: str, html_content: str):
     """Sets the current page content without navigation."""
     await execute_command(
         websocket, {
@@ -77,7 +80,7 @@ async def set_html_content(websocket, context_id, html_content):
         })
 
 
-async def get_tree(websocket, context_id=None):
+async def get_tree(websocket, context_id: str | None = None) -> dict:
     """Get the tree of browsing contexts."""
     params = {}
     if context_id is not None:
@@ -88,7 +91,10 @@ async def get_tree(websocket, context_id=None):
     })
 
 
-async def goto_url(websocket, context_id, url, wait="interactive"):
+async def goto_url(websocket,
+                   context_id: str,
+                   url: str,
+                   wait: str = "interactive") -> dict:
     """Open given URL in the given context."""
     return await execute_command(
         websocket, {
@@ -101,7 +107,7 @@ async def goto_url(websocket, context_id, url, wait="interactive"):
         })
 
 
-async def execute_command(websocket, command):
+async def execute_command(websocket, command: dict) -> dict:
     if "id" not in command:
         command["id"] = get_next_command_id()
 
@@ -119,7 +125,7 @@ async def execute_command(websocket, command):
             })
 
 
-async def wait_for_event(websocket, event_method):
+async def wait_for_event(websocket, event_method: str) -> dict:
     """Wait and return a specific event from Bidi server."""
     while True:
         event_response = await read_JSON_message(websocket)
@@ -167,15 +173,15 @@ def AnyExtending(expected: list | dict):
     >>> assert {"a": {"a1": [1, 2]}, "b": 2} == AnyExtending({"a": {"a1": [1, 2]}})
     """
     if type(expected) is list:
-        result = []
+        list_result = []
         for index, _ in enumerate(expected):
-            result.append(AnyExtending(expected[index]))
-        return result
+            list_result.append(AnyExtending(expected[index]))
+        return list_result
 
     if type(expected) is dict:
-        result = {}
+        dict_result = {}
         for key in expected.keys():
-            result[key] = AnyExtending(expected[key])
-        return AnyWithEntries(result)
+            dict_result[key] = AnyExtending(expected[key])
+        return AnyWithEntries(dict_result)
 
     return expected