more

Author: KazuCocoa

appium/webdriver/appium_connection.py

@@ -13,9 +13,8 @@
 # limitations under the License.
 
 import uuid
-from typing import TYPE_CHECKING, Any, Dict, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, Optional
 
-import urllib3
 from selenium.webdriver.remote.remote_connection import RemoteConnection
 
 from appium.common.helper import library_version
@@ -30,44 +29,7 @@
 class AppiumConnection(RemoteConnection):
     _proxy_url: Optional[str]
 
-    user_agent = f'{PREFIX_HEADER}{library_version()} ({RemoteConnection.user_agent})'
-
-    def __init__(
-        self,
-        remote_server_addr: str,
-        keep_alive: bool = False,
-        ignore_proxy: Optional[bool] = False,
-        init_args_for_pool_manager: Union[Dict[str, Any], None] = None,
-    ):
-        # Need to call before super().__init__ in order to pass arguments for the pool manager in the super.
-        self._init_args_for_pool_manager = init_args_for_pool_manager or {}
-
-        super().__init__(remote_server_addr, keep_alive=keep_alive, ignore_proxy=ignore_proxy)
-
-    def _get_connection_manager(self) -> Union[urllib3.PoolManager, urllib3.ProxyManager]:
-        # https://github.com/SeleniumHQ/selenium/blob/0e0194b0e52a34e7df4b841f1ed74506beea5c3e/py/selenium/webdriver/remote/remote_connection.py#L134
-        pool_manager_init_args = {'timeout': self.get_timeout()}
-
-        if self._ca_certs:
-            pool_manager_init_args['cert_reqs'] = 'CERT_REQUIRED'
-            pool_manager_init_args['ca_certs'] = self._ca_certs
-        else:
-            # This line is necessary to disable certificate verification
-            pool_manager_init_args['cert_reqs'] = 'CERT_NONE'
-
-        pool_manager_init_args.update(self._init_args_for_pool_manager)
-
-        if self._proxy_url:
-            if self._proxy_url.lower().startswith('sock'):
-                from urllib3.contrib.socks import SOCKSProxyManager
-
-                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)
-
-        return urllib3.PoolManager(**pool_manager_init_args)
+    RemoteConnection.user_agent = f'{PREFIX_HEADER}{library_version()} ({RemoteConnection.user_agent})'
 
     @classmethod
     def get_remote_connection_headers(cls, parsed_url: 'ParseResult', keep_alive: bool = True) -> Dict[str, Any]:

test/unit/webdriver/webdriver_test.py

@@ -54,7 +54,7 @@ def test_create_session(self):
 
         request = httpretty.HTTPretty.latest_requests[0]
         assert request.headers['content-type'] == 'application/json;charset=UTF-8'
-        assert f'appium/{appium_version.version} (selenium' in request.headers['user-agent']
+        assert request.headers['user-agent'] == 'appium/4.1.0 (selenium/4.25.0 (python mac)'
 
         request_json = json.loads(httpretty.HTTPretty.latest_requests[0].body.decode('utf-8'))
         assert request_json.get('capabilities') is not None