refactor[py]: Add type hints to remote_connection.py

- Added explicit type annotations to selenium.webdriver.remote.remote_connection.py
- Improved code clarity and static type checking
- Ensured compatibility with modern type checkers like Pyright and Mypy

Author: duxtec

py/selenium/webdriver/remote/remote_connection.py

@@ -275,19 +275,11 @@ class RemoteConnection:
     import certifi
 
     _timeout = (
-        float(
-            os.getenv(
-                "GLOBAL_DEFAULT_TIMEOUT", str(socket.getdefaulttimeout())
-            )
-        )
+        float(os.getenv("GLOBAL_DEFAULT_TIMEOUT", str(socket.getdefaulttimeout())))
         if os.getenv("GLOBAL_DEFAULT_TIMEOUT") is not None
         else socket.getdefaulttimeout()
     )
-    _ca_certs = (
-        os.getenv("REQUESTS_CA_BUNDLE")
-        if "REQUESTS_CA_BUNDLE" in os.environ
-        else certifi.where()
-    )
+    _ca_certs = os.getenv("REQUESTS_CA_BUNDLE") if "REQUESTS_CA_BUNDLE" in os.environ else certifi.where()
     _client_config: ClientConfig = None
 
     system = platform.system().lower()
@@ -368,9 +360,7 @@ def set_certificate_bundle_path(cls, path: str):
         cls._client_config.ca_certs = path
 
     @classmethod
-    def get_remote_connection_headers(
-        cls, parsed_url: str, keep_alive: bool = False
-    ) -> dict[str, Any]:
+    def get_remote_connection_headers(cls, parsed_url: str, keep_alive: bool = False) -> dict[str, Any]:
         """Get headers for remote request.
 
         :Args:
@@ -389,12 +379,8 @@ def get_remote_connection_headers(
                 "Embedding username and password in URL could be insecure, use ClientConfig instead",
                 stacklevel=2,
             )
-            base64string = b64encode(
-                f"{parsed_url.username}:{parsed_url.password}".encode()
-            )
-            headers.update(
-                {"Authorization": f"Basic {base64string.decode()}"}
-            )
+            base64string = b64encode(f"{parsed_url.username}:{parsed_url.password}".encode())
+            headers.update({"Authorization": f"Basic {base64string.decode()}"})
 
         if keep_alive:
             headers.update({"Connection": "keep-alive"})
@@ -411,25 +397,19 @@ def _identify_http_proxy_auth(self):
 
     def _separate_http_proxy_auth(self):
         parsed_url = urlparse(self._proxy_url)
-        proxy_without_auth = (
-            f"{parsed_url.scheme}://{parsed_url.hostname}:{parsed_url.port}"
-        )
+        proxy_without_auth = f"{parsed_url.scheme}://{parsed_url.hostname}:{parsed_url.port}"
         auth = f"{parsed_url.username}:{parsed_url.password}"
         return proxy_without_auth, auth
 
     def _get_connection_manager(self):
         pool_manager_init_args = {"timeout": self._client_config.timeout}
         pool_manager_init_args.update(
-            self._client_config.init_args_for_pool_manager.get(
-                "init_args_for_pool_manager", {}
-            )
+            self._client_config.init_args_for_pool_manager.get("init_args_for_pool_manager", {})
         )
 
         if self._client_config.ignore_certificates:
             pool_manager_init_args["cert_reqs"] = "CERT_NONE"
-            urllib3.disable_warnings(
-                urllib3.exceptions.InsecureRequestWarning
-            )
+            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
         elif self._client_config.ca_certs:
             pool_manager_init_args["cert_reqs"] = "CERT_REQUIRED"
             pool_manager_init_args["ca_certs"] = self._client_config.ca_certs
@@ -438,21 +418,11 @@ def _get_connection_manager(self):
             if self._proxy_url.lower().startswith("sock"):
                 from urllib3.contrib.socks import SOCKSProxyManager
 
-                return SOCKSProxyManager(
-                    self._proxy_url, **pool_manager_init_args
-                )
+                return SOCKSProxyManager(self._proxy_url, **pool_manager_init_args)
             if self._identify_http_proxy_auth():
-                self._proxy_url, self._basic_proxy_auth = (
-                    self._separate_http_proxy_auth()
-                )
-                pool_manager_init_args["proxy_headers"] = (
-                    urllib3.make_headers(
-                        proxy_basic_auth=self._basic_proxy_auth
-                    )
-                )
-            return urllib3.ProxyManager(
-                self._proxy_url, **pool_manager_init_args
-            )
+                self._proxy_url, self._basic_proxy_auth = self._separate_http_proxy_auth()
+                pool_manager_init_args["proxy_headers"] = urllib3.make_headers(proxy_basic_auth=self._basic_proxy_auth)
+            return urllib3.ProxyManager(self._proxy_url, **pool_manager_init_args)
 
         return urllib3.PoolManager(**pool_manager_init_args)
 
@@ -476,13 +446,8 @@ def __init__(
         RemoteConnection._timeout = self._client_config.timeout
         RemoteConnection._ca_certs = self._client_config.ca_certs
         RemoteConnection._client_config = self._client_config
-        RemoteConnection.extra_headers = (
-            self._client_config.extra_headers
-            or RemoteConnection.extra_headers
-        )
-        RemoteConnection.user_agent = (
-            self._client_config.user_agent or RemoteConnection.user_agent
-        )
+        RemoteConnection.extra_headers = self._client_config.extra_headers or RemoteConnection.extra_headers
+        RemoteConnection.user_agent = self._client_config.user_agent or RemoteConnection.user_agent
 
         if remote_server_addr:
             warnings.warn(
@@ -547,17 +512,11 @@ def execute(self, command: str, params: dict[Any, Any]) -> dict[str, Any]:
          - params - A dictionary of named parameters to send with the command as
            its JSON payload.
         """
-        command_info = self._commands.get(command) or self.extra_commands.get(
-            command
-        )
+        command_info = self._commands.get(command) or self.extra_commands.get(command)
         assert command_info is not None, f"Unrecognised command {command}"
         path_string = command_info[1]
         path = string.Template(path_string).substitute(params)
-        substitute_params = {
-            word[1:]
-            for word in path_string.split("/")
-            if word.startswith("$")
-        }  # remove dollar sign
+        substitute_params = {word[1:] for word in path_string.split("/") if word.startswith("$")}  # remove dollar sign
         if isinstance(params, dict) and substitute_params:
             for word in substitute_params:
                 del params[word]
@@ -567,9 +526,7 @@ def execute(self, command: str, params: dict[Any, Any]) -> dict[str, Any]:
         LOGGER.debug("%s %s %s", command_info[0], url, str(trimmed))
         return self._request(command_info[0], url, body=data)
 
-    def _request(
-        self, method: str, url: str, body: str | None = None
-    ) -> dict[Any, Any]:
+    def _request(self, method: str, url: str, body: str | None = None) -> dict[Any, Any]:
         """Send an HTTP request to the remote server.
 
         :Args:
@@ -581,9 +538,7 @@ def _request(
           A dictionary with the server's parsed JSON response.
         """
         parsed_url = parse.urlparse(url)
-        headers = self.get_remote_connection_headers(
-            parsed_url, self._client_config.keep_alive
-        )
+        headers = self.get_remote_connection_headers(parsed_url, self._client_config.keep_alive)
         auth_header = self._client_config.get_auth_header()
 
         if auth_header:
@@ -621,9 +576,7 @@ def _request(
         )
         try:
             if 300 <= statuscode < 304:
-                return self._request(
-                    "GET", response.headers.get("location", None)
-                )
+                return self._request("GET", response.headers.get("location", None))
             if 399 < statuscode <= 500:
                 if statuscode == 401:
                     return {
@@ -636,9 +589,7 @@ def _request(
                 }
             content_type = []
             if response.headers.get("Content-Type", None):
-                content_type = response.headers.get(
-                    "Content-Type", None
-                ).split(";")
+                content_type = response.headers.get("Content-Type", None).split(";")
             if not any([x.startswith("image/png") for x in content_type]):
                 try:
                     data = utils.load_json(data.strip())
@@ -665,9 +616,7 @@ def close(self):
         if hasattr(self, "_conn"):
             self._conn.clear()
 
-    def _trim_large_entries(
-        self, input_dict: dict[Any, Any], max_length: int = 100
-    ) -> dict[str, str]:
+    def _trim_large_entries(self, input_dict: dict[Any, Any], max_length: int = 100) -> dict[str, str]:
         """Truncate string values in a dictionary if they exceed max_length.
 
         :param dict: Dictionary with potentially large values
@@ -677,9 +626,7 @@ def _trim_large_entries(
         output_dictionary = {}
         for key, value in input_dict.items():
             if isinstance(value, dict):
-                output_dictionary[key] = self._trim_large_entries(
-                    value, max_length
-                )
+                output_dictionary[key] = self._trim_large_entries(value, max_length)
             elif isinstance(value, str) and len(value) > max_length:
                 output_dictionary[key] = value[:max_length] + "..."
             else: