[do not merge]

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,51 +29,16 @@
 class AppiumConnection(RemoteConnection):
     _proxy_url: Optional[str]
 
-    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]:
         """Override get_remote_connection_headers in RemoteConnection"""
         headers = RemoteConnection.get_remote_connection_headers(parsed_url, keep_alive=keep_alive)
-        # e.g. appium/0.49 (selenium/3.141.0 (python linux))
-        headers['User-Agent'] = f'{PREFIX_HEADER}{library_version()} ({headers["User-Agent"]})'
         if parsed_url.path.endswith('/session'):
             # https://github.com/appium/appium-base-driver/pull/400
-            headers['X-Idempotency-Key'] = str(uuid.uuid4())
+            RemoteConnection.extra_headers = {'X-Idempotency-Key': str(uuid.uuid4())}
+        else:
+            RemoteConnection.extra_headers = {}
 
         return headers

appium/webdriver/extensions/context.py

@@ -58,8 +58,6 @@ def context(self) -> str:
         return self.current_context
 
     def _add_commands(self) -> None:
-        # noinspection PyProtectedMember,PyUnresolvedReferences
-        commands = self.command_executor._commands
-        commands[Command.CONTEXTS] = ('GET', '/session/$sessionId/contexts')
-        commands[Command.GET_CURRENT_CONTEXT] = ('GET', '/session/$sessionId/context')
-        commands[Command.SWITCH_TO_CONTEXT] = ('POST', '/session/$sessionId/context')
+        self.command_executor.add_command(Command.CONTEXTS, 'GET', '/session/$sessionId/contexts')
+        self.command_executor.add_command(Command.GET_CURRENT_CONTEXT, 'GET', '/session/$sessionId/context')
+        self.command_executor.add_command(Command.SWITCH_TO_CONTEXT, 'POST', '/session/$sessionId/context')

appium/webdriver/webdriver.py

@@ -62,6 +62,11 @@
 from .webelement import WebElement as MobileWebElement
 
 
+class AppiumLocatorConverter:
+    def convert(self, by, value):
+        return (by, value)
+
+
 class ExtensionBase:
     """
     Used to define an extension command as driver's methods.
@@ -226,6 +231,8 @@ def __init__(
         super().__init__(
             command_executor=command_executor,
             options=options,
+            locator_converter=AppiumLocatorConverter(),
+            web_element_cls=MobileWebElement,
         )
 
         if hasattr(self, 'command_executor'):
@@ -346,72 +353,6 @@ def get_status(self) -> Dict:
         """
         return self.execute(Command.GET_STATUS)['value']
 
-    def find_element(self, by: str = AppiumBy.ID, value: Union[str, Dict, None] = None) -> MobileWebElement:
-        """
-        Find an element given a AppiumBy strategy and locator
-
-        Args:
-            by: The strategy
-            value: The locator
-
-        Usage:
-            driver.find_element(by=AppiumBy.ACCESSIBILITY_ID, value='accessibility_id')
-
-        Returns:
-            `appium.webdriver.webelement.WebElement`: The found element
-
-        """
-        # We prefer to patch locators in the client code
-        # Checking current context every time a locator is accessed could significantly slow down tests
-        # Check https://github.com/appium/python-client/pull/724 before submitting any issue
-        # if by == By.ID:
-        #     by = By.CSS_SELECTOR
-        #     value = '[id="%s"]' % value
-        # elif by == By.TAG_NAME:
-        #     by = By.CSS_SELECTOR
-        # elif by == By.CLASS_NAME:
-        #     by = By.CSS_SELECTOR
-        #     value = ".%s" % value
-        # elif by == By.NAME:
-        #     by = By.CSS_SELECTOR
-        #     value = '[name="%s"]' % value
-
-        return self.execute(RemoteCommand.FIND_ELEMENT, {'using': by, 'value': value})['value']
-
-    def find_elements(self, by: str = AppiumBy.ID, value: Union[str, Dict, None] = None) -> Union[List[MobileWebElement], List]:
-        """
-        Find elements given a AppiumBy strategy and locator
-
-        Args:
-            by: The strategy
-            value: The locator
-
-        Usage:
-            driver.find_elements(by=AppiumBy.ACCESSIBILITY_ID, value='accessibility_id')
-
-        Returns:
-            :obj:`list` of :obj:`appium.webdriver.webelement.WebElement`: The found elements
-        """
-        # We prefer to patch locators in the client code
-        # Checking current context every time a locator is accessed could significantly slow down tests
-        # Check https://github.com/appium/python-client/pull/724 before submitting any issue
-        # if by == By.ID:
-        #     by = By.CSS_SELECTOR
-        #     value = '[id="%s"]' % value
-        # elif by == By.TAG_NAME:
-        #     by = By.CSS_SELECTOR
-        # elif by == By.CLASS_NAME:
-        #     by = By.CSS_SELECTOR
-        #     value = ".%s" % value
-        # elif by == By.NAME:
-        #     by = By.CSS_SELECTOR
-        #     value = '[name="%s"]' % value
-
-        # Return empty list if driver returns null
-        # See https://github.com/SeleniumHQ/selenium/issues/4555
-
-        return self.execute(RemoteCommand.FIND_ELEMENTS, {'using': by, 'value': value})['value'] or []
-
     def create_web_element(self, element_id: Union[int, str]) -> MobileWebElement:
         """Creates a web element with the specified element_id.
 

appium/webdriver/webelement.py

@@ -12,15 +12,13 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from typing import Callable, Dict, List, Optional, Union
+from typing import Callable, Dict, Optional, Union
 
 from selenium.webdriver.common.utils import keys_to_typing
 from selenium.webdriver.remote.command import Command as RemoteCommand
 from selenium.webdriver.remote.webelement import WebElement as SeleniumWebElement
 from typing_extensions import Self
 
-from appium.webdriver.common.appiumby import AppiumBy
-
 from .mobilecommand import MobileCommand as Command
 
 
@@ -80,70 +78,6 @@ def is_displayed(self) -> bool:
         """
         return self._execute(Command.IS_ELEMENT_DISPLAYED)['value']
 
-    def find_element(self, by: str = AppiumBy.ID, value: Union[str, Dict, None] = None) -> 'WebElement':
-        """Find an element given a AppiumBy strategy and locator
-
-        Override for Appium
-
-        Prefer the find_element_by_* methods when possible.
-
-        Args:
-            by: The strategy
-            value: The locator
-
-        Usage:
-            element = element.find_element(AppiumBy.ID, 'foo')
-
-        Returns:
-            `appium.webdriver.webelement.WebElement`
-        """
-        # We prefer to patch locators in the client code
-        # Checking current context every time a locator is accessed could significantly slow down tests
-        # Check https://github.com/appium/python-client/pull/724 before submitting any issue
-        # if by == By.ID:
-        #     by = By.CSS_SELECTOR
-        #     value = '[id="%s"]' % value
-        # elif by == By.TAG_NAME:
-        #     by = By.CSS_SELECTOR
-        # elif by == By.CLASS_NAME:
-        #     by = By.CSS_SELECTOR
-        #     value = ".%s" % value
-        # elif by == By.NAME:
-        #     by = By.CSS_SELECTOR
-        #     value = '[name="%s"]' % value
-
-        return self._execute(RemoteCommand.FIND_CHILD_ELEMENT, {'using': by, 'value': value})['value']
-
-    def find_elements(self, by: str = AppiumBy.ID, value: Union[str, Dict, None] = None) -> List['WebElement']:
-        """Find elements given a AppiumBy strategy and locator
-
-        Args:
-            by: The strategy
-            value: The locator
-
-        Usage:
-            element = element.find_elements(AppiumBy.CLASS_NAME, 'foo')
-
-        Returns:
-            :obj:`list` of :obj:`appium.webdriver.webelement.WebElement`
-        """
-        # We prefer to patch locators in the client code
-        # Checking current context every time a locator is accessed could significantly slow down tests
-        # Check https://github.com/appium/python-client/pull/724 before submitting any issue
-        # if by == By.ID:
-        #     by = By.CSS_SELECTOR
-        #     value = '[id="%s"]' % value
-        # elif by == By.TAG_NAME:
-        #     by = By.CSS_SELECTOR
-        # elif by == By.CLASS_NAME:
-        #     by = By.CSS_SELECTOR
-        #     value = ".%s" % value
-        # elif by == By.NAME:
-        #     by = By.CSS_SELECTOR
-        #     value = '[name="%s"]' % value
-
-        return self._execute(RemoteCommand.FIND_CHILD_ELEMENTS, {'using': by, 'value': value})['value']
-
     def clear(self) -> Self:
         """Clears text.
 

test/unit/webdriver/search_context/android_test.py

@@ -22,6 +22,62 @@
 
 
 class TestWebDriverAndroidSearchContext(object):
+    @httpretty.activate
+    def test_find_element_by_id(self):
+        driver = android_w3c_driver()
+        httpretty.register_uri(
+            httpretty.POST,
+            appium_command('/session/1234567890/element'),
+            body='{"value": {"element-6066-11e4-a52e-4f735466cecf": "element-id"}}',
+        )
+        el = driver.find_element(
+            by=AppiumBy.ID,
+            value='id data',
+        )
+
+        d = get_httpretty_request_body(httpretty.last_request())
+        assert d['using'] == 'id'
+        assert d['value'] == 'id data'
+        assert isinstance(el, MobileWebElement)
+
+    @httpretty.activate
+    def test_find_elements_by_id(self):
+        driver = android_w3c_driver()
+        httpretty.register_uri(
+            httpretty.POST,
+            appium_command('/session/1234567890/elements'),
+            body='{"value": [{"element-6066-11e4-a52e-4f735466cecf": "element-id1"}, '
+            '{"element-6066-11e4-a52e-4f735466cecf": "element-id2"}]}',
+        )
+        els = driver.find_elements(
+            by=AppiumBy.ID,
+            value='id data',
+        )
+
+        d = get_httpretty_request_body(httpretty.last_request())
+        assert d['using'] == 'id'
+        assert d['value'] == 'id data'
+        assert isinstance(els[0], MobileWebElement)
+
+    @httpretty.activate
+    def test_find_child_element_by_id(self):
+        driver = android_w3c_driver()
+        element = MobileWebElement(driver, 'element_id')
+        httpretty.register_uri(
+            httpretty.POST,
+            appium_command('/session/1234567890/element/element_id/element'),
+            body='{"value": {"element-6066-11e4-a52e-4f735466cecf": "child-element-id"}}',
+        )
+        el = element.find_element(
+            by=AppiumBy.ID,
+            value='id data',
+        )
+
+        d = get_httpretty_request_body(httpretty.last_request())
+        assert d['using'] == 'id'
+        assert d['value'] == 'id data'
+        assert isinstance(el, MobileWebElement)
+
     @httpretty.activate
     def test_find_element_by_android_data_matcher(self):
         driver = android_w3c_driver()

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