diff --git a/Makefile b/Makefile index 0308854c..0170afb2 100644 --- a/Makefile +++ b/Makefile @@ -7,7 +7,7 @@ check-all: ## Run all lint checks and unittest .PHONY: isort isort: ## Run isort - python -m isort $(ARGS) . + python -m isort --profile black $(ARGS) . .PHONY: black black: ## Run black diff --git a/Pipfile b/Pipfile index c58b1f53..66b0ac0f 100644 --- a/Pipfile +++ b/Pipfile @@ -4,10 +4,10 @@ url = "https://pypi.org/simple" verify_ssl = true [dev-packages] -black = "~=22.12.0" +black = "<24.0.0" httpretty = "~=1.1" -isort = "~=5.11" -mypy = "~=1.2" +isort = "<6.0" +mypy = "<2.0" mock = "~=5.0" pre-commit = "~=2.21" pylint = "~=2.17.3" diff --git a/appium/protocols/webdriver/can_remember_extension_presence.py b/appium/protocols/webdriver/can_remember_extension_presence.py new file mode 100644 index 00000000..bf000393 --- /dev/null +++ b/appium/protocols/webdriver/can_remember_extension_presence.py @@ -0,0 +1,27 @@ +#!/usr/bin/env python + +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from typing import TypeVar + +from ..protocol import Protocol + +T = TypeVar('T') + + +class CanRememberExtensionPresence(Protocol): + def assert_extension_exists(self: T, ext_name: str) -> T: + ... + + def mark_extension_absence(self: T, ext_name: str) -> T: + ... diff --git a/appium/webdriver/appium_connection.py b/appium/webdriver/appium_connection.py index ce9f296f..87915e86 100644 --- a/appium/webdriver/appium_connection.py +++ b/appium/webdriver/appium_connection.py @@ -32,7 +32,6 @@ def __init__( 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 {} diff --git a/appium/webdriver/extensions/action_helpers.py b/appium/webdriver/extensions/action_helpers.py index e124450d..8c424a40 100644 --- a/appium/webdriver/extensions/action_helpers.py +++ b/appium/webdriver/extensions/action_helpers.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import List, Optional, Tuple, TypeVar +from typing import TYPE_CHECKING, List, Optional, Tuple, cast from selenium.webdriver.common.action_chains import ActionChains from selenium.webdriver.common.actions import interaction @@ -20,14 +20,14 @@ from selenium.webdriver.common.actions.mouse_button import MouseButton from selenium.webdriver.common.actions.pointer_input import PointerInput -from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands from appium.webdriver.webelement import WebElement -T = TypeVar('T', bound=CanExecuteCommands) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver class ActionHelpers: - def scroll(self: T, origin_el: WebElement, destination_el: WebElement, duration: Optional[int] = None) -> T: + def scroll(self, origin_el: WebElement, destination_el: WebElement, duration: Optional[int] = None) -> 'WebDriver': """Scrolls from one element to another Args: @@ -59,9 +59,9 @@ def scroll(self: T, origin_el: WebElement, destination_el: WebElement, duration: actions.w3c_actions.pointer_action.move_to(destination_el) actions.w3c_actions.pointer_action.release() actions.perform() - return self + return cast('WebDriver', self) - def drag_and_drop(self: T, origin_el: WebElement, destination_el: WebElement) -> T: + def drag_and_drop(self, origin_el: WebElement, destination_el: WebElement) -> 'WebDriver': """Drag the origin element to the destination element Args: @@ -77,9 +77,9 @@ def drag_and_drop(self: T, origin_el: WebElement, destination_el: WebElement) -> actions.w3c_actions.pointer_action.move_to(destination_el) actions.w3c_actions.pointer_action.release() actions.perform() - return self + return cast('WebDriver', self) - def tap(self: T, positions: List[Tuple[int, int]], duration: Optional[int] = None) -> T: + def tap(self, positions: List[Tuple[int, int]], duration: Optional[int] = None) -> 'WebDriver': """Taps on an particular place with up to five fingers, holding for a certain time @@ -127,9 +127,9 @@ def tap(self: T, positions: List[Tuple[int, int]], duration: Optional[int] = Non new_input.create_pause(0.1) new_input.create_pointer_up(MouseButton.LEFT) actions.perform() - return self + return cast('WebDriver', self) - def swipe(self: T, start_x: int, start_y: int, end_x: int, end_y: int, duration: int = 0) -> T: + def swipe(self, start_x: int, start_y: int, end_x: int, end_y: int, duration: int = 0) -> 'WebDriver': """Swipe from one point to another point, for an optional duration. Args: @@ -156,9 +156,9 @@ def swipe(self: T, start_x: int, start_y: int, end_x: int, end_y: int, duration: actions.w3c_actions.pointer_action.move_to_location(end_x, end_y) actions.w3c_actions.pointer_action.release() actions.perform() - return self + return cast('WebDriver', self) - def flick(self: T, start_x: int, start_y: int, end_x: int, end_y: int) -> T: + def flick(self, start_x: int, start_y: int, end_x: int, end_y: int) -> 'WebDriver': """Flick from one point to another point. Args: @@ -180,4 +180,4 @@ def flick(self: T, start_x: int, start_y: int, end_x: int, end_y: int) -> T: actions.w3c_actions.pointer_action.move_to_location(end_x, end_y) actions.w3c_actions.pointer_action.release() actions.perform() - return self + return cast('WebDriver', self) diff --git a/appium/webdriver/extensions/android/power.py b/appium/webdriver/extensions/android/power.py index c3d762f8..c1b9b211 100644 --- a/appium/webdriver/extensions/android/power.py +++ b/appium/webdriver/extensions/android/power.py @@ -21,7 +21,6 @@ class Power(CanExecuteCommands): - AC_OFF, AC_ON = 'off', 'on' def set_power_capacity(self: T, percent: int) -> T: diff --git a/appium/webdriver/extensions/applications.py b/appium/webdriver/extensions/applications.py index ff9b7aec..d4951b9d 100644 --- a/appium/webdriver/extensions/applications.py +++ b/appium/webdriver/extensions/applications.py @@ -12,30 +12,40 @@ # See the License for the specific language governing permissions and # limitations under the License. import warnings -from typing import Any, Dict, TypeVar, Union +from typing import TYPE_CHECKING, Any, Dict, Union, cast + +from selenium.common.exceptions import InvalidArgumentException, UnknownMethodException from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands +from appium.protocols.webdriver.can_execute_scripts import CanExecuteScripts +from appium.protocols.webdriver.can_remember_extension_presence import CanRememberExtensionPresence from ..mobilecommand import MobileCommand as Command -T = TypeVar('T', bound=CanExecuteCommands) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver -class Applications(CanExecuteCommands): - def background_app(self: T, seconds: int) -> T: +class Applications(CanExecuteCommands, CanExecuteScripts, CanRememberExtensionPresence): + def background_app(self, seconds: int) -> 'WebDriver': """Puts the application in the background on the device for a certain duration. Args: - seconds: the duration for the application to remain in the background + seconds: the duration for the application to remain in the background. + Providing a negative value will continue immediately after putting the app + under test to the background. Returns: Union['WebDriver', 'Applications']: Self instance """ - data = { - 'seconds': seconds, - } - self.execute(Command.BACKGROUND, data) - return self + ext_name = 'mobile: backgroundApp' + args = {'seconds': seconds} + try: + self.assert_extension_exists(ext_name).execute_script(ext_name, args) + except UnknownMethodException: + # TODO: Remove the fallback + self.mark_extension_absence(ext_name).execute(Command.BACKGROUND, args) + return cast('WebDriver', self) def is_app_installed(self, bundle_id: str) -> bool: """Checks whether the application specified by `bundle_id` is installed on the device. @@ -46,12 +56,25 @@ def is_app_installed(self, bundle_id: str) -> bool: Returns: `True` if app is installed """ - data = { - 'bundleId': bundle_id, - } - return self.execute(Command.IS_APP_INSTALLED, data)['value'] - - def install_app(self: T, app_path: str, **options: Any) -> T: + ext_name = 'mobile: isAppInstalled' + try: + return self.assert_extension_exists(ext_name).execute_script( + ext_name, + { + 'bundleId': bundle_id, + 'appId': bundle_id, + }, + ) + except (UnknownMethodException, InvalidArgumentException): + # TODO: Remove the fallback + return self.mark_extension_absence(ext_name).execute( + Command.IS_APP_INSTALLED, + { + 'bundleId': bundle_id, + }, + )['value'] + + def install_app(self, app_path: str, **options: Any) -> 'WebDriver': """Install the application found at `app_path` on the device. Args: @@ -71,15 +94,25 @@ def install_app(self: T, app_path: str, **options: Any) -> T: Returns: Union['WebDriver', 'Applications']: Self instance """ - data: Dict[str, Any] = { - 'appPath': app_path, - } - if options: - data.update({'options': options}) - self.execute(Command.INSTALL_APP, data) - return self - - def remove_app(self: T, app_id: str, **options: Any) -> T: + ext_name = 'mobile: installApp' + try: + self.assert_extension_exists(ext_name).execute_script( + 'mobile: installApp', + { + 'app': app_path, + 'appPath': app_path, + **(options or {}), + }, + ) + except (UnknownMethodException, InvalidArgumentException): + # TODO: Remove the fallback + data: Dict[str, Any] = {'appPath': app_path} + if options: + data.update({'options': options}) + self.mark_extension_absence(ext_name).execute(Command.INSTALL_APP, data) + return cast('WebDriver', self) + + def remove_app(self, app_id: str, **options: Any) -> 'WebDriver': """Remove the specified application from the device. Args: @@ -94,15 +127,25 @@ def remove_app(self: T, app_id: str, **options: Any) -> T: Returns: Union['WebDriver', 'Applications']: Self instance """ - data: Dict[str, Any] = { - 'appId': app_id, - } - if options: - data.update({'options': options}) - self.execute(Command.REMOVE_APP, data) - return self - - def launch_app(self: T) -> T: + ext_name = 'mobile: removeApp' + try: + self.assert_extension_exists(ext_name).execute_script( + ext_name, + { + 'appId': app_id, + 'bundleId': app_id, + **(options or {}), + }, + ) + except (UnknownMethodException, InvalidArgumentException): + # TODO: Remove the fallback + data: Dict[str, Any] = {'appId': app_id} + if options: + data.update({'options': options}) + self.mark_extension_absence(ext_name).execute(Command.REMOVE_APP, data) + return cast('WebDriver', self) + + def launch_app(self) -> 'WebDriver': """Start on the device the application specified in the desired capabilities. deprecated:: 2.0.0 @@ -116,9 +159,9 @@ def launch_app(self: T) -> T: ) self.execute(Command.LAUNCH_APP) - return self + return cast('WebDriver', self) - def close_app(self: T) -> T: + def close_app(self) -> 'WebDriver': """Stop the running application, specified in the desired capabilities, on the device. deprecated:: 2.0.0 @@ -133,7 +176,7 @@ def close_app(self: T) -> T: ) self.execute(Command.CLOSE_APP) - return self + return cast('WebDriver', self) def terminate_app(self, app_id: str, **options: Any) -> bool: """Terminates the application if it is running. @@ -148,14 +191,24 @@ def terminate_app(self, app_id: str, **options: Any) -> bool: Returns: True if the app has been successfully terminated """ - data: Dict[str, Any] = { - 'appId': app_id, - } - if options: - data.update({'options': options}) - return self.execute(Command.TERMINATE_APP, data)['value'] - - def activate_app(self: T, app_id: str) -> T: + ext_name = 'mobile: terminateApp' + try: + return self.assert_extension_exists(ext_name).execute_script( + ext_name, + { + 'appId': app_id, + 'bundleId': app_id, + **(options or {}), + }, + ) + except (UnknownMethodException, InvalidArgumentException): + # TODO: Remove the fallback + data: Dict[str, Any] = {'appId': app_id} + if options: + data.update({'options': options}) + return self.mark_extension_absence(ext_name).execute(Command.TERMINATE_APP, data)['value'] + + def activate_app(self, app_id: str) -> 'WebDriver': """Activates the application if it is not running or is running in the background. @@ -165,11 +218,19 @@ def activate_app(self: T, app_id: str) -> T: Returns: Union['WebDriver', 'Applications']: Self instance """ - data = { - 'appId': app_id, - } - self.execute(Command.ACTIVATE_APP, data) - return self + ext_name = 'mobile: activateApp' + try: + self.assert_extension_exists(ext_name).execute_script( + ext_name, + { + 'appId': app_id, + 'bundleId': app_id, + }, + ) + except (UnknownMethodException, InvalidArgumentException): + # TODO: Remove the fallback + self.mark_extension_absence(ext_name).execute(Command.ACTIVATE_APP, {'appId': app_id}) + return cast('WebDriver', self) def query_app_state(self, app_id: str) -> int: """Queries the state of the application. @@ -181,10 +242,23 @@ def query_app_state(self, app_id: str) -> int: One of possible application state constants. See ApplicationState class for more details. """ - data = { - 'appId': app_id, - } - return self.execute(Command.QUERY_APP_STATE, data)['value'] + ext_name = 'mobile: queryAppState' + try: + return self.assert_extension_exists(ext_name).execute_script( + ext_name, + { + 'appId': app_id, + 'bundleId': app_id, + }, + ) + except (UnknownMethodException, InvalidArgumentException): + # TODO: Remove the fallback + return self.mark_extension_absence(ext_name).execute( + Command.QUERY_APP_STATE, + { + 'appId': app_id, + }, + )['value'] def app_strings(self, language: Union[str, None] = None, string_file: Union[str, None] = None) -> Dict[str, str]: """Returns the application strings from the device for the specified @@ -192,19 +266,24 @@ def app_strings(self, language: Union[str, None] = None, string_file: Union[str, Args: language: strings language code - string_file: the name of the string file to query + string_file: the name of the string file to query. Only relevant for XCUITest driver Returns: The key is string id and the value is the content. """ + ext_name = 'mobile: getAppStrings' data = {} if language is not None: data['language'] = language if string_file is not None: data['stringFile'] = string_file - return self.execute(Command.GET_APP_STRINGS, data)['value'] + try: + return self.assert_extension_exists(ext_name).execute_script(ext_name, data) + except UnknownMethodException: + # TODO: Remove the fallback + return self.mark_extension_absence(ext_name).execute(Command.GET_APP_STRINGS, data)['value'] - def reset(self: T) -> T: + def reset(self) -> 'WebDriver': """Resets the current application on the device. deprecated:: 2.0.0 @@ -218,7 +297,7 @@ def reset(self: T) -> T: ) self.execute(Command.RESET) - return self + return cast('WebDriver', self) def _add_commands(self) -> None: # noinspection PyProtectedMember,PyUnresolvedReferences diff --git a/appium/webdriver/extensions/clipboard.py b/appium/webdriver/extensions/clipboard.py index 4009a93c..de7f2390 100644 --- a/appium/webdriver/extensions/clipboard.py +++ b/appium/webdriver/extensions/clipboard.py @@ -13,20 +13,21 @@ # limitations under the License. import base64 -from typing import Optional, TypeVar +from typing import TYPE_CHECKING, Optional, cast from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands from appium.webdriver.clipboard_content_type import ClipboardContentType from ..mobilecommand import MobileCommand as Command -T = TypeVar('T', bound='Clipboard') +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver class Clipboard(CanExecuteCommands): def set_clipboard( - self: T, content: bytes, content_type: str = ClipboardContentType.PLAINTEXT, label: Optional[str] = None - ) -> T: + self, content: bytes, content_type: str = ClipboardContentType.PLAINTEXT, label: Optional[str] = None + ) -> 'WebDriver': """Set the content of the system clipboard Args: @@ -45,9 +46,9 @@ def set_clipboard( if label: options['label'] = label self.execute(Command.SET_CLIPBOARD, options) - return self + return cast('WebDriver', self) - def set_clipboard_text(self: T, text: str, label: Optional[str] = None) -> T: + def set_clipboard_text(self, text: str, label: Optional[str] = None) -> 'WebDriver': """Copies the given text to the system clipboard Args: diff --git a/appium/webdriver/extensions/device_time.py b/appium/webdriver/extensions/device_time.py index 3bf0d44e..a81717eb 100644 --- a/appium/webdriver/extensions/device_time.py +++ b/appium/webdriver/extensions/device_time.py @@ -14,12 +14,16 @@ from typing import Optional +from selenium.common.exceptions import UnknownMethodException + from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands +from appium.protocols.webdriver.can_execute_scripts import CanExecuteScripts +from appium.protocols.webdriver.can_remember_extension_presence import CanRememberExtensionPresence from ..mobilecommand import MobileCommand as Command -class DeviceTime(CanExecuteCommands): +class DeviceTime(CanExecuteCommands, CanExecuteScripts, CanRememberExtensionPresence): @property def device_time(self) -> str: """Returns the date and time from the device. @@ -27,7 +31,12 @@ def device_time(self) -> str: Return: str: The date and time """ - return self.execute(Command.GET_DEVICE_TIME_GET, {})['value'] + ext_name = 'mobile: getDeviceTime' + try: + return self.assert_extension_exists(ext_name).execute_script(ext_name) + except UnknownMethodException: + # TODO: Remove the fallback + return self.mark_extension_absence(ext_name).execute(Command.GET_DEVICE_TIME_GET, {})['value'] def get_device_time(self, format: Optional[str] = None) -> str: """Returns the date and time from the device. @@ -45,9 +54,15 @@ def get_device_time(self, format: Optional[str] = None) -> str: Return: str: The date and time """ + ext_name = 'mobile: getDeviceTime' if format is None: return self.device_time - return self.execute(Command.GET_DEVICE_TIME_POST, {'format': format})['value'] + try: + return self.assert_extension_exists(ext_name).execute_script(ext_name, {'format': format}) + except UnknownMethodException: + return self.mark_extension_absence(ext_name).execute(Command.GET_DEVICE_TIME_POST, {'format': format})[ + 'value' + ] def _add_commands(self) -> None: # noinspection PyProtectedMember,PyUnresolvedReferences diff --git a/appium/webdriver/extensions/execute_driver.py b/appium/webdriver/extensions/execute_driver.py index 010c297d..fcfff282 100644 --- a/appium/webdriver/extensions/execute_driver.py +++ b/appium/webdriver/extensions/execute_driver.py @@ -20,7 +20,6 @@ class ExecuteDriver(CanExecuteCommands): - # TODO Inner class case def execute_driver(self, script: str, script_type: str = 'webdriverio', timeout_ms: Optional[int] = None) -> Any: """Run a set of script against the current session, allowing execution of many commands in one Appium request. diff --git a/appium/webdriver/extensions/execute_mobile_command.py b/appium/webdriver/extensions/execute_mobile_command.py index 2518db4c..39f217d5 100644 --- a/appium/webdriver/extensions/execute_mobile_command.py +++ b/appium/webdriver/extensions/execute_mobile_command.py @@ -12,15 +12,16 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Any, Dict, TypeVar +from typing import TYPE_CHECKING, Any, Dict, cast from appium.protocols.webdriver.can_execute_scripts import CanExecuteScripts -T = TypeVar('T', bound=CanExecuteScripts) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver class ExecuteMobileCommand(CanExecuteScripts): - def press_button(self: T, button_name: str) -> T: + def press_button(self, button_name: str) -> 'WebDriver': """Sends a physical button name to the device to simulate the user pressing. iOS only. @@ -36,7 +37,7 @@ def press_button(self: T, button_name: str) -> T: """ data = {'name': button_name} self.execute_script('mobile: pressButton', data) - return self + return cast('WebDriver', self) @property def battery_info(self) -> Dict[str, Any]: diff --git a/appium/webdriver/extensions/hw_actions.py b/appium/webdriver/extensions/hw_actions.py index 75ca1482..2b284511 100644 --- a/appium/webdriver/extensions/hw_actions.py +++ b/appium/webdriver/extensions/hw_actions.py @@ -12,17 +12,22 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Any, Optional, TypeVar +from typing import TYPE_CHECKING, Optional, cast + +from selenium.common.exceptions import UnknownMethodException from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands +from appium.protocols.webdriver.can_execute_scripts import CanExecuteScripts +from appium.protocols.webdriver.can_remember_extension_presence import CanRememberExtensionPresence from ..mobilecommand import MobileCommand as Command -T = TypeVar('T', bound=CanExecuteCommands) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver -class HardwareActions(CanExecuteCommands): - def lock(self: T, seconds: Optional[int] = None) -> T: +class HardwareActions(CanExecuteCommands, CanExecuteScripts, CanRememberExtensionPresence): + def lock(self, seconds: Optional[int] = None) -> 'WebDriver': """Lock the device. No changes are made if the device is already unlocked. Args: @@ -34,20 +39,30 @@ def lock(self: T, seconds: Optional[int] = None) -> T: Returns: Union['WebDriver', 'HardwareActions']: Self instance """ - if seconds is None: - self.execute(Command.LOCK) - else: - self.execute(Command.LOCK, {'seconds': seconds}) - return self - - def unlock(self: T) -> T: + ext_name = 'mobile: lock' + args = {'seconds': seconds or 0} + try: + self.assert_extension_exists(ext_name).execute_script(ext_name, args) + except UnknownMethodException: + # TODO: Remove the fallback + self.mark_extension_absence(ext_name).execute(Command.LOCK, args) + return cast('WebDriver', self) + + def unlock(self) -> 'WebDriver': """Unlock the device. No changes are made if the device is already locked. Returns: Union['WebDriver', 'HardwareActions']: Self instance """ - self.execute(Command.UNLOCK) - return self + ext_name = 'mobile: unlock' + try: + if not self.assert_extension_exists(ext_name).execute_script('mobile: isLocked'): + return cast('WebDriver', self) + self.execute_script(ext_name) + except UnknownMethodException: + # TODO: Remove the fallback + self.mark_extension_absence(ext_name).execute(Command.UNLOCK) + return cast('WebDriver', self) def is_locked(self) -> bool: """Checks whether the device is locked. @@ -55,18 +70,28 @@ def is_locked(self) -> bool: Returns: `True` if the device is locked """ - return self.execute(Command.IS_LOCKED)['value'] - - def shake(self: T) -> T: + ext_name = 'mobile: isLocked' + try: + return self.assert_extension_exists(ext_name).execute_script('mobile: isLocked') + except UnknownMethodException: + # TODO: Remove the fallback + return self.mark_extension_absence(ext_name).execute(Command.IS_LOCKED)['value'] + + def shake(self) -> 'WebDriver': """Shake the device. Returns: Union['WebDriver', 'HardwareActions']: Self instance """ - self.execute(Command.SHAKE) - return self - - def touch_id(self: T, match: bool) -> T: + ext_name = 'mobile: shake' + try: + self.assert_extension_exists(ext_name).execute_script(ext_name) + except UnknownMethodException: + # TODO: Remove the fallback + self.mark_extension_absence(ext_name).execute(Command.SHAKE) + return cast('WebDriver', self) + + def touch_id(self, match: bool) -> 'WebDriver': """Simulate touchId on iOS Simulator Args: @@ -75,29 +100,38 @@ def touch_id(self: T, match: bool) -> T: Returns: Union['WebDriver', 'HardwareActions']: Self instance """ - data = {'match': match} - self.execute(Command.TOUCH_ID, data) - return self + self.execute_script( + 'mobile: sendBiometricMatch', + { + 'type': 'touchId', + 'match': match, + }, + ) + return cast('WebDriver', self) - def toggle_touch_id_enrollment(self: T) -> T: + def toggle_touch_id_enrollment(self) -> 'WebDriver': """Toggle enroll touchId on iOS Simulator Returns: Union['WebDriver', 'HardwareActions']: Self instance """ - self.execute(Command.TOGGLE_TOUCH_ID_ENROLLMENT) - return self + is_enrolled = self.execute_script('mobile: isBiometricEnrolled') + self.execute_script('mobile: enrollBiometric', {'isEnabled': not is_enrolled}) + return cast('WebDriver', self) - def finger_print(self, finger_id: int) -> Any: + def finger_print(self, finger_id: int) -> 'WebDriver': """Authenticate users by using their finger print scans on supported Android emulators. Args: finger_id: Finger prints stored in Android Keystore system (from 1 to 10) - - Returns: - TODO """ - return self.execute(Command.FINGER_PRINT, {'fingerprintId': finger_id})['value'] + ext_name = 'mobile: fingerprint' + args = {'fingerprintId': finger_id} + try: + self.assert_extension_exists(ext_name).execute_script(ext_name, args) + except UnknownMethodException: + self.mark_extension_absence(ext_name).execute(Command.FINGER_PRINT, args) + return cast('WebDriver', self) def _add_commands(self) -> None: # noinspection PyProtectedMember,PyUnresolvedReferences diff --git a/appium/webdriver/extensions/ime.py b/appium/webdriver/extensions/ime.py index 24f1c676..2a5f188c 100644 --- a/appium/webdriver/extensions/ime.py +++ b/appium/webdriver/extensions/ime.py @@ -12,13 +12,15 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import List, TypeVar +import warnings +from typing import TYPE_CHECKING, List, cast from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands from ..mobilecommand import MobileCommand as Command -T = TypeVar('T', bound=CanExecuteCommands) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver class IME(CanExecuteCommands): @@ -29,25 +31,43 @@ def available_ime_engines(self) -> List[str]: Package and activity are returned (e.g., ['com.android.inputmethod.latin/.LatinIME']) Android only. + deprecated:: 2.0.0 + Returns: :obj:`list` of :obj:`str`: The available input methods for an Android device """ + warnings.warn( + 'The "available_ime_engines" API is deprecated and will be removed in future versions. ' + 'Use "mobile: shell" extension instead', + DeprecationWarning, + ) + return self.execute(Command.GET_AVAILABLE_IME_ENGINES, {})['value'] # pylint: disable=unsubscriptable-object def is_ime_active(self) -> bool: """Checks whether the device has IME service active. Android only. + deprecated:: 2.0.0 + Returns: `True` if IME service is active """ + warnings.warn( + 'The "is_ime_active" API is deprecated and will be removed in future versions. ' + 'Use "mobile: shell" extension instead', + DeprecationWarning, + ) + return self.execute(Command.IS_IME_ACTIVE, {})['value'] # pylint: disable=unsubscriptable-object - def activate_ime_engine(self: T, engine: str) -> T: + def activate_ime_engine(self, engine: str) -> 'WebDriver': """Activates the given IME engine on the device. Android only. + deprecated:: 2.0.0 + Args: engine: the package and activity of the IME engine to activate (e.g., 'com.android.inputmethod.latin/.LatinIME') @@ -55,20 +75,34 @@ def activate_ime_engine(self: T, engine: str) -> T: Returns: Union['WebDriver', 'IME']: Self instance """ + warnings.warn( + 'The "activate_ime_engine" API is deprecated and will be removed in future versions. ' + 'Use "mobile: shell" extension instead', + DeprecationWarning, + ) + data = {'engine': engine} self.execute(Command.ACTIVATE_IME_ENGINE, data) - return self + return cast('WebDriver', self) - def deactivate_ime_engine(self: T) -> T: + def deactivate_ime_engine(self) -> 'WebDriver': """Deactivates the currently active IME engine on the device. Android only. + deprecated:: 2.0.0 + Returns: Union['WebDriver', 'IME']: Self instance """ + warnings.warn( + 'The "deactivate_ime_engine" API is deprecated and will be removed in future versions. ' + 'Use "mobile: shell" extension instead', + DeprecationWarning, + ) + self.execute(Command.DEACTIVATE_IME_ENGINE, {}) - return self + return cast('WebDriver', self) @property def active_ime_engine(self) -> str: @@ -77,9 +111,17 @@ def active_ime_engine(self) -> str: Android only. + deprecated:: 2.0.0 + Returns: str: The activity and package of the currently active IME engine """ + warnings.warn( + 'The "active_ime_engine" API is deprecated and will be removed in future versions. ' + 'Use "mobile: shell" extension instead', + DeprecationWarning, + ) + return self.execute(Command.GET_ACTIVE_IME_ENGINE, {})['value'] # pylint: disable=unsubscriptable-object def _add_commands(self) -> None: diff --git a/appium/webdriver/extensions/keyboard.py b/appium/webdriver/extensions/keyboard.py index 9e7491a7..7dfba4f7 100644 --- a/appium/webdriver/extensions/keyboard.py +++ b/appium/webdriver/extensions/keyboard.py @@ -12,19 +12,24 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Dict, Optional, TypeVar +from typing import TYPE_CHECKING, Dict, Optional, cast + +from selenium.common.exceptions import UnknownMethodException from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands +from appium.protocols.webdriver.can_execute_scripts import CanExecuteScripts +from appium.protocols.webdriver.can_remember_extension_presence import CanRememberExtensionPresence from ..mobilecommand import MobileCommand as Command -T = TypeVar('T', bound=CanExecuteCommands) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver -class Keyboard(CanExecuteCommands): +class Keyboard(CanExecuteCommands, CanExecuteScripts, CanRememberExtensionPresence): def hide_keyboard( - self: T, key_name: Optional[str] = None, key: Optional[str] = None, strategy: Optional[str] = None - ) -> T: + self, key_name: Optional[str] = None, key: Optional[str] = None, strategy: Optional[str] = None + ) -> 'WebDriver': """Hides the software keyboard on the device. In iOS, use `key_name` to press @@ -38,16 +43,23 @@ def hide_keyboard( Returns: Union['WebDriver', 'Keyboard']: Self instance """ - data: Dict[str, Optional[str]] = {} - if key_name is not None: - data['keyName'] = key_name - elif key is not None: - data['key'] = key - elif strategy is None: - strategy = 'tapOutside' - data['strategy'] = strategy - self.execute(Command.HIDE_KEYBOARD, data) - return self + ext_name = 'mobile: hideKeyboard' + try: + self.assert_extension_exists(ext_name).execute_script( + ext_name, {**({'keys': [key or key_name]} if key or key_name else {})} + ) + except UnknownMethodException: + # TODO: Remove the fallback + data: Dict[str, Optional[str]] = {} + if key_name is not None: + data['keyName'] = key_name + elif key is not None: + data['key'] = key + elif strategy is None: + strategy = 'tapOutside' + data['strategy'] = strategy + self.mark_extension_absence(ext_name).execute(Command.HIDE_KEYBOARD, data) + return cast('WebDriver', self) def is_keyboard_shown(self) -> bool: """Attempts to detect whether a software keyboard is present @@ -55,9 +67,13 @@ def is_keyboard_shown(self) -> bool: Returns: `True` if keyboard is shown """ - return self.execute(Command.IS_KEYBOARD_SHOWN)['value'] + ext_name = 'mobile: isKeyboardShown' + try: + return self.assert_extension_exists(ext_name).execute_script(ext_name) + except UnknownMethodException: + return self.mark_extension_absence(ext_name).execute(Command.IS_KEYBOARD_SHOWN)['value'] - def keyevent(self: T, keycode: int, metastate: Optional[int] = None) -> T: + def keyevent(self, keycode: int, metastate: Optional[int] = None) -> 'WebDriver': """Sends a keycode to the device. Android only. @@ -70,15 +86,9 @@ def keyevent(self: T, keycode: int, metastate: Optional[int] = None) -> T: Returns: Union['WebDriver', 'Keyboard']: Self instance """ - data = { - 'keycode': keycode, - } - if metastate is not None: - data['metastate'] = metastate - self.execute(Command.KEY_EVENT, data) - return self + return self.press_keycode(keycode=keycode, metastate=metastate) - def press_keycode(self: T, keycode: int, metastate: Optional[int] = None, flags: Optional[int] = None) -> T: + def press_keycode(self, keycode: int, metastate: Optional[int] = None, flags: Optional[int] = None) -> 'WebDriver': """Sends a keycode to the device. Android only. Possible keycodes can be found @@ -92,17 +102,22 @@ def press_keycode(self: T, keycode: int, metastate: Optional[int] = None, flags: Returns: Union['WebDriver', 'Keyboard']: Self instance """ - data = { - 'keycode': keycode, - } + ext_name = 'mobile: pressKey' + args = {'keycode': keycode} if metastate is not None: - data['metastate'] = metastate + args['metastate'] = metastate if flags is not None: - data['flags'] = flags - self.execute(Command.PRESS_KEYCODE, data) - return self - - def long_press_keycode(self: T, keycode: int, metastate: Optional[int] = None, flags: Optional[int] = None) -> T: + args['flags'] = flags + try: + self.assert_extension_exists(ext_name).execute_script(ext_name, args) + except UnknownMethodException: + # TODO: Remove the fallback + self.mark_extension_absence(ext_name).execute(Command.PRESS_KEYCODE, args) + return cast('WebDriver', self) + + def long_press_keycode( + self, keycode: int, metastate: Optional[int] = None, flags: Optional[int] = None + ) -> 'WebDriver': """Sends a long press of keycode to the device. Android only. Possible keycodes can be found in @@ -116,13 +131,24 @@ def long_press_keycode(self: T, keycode: int, metastate: Optional[int] = None, f Returns: Union['WebDriver', 'Keyboard']: Self instance """ - data = {'keycode': keycode} + ext_name = 'mobile: pressKey' + args = {'keycode': keycode} if metastate is not None: - data['metastate'] = metastate + args['metastate'] = metastate if flags is not None: - data['flags'] = flags - self.execute(Command.LONG_PRESS_KEYCODE, data) - return self + args['flags'] = flags + try: + self.assert_extension_exists(ext_name).execute_script( + ext_name, + { + **args, + 'isLongPress': True, + }, + ) + except UnknownMethodException: + # TODO: Remove the fallback + self.mark_extension_absence(ext_name).execute(Command.LONG_PRESS_KEYCODE, args) + return cast('WebDriver', self) def _add_commands(self) -> None: # noinspection PyProtectedMember,PyUnresolvedReferences diff --git a/appium/webdriver/extensions/location.py b/appium/webdriver/extensions/location.py index aff5390c..906e4437 100644 --- a/appium/webdriver/extensions/location.py +++ b/appium/webdriver/extensions/location.py @@ -12,35 +12,44 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Dict, TypeVar, Union +from typing import TYPE_CHECKING, Dict, Union, cast + +from selenium.common.exceptions import UnknownMethodException from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands +from appium.protocols.webdriver.can_execute_scripts import CanExecuteScripts from ..mobilecommand import MobileCommand as Command -T = TypeVar('T', bound=CanExecuteCommands) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver -class Location(CanExecuteCommands): - def toggle_location_services(self: T) -> T: +class Location(CanExecuteCommands, CanExecuteScripts): + def toggle_location_services(self) -> 'WebDriver': """Toggle the location services on the device. + This API only reliably since Android 12 (API level 31) Android only. Returns: Union['WebDriver', 'Location']: Self instance """ - self.execute(Command.TOGGLE_LOCATION_SERVICES, {}) - return self + try: + self.execute_script('mobile: toggleGps') + except UnknownMethodException: + # TODO: Remove the fallback + self.execute(Command.TOGGLE_LOCATION_SERVICES) + return cast('WebDriver', self) def set_location( - self: T, + self, latitude: Union[float, str], longitude: Union[float, str], altitude: Union[float, str, None] = None, speed: Union[float, str, None] = None, satellites: Union[float, str, None] = None, - ) -> T: + ) -> 'WebDriver': """Set the location of the device Args: @@ -66,7 +75,7 @@ def set_location( if satellites is not None: data['location']['satellites'] = satellites self.execute(Command.SET_LOCATION, data) - return self + return cast('WebDriver', self) @property def location(self) -> Dict[str, float]: diff --git a/appium/webdriver/extensions/log_event.py b/appium/webdriver/extensions/log_event.py index f24a8d30..0ff1d1f6 100644 --- a/appium/webdriver/extensions/log_event.py +++ b/appium/webdriver/extensions/log_event.py @@ -12,13 +12,14 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Dict, List, TypeVar, Union +from typing import TYPE_CHECKING, Dict, List, Union, cast from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands from ..mobilecommand import MobileCommand as Command -T = TypeVar('T', bound=CanExecuteCommands) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver class LogEvent(CanExecuteCommands): @@ -45,7 +46,7 @@ def get_events(self, type: Union[List[str], None] = None) -> Dict[str, Union[str data['type'] = type return self.execute(Command.GET_EVENTS, data)['value'] - def log_event(self: T, vendor: str, event: str) -> T: + def log_event(self, vendor: str, event: str) -> 'WebDriver': """Log a custom event on the Appium server. (Since Appium 1.16.0) @@ -61,7 +62,7 @@ def log_event(self: T, vendor: str, event: str) -> T: """ data = {'vendor': vendor, 'event': event} self.execute(Command.LOG_EVENT, data) - return self + return cast('WebDriver', self) def _add_commands(self) -> None: # noinspection PyProtectedMember,PyUnresolvedReferences diff --git a/appium/webdriver/extensions/remote_fs.py b/appium/webdriver/extensions/remote_fs.py index 9f65576d..a7beeb22 100644 --- a/appium/webdriver/extensions/remote_fs.py +++ b/appium/webdriver/extensions/remote_fs.py @@ -13,18 +13,21 @@ # limitations under the License. import base64 -from typing import Optional, TypeVar +from typing import TYPE_CHECKING, Optional, cast -from selenium.common.exceptions import InvalidArgumentException +from selenium.common.exceptions import InvalidArgumentException, UnknownMethodException from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands +from appium.protocols.webdriver.can_execute_scripts import CanExecuteScripts +from appium.protocols.webdriver.can_remember_extension_presence import CanRememberExtensionPresence from ..mobilecommand import MobileCommand as Command -T = TypeVar('T', bound=CanExecuteCommands) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver -class RemoteFS(CanExecuteCommands): +class RemoteFS(CanExecuteCommands, CanExecuteScripts, CanRememberExtensionPresence): def pull_file(self, path: str) -> str: """Retrieves the file at `path`. @@ -34,10 +37,12 @@ def pull_file(self, path: str) -> str: Returns: The file's contents encoded as Base64. """ - data = { - 'path': path, - } - return self.execute(Command.PULL_FILE, data)['value'] + ext_name = 'mobile: pullFile' + try: + return self.assert_extension_exists(ext_name).execute_script(ext_name, {'remotePath': path}) + except UnknownMethodException: + # TODO: Remove the fallback + return self.mark_extension_absence(ext_name).execute(Command.PULL_FILE, {'path': path})['value'] def pull_folder(self, path: str) -> str: """Retrieves a folder at `path`. @@ -48,14 +53,16 @@ def pull_folder(self, path: str) -> str: Returns: The folder's contents zipped and encoded as Base64. """ - data = { - 'path': path, - } - return self.execute(Command.PULL_FOLDER, data)['value'] + ext_name = 'mobile: pullFolder' + try: + return self.assert_extension_exists(ext_name).execute_script(ext_name, {'remotePath': path}) + except UnknownMethodException: + # TODO: Remove the fallback + return self.mark_extension_absence(ext_name).execute(Command.PULL_FOLDER, {'path': path})['value'] def push_file( - self: T, destination_path: str, base64data: Optional[str] = None, source_path: Optional[str] = None - ) -> T: + self, destination_path: str, base64data: Optional[str] = None, source_path: Optional[str] = None + ) -> 'WebDriver': """Puts the data from the file at `source_path`, encoded as Base64, in the file specified as `path`. Specify either `base64data` or `source_path`, if both specified default to `source_path` @@ -81,12 +88,25 @@ def push_file( raise InvalidArgumentException(message) from e base64data = base64.b64encode(file_data).decode('utf-8') - data = { - 'path': destination_path, - 'data': base64data, - } - self.execute(Command.PUSH_FILE, data) - return self + ext_name = 'mobile: pushFile' + try: + self.assert_extension_exists(ext_name).execute_script( + ext_name, + { + 'remotePath': destination_path, + 'payload': base64data, + }, + ) + except UnknownMethodException: + # TODO: Remove the fallback + self.mark_extension_absence(ext_name).execute( + Command.PUSH_FILE, + { + 'path': destination_path, + 'data': base64data, + }, + ) + return cast('WebDriver', self) def _add_commands(self) -> None: # noinspection PyProtectedMember,PyUnresolvedReferences diff --git a/appium/webdriver/extensions/session.py b/appium/webdriver/extensions/session.py index e632f47a..dfd387ca 100644 --- a/appium/webdriver/extensions/session.py +++ b/appium/webdriver/extensions/session.py @@ -12,6 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. +import warnings from typing import Any, Dict, List from appium.common.logger import logger @@ -24,6 +25,7 @@ class Session(CanExecuteCommands): @property def session(self) -> Dict[str, Any]: """Retrieves session information from the current session + deprecated:: 2.0.0 Usage: session = driver.session @@ -31,11 +33,17 @@ def session(self) -> Dict[str, Any]: Returns: `dict`: containing information from the current session """ + warnings.warn( + 'The "session" API is deprecated and will be removed in future versions', + DeprecationWarning, + ) + return self.execute(Command.GET_SESSION)['value'] @property def all_sessions(self) -> List[Dict[str, Any]]: """Retrieves all sessions that are open + deprecated:: 2.0.0 Usage: sessions = driver.all_sessions @@ -43,6 +51,11 @@ def all_sessions(self) -> List[Dict[str, Any]]: Returns: :obj:`list` of :obj:`dict`: containing all open sessions """ + warnings.warn( + 'The "all_sessions" API is deprecated and will be removed in future versions', + DeprecationWarning, + ) + return self.execute(Command.GET_ALL_SESSIONS)['value'] @property diff --git a/appium/webdriver/extensions/settings.py b/appium/webdriver/extensions/settings.py index 3c3026f7..c45e819c 100644 --- a/appium/webdriver/extensions/settings.py +++ b/appium/webdriver/extensions/settings.py @@ -12,13 +12,14 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Any, Dict, TypeVar +from typing import TYPE_CHECKING, Any, Dict, cast from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands from ..mobilecommand import MobileCommand as Command -T = TypeVar('T', bound=CanExecuteCommands) +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver class Settings(CanExecuteCommands): @@ -33,7 +34,7 @@ def get_settings(self) -> Dict[str, Any]: """ return self.execute(Command.GET_SETTINGS, {})['value'] - def update_settings(self: T, settings: Dict[str, Any]) -> T: + def update_settings(self, settings: Dict[str, Any]) -> 'WebDriver': """Set settings for the current session. For more on settings, see: https://github.com/appium/appium/blob/master/docs/en/advanced-concepts/settings.md @@ -41,10 +42,8 @@ def update_settings(self: T, settings: Dict[str, Any]) -> T: Args: settings: dictionary of settings to apply to the current test session """ - data = {"settings": settings} - - self.execute(Command.UPDATE_SETTINGS, data) - return self # type: ignore + self.execute(Command.UPDATE_SETTINGS, {"settings": settings}) + return cast('WebDriver', self) def _add_commands(self) -> None: # noinspection PyProtectedMember,PyUnresolvedReferences diff --git a/appium/webdriver/switch_to.py b/appium/webdriver/switch_to.py index e6d46ec8..21e9393e 100644 --- a/appium/webdriver/switch_to.py +++ b/appium/webdriver/switch_to.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -from typing import Optional, TypeVar +from typing import TYPE_CHECKING, Optional, cast from selenium.webdriver.remote.switch_to import SwitchTo @@ -21,16 +21,16 @@ from .mobilecommand import MobileCommand +if TYPE_CHECKING: + from appium.webdriver.webdriver import WebDriver + class HasDriver(Protocol): _driver: CanExecuteCommands -T = TypeVar('T', bound=HasDriver) - - class MobileSwitchTo(SwitchTo, HasDriver): - def context(self: T, context_name: Optional[str]) -> T: + def context(self, context_name: Optional[str]) -> 'WebDriver': """Sets the context for the current session. Passing `None` is equal to switching to native context. @@ -41,4 +41,4 @@ def context(self: T, context_name: Optional[str]) -> T: driver.switch_to.context('WEBVIEW_1') """ self._driver.execute(MobileCommand.SWITCH_TO_CONTEXT, {'name': context_name}) - return self + return cast('WebDriver', self) diff --git a/appium/webdriver/webdriver.py b/appium/webdriver/webdriver.py index 49f857fd..2af8220a 100644 --- a/appium/webdriver/webdriver.py +++ b/appium/webdriver/webdriver.py @@ -15,17 +15,21 @@ # pylint: disable=too-many-lines,too-many-public-methods,too-many-statements,no-self-use import warnings -from typing import Any, Callable, Dict, List, Optional, Tuple, TypeVar, Union +from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union from selenium import webdriver -from selenium.common.exceptions import InvalidArgumentException, SessionNotCreatedException, WebDriverException +from selenium.common.exceptions import ( + InvalidArgumentException, + SessionNotCreatedException, + UnknownMethodException, + WebDriverException, +) from selenium.webdriver.common.by import By from selenium.webdriver.remote.command import Command as RemoteCommand from selenium.webdriver.remote.remote_connection import RemoteConnection from appium.common.logger import logger from appium.options.common.base import AppiumOptions -from appium.protocols.webdriver.can_execute_commands import CanExecuteCommands from appium.webdriver.common.appiumby import AppiumBy from .appium_connection import AppiumConnection @@ -60,8 +64,6 @@ from .switch_to import MobileSwitchTo from .webelement import WebElement as MobileWebElement -T = TypeVar('T', bound=CanExecuteCommands) - class ExtensionBase: """ @@ -214,7 +216,6 @@ def __init__( strict_ssl: bool = True, options: Union[AppiumOptions, List[AppiumOptions], None] = None, ): - if strict_ssl is False: # pylint: disable=E1101 # noinspection PyPackageRequirements @@ -258,6 +259,8 @@ def __init__( By.IMAGE = AppiumBy.IMAGE By.CUSTOM = AppiumBy.CUSTOM + self._absent_extensions: Set[str] = set() + self._extensions = extensions or [] for extension in self._extensions: instance = extension(self.execute) @@ -436,7 +439,7 @@ def create_web_element(self, element_id: Union[int, str]) -> MobileWebElement: """ return MobileWebElement(self, element_id) - def set_value(self: T, element: MobileWebElement, value: str) -> T: + def set_value(self, element: MobileWebElement, value: str) -> 'WebDriver': """Set the value on an element in the application. deprecated:: 2.8.1 @@ -497,9 +500,34 @@ def orientation(self, value: str) -> None: if value.upper() in allowed_values: self.execute(Command.SET_SCREEN_ORIENTATION, {'orientation': value}) else: - raise WebDriverException("You can only set the orientation to 'LANDSCAPE' and 'PORTRAIT'") + def assert_extension_exists(self, ext_name: str) -> 'WebDriver': + """ + Verifies if the given extension is not present in the list of absent extensions + for the given driver instance. + This API is designed for private usage. + + :param ext_name: extension name + :return: self instance for chaining + :raise UnknownMethodException: If the extension has been marked as absent once + """ + if ext_name in self._absent_extensions: + raise UnknownMethodException() + return self + + def mark_extension_absence(self, ext_name: str) -> 'WebDriver': + """ + Marks the given extension as absent for the given driver instance. + This API is designed for private usage. + + :param ext_name: extension name + :return: self instance for chaining + """ + logger.debug(f'Marking driver extension "{ext_name}" as absent for the current instance') + self._absent_extensions.add(ext_name) + return self + def _add_commands(self) -> None: # call the overridden command binders from all mixin classes except for # appium.webdriver.webdriver.WebDriver and its sub-classes diff --git a/ci-jobs/functional/setup_appium.yml b/ci-jobs/functional/setup_appium.yml index 05931e40..69ac8627 100644 --- a/ci-jobs/functional/setup_appium.yml +++ b/ci-jobs/functional/setup_appium.yml @@ -12,7 +12,7 @@ steps: displayName: Resolve dependencies (Appium server) - script: | pip install --upgrade pip - pip install pipenv + pip install --upgrade pipenv pipenv lock --clear pipenv install --system displayName: Resolve dependencies (Python) diff --git a/test/functional/ios/webdriver_tests.py b/test/functional/ios/webdriver_tests.py index 597c42f0..cba16233 100644 --- a/test/functional/ios/webdriver_tests.py +++ b/test/functional/ios/webdriver_tests.py @@ -33,7 +33,6 @@ class TestWebDriver(BaseTestCase): - # TODO Due to not created 2nd session somehow @pytest.mark.skipif(condition=is_ci(), reason='Need to fix flaky test during running on CI.') def test_all_sessions(self) -> None: diff --git a/test/unit/webdriver/app_test.py b/test/unit/webdriver/app_test.py index 02bbb8df..1a8b0247 100644 --- a/test/unit/webdriver/app_test.py +++ b/test/unit/webdriver/app_test.py @@ -37,6 +37,7 @@ def test_install_app(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/install_app'), body='{"value": ""}' ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": ""}') result = driver.install_app('path/to/app') assert {'app': 'path/to/app'}, get_httpretty_request_body(httpretty.last_request()) @@ -48,6 +49,7 @@ def test_remove_app(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/remove_app'), body='{"value": ""}' ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": ""}') result = driver.remove_app('com.app.id') assert {'app': 'com.app.id'}, get_httpretty_request_body(httpretty.last_request()) @@ -59,6 +61,9 @@ def test_app_installed(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/app_installed'), body='{"value": true}' ) + httpretty.register_uri( + httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": true}' + ) result = driver.is_app_installed("com.app.id") assert {'app': "com.app.id"}, get_httpretty_request_body(httpretty.last_request()) assert result is True @@ -69,6 +74,9 @@ def test_terminate_app(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/terminate_app'), body='{"value": true}' ) + httpretty.register_uri( + httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": true}' + ) result = driver.terminate_app("com.app.id") assert {'app': "com.app.id"}, get_httpretty_request_body(httpretty.last_request()) assert result is True @@ -79,6 +87,7 @@ def test_activate_app(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/activate_app'), body='{"value": ""}' ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": ""}') result = driver.activate_app("com.app.id") assert {'app': 'com.app.id'}, get_httpretty_request_body(httpretty.last_request()) @@ -90,6 +99,7 @@ def test_background_app(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/app/background'), body='{"value": ""}' ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": ""}') result = driver.background_app(0) assert {'app': 0}, get_httpretty_request_body(httpretty.last_request()) assert isinstance(result, WebDriver) @@ -116,6 +126,7 @@ def test_query_app_state(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/app_state'), body='{"value": 3 }' ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": 3}') result = driver.query_app_state('com.app.id') assert {'app': 3}, get_httpretty_request_body(httpretty.last_request()) diff --git a/test/unit/webdriver/device/device_time_test.py b/test/unit/webdriver/device/device_time_test.py index 59b5dcb9..8c6f5e0b 100644 --- a/test/unit/webdriver/device/device_time_test.py +++ b/test/unit/webdriver/device/device_time_test.py @@ -26,6 +26,11 @@ def test_device_time(self): appium_command('/session/1234567890/appium/device/system_time'), body='{"value": "2019-01-05T14:46:44+09:00"}', ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + body='{"value": "2019-01-05T14:46:44+09:00"}', + ) assert driver.device_time == '2019-01-05T14:46:44+09:00' @httpretty.activate @@ -36,6 +41,11 @@ def test_get_device_time(self): appium_command('/session/1234567890/appium/device/system_time'), body='{"value": "2019-01-05T14:46:44+09:00"}', ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + body='{"value": "2019-01-05T14:46:44+09:00"}', + ) assert driver.get_device_time() == '2019-01-05T14:46:44+09:00' @httpretty.activate @@ -46,7 +56,12 @@ def test_get_formatted_device_time(self): appium_command('/session/1234567890/appium/device/system_time'), body='{"value": "2019-01-08"}', ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + body='{"value": "2019-01-08"}', + ) assert driver.get_device_time('YYYY-MM-DD') == '2019-01-08' d = get_httpretty_request_body(httpretty.last_request()) - assert d['format'] == 'YYYY-MM-DD' + assert d.get('format', d['args'][0]['format']) == 'YYYY-MM-DD' diff --git a/test/unit/webdriver/device/fingerprint_test.py b/test/unit/webdriver/device/fingerprint_test.py index b4d3b6e7..e6facf86 100644 --- a/test/unit/webdriver/device/fingerprint_test.py +++ b/test/unit/webdriver/device/fingerprint_test.py @@ -14,6 +14,7 @@ import httpretty +from appium.webdriver.webdriver import WebDriver from test.unit.helper.test_helper import android_w3c_driver, appium_command, get_httpretty_request_body @@ -26,8 +27,13 @@ def test_finger_print(self): appium_command('/session/1234567890/appium/device/finger_print'), # body is None ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + # body is None + ) - assert driver.finger_print(1) is None + assert isinstance(driver.finger_print(1), WebDriver) d = get_httpretty_request_body(httpretty.last_request()) - assert d['fingerprintId'] == 1 + assert d.get('fingerprintId', d['args'][0]['fingerprintId']) == 1 diff --git a/test/unit/webdriver/device/keyboard_test.py b/test/unit/webdriver/device/keyboard_test.py index 91de8cd8..e3401839 100644 --- a/test/unit/webdriver/device/keyboard_test.py +++ b/test/unit/webdriver/device/keyboard_test.py @@ -23,6 +23,7 @@ class TestWebDriverKeyboard(object): def test_hide_keyboard(self): driver = android_w3c_driver() httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/appium/device/hide_keyboard')) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync')) assert isinstance(driver.hide_keyboard(), WebDriver) @httpretty.activate @@ -31,9 +32,12 @@ def test_press_keycode(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/press_keycode'), body='{"value": "86"}' ) + httpretty.register_uri( + httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": "86"}' + ) driver.press_keycode(86) d = get_httpretty_request_body((httpretty.last_request())) - assert d['keycode'] == 86 + assert d.get('keycode', d['args'][0]['keycode']) == 86 @httpretty.activate def test_long_press_keycode(self): @@ -43,9 +47,12 @@ def test_long_press_keycode(self): appium_command('/session/1234567890/appium/device/long_press_keycode'), body='{"value": "86"}', ) + httpretty.register_uri( + httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": "86"}' + ) driver.long_press_keycode(86) d = get_httpretty_request_body((httpretty.last_request())) - assert d['keycode'] == 86 + assert d.get('keycode', d['args'][0]['keycode']) == 86 @httpretty.activate def test_keyevent(self): @@ -53,6 +60,9 @@ def test_keyevent(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/keyevent'), body='{keycode: 86}' ) + httpretty.register_uri( + httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": "86"}' + ) assert isinstance(driver.keyevent(86), WebDriver) @httpretty.activate @@ -63,10 +73,11 @@ def test_press_keycode_with_flags(self): appium_command('/session/1234567890/appium/device/press_keycode'), body='{keycode: 86, metastate: 2097153, flags: 44}', ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync')) # metastate is META_SHIFT_ON and META_NUM_LOCK_ON # flags is CANCELFLAG_CANCELEDED, FLAG_KEEP_TOUCH_MODE, FLAG_FROM_SYSTEM assert isinstance( - driver.press_keycode(86, metastate=[0x00000001, 0x00200000], flags=[0x20, 0x00000004, 0x00000008]), + driver.press_keycode(86, metastate=0x00000001 | 0x00200000, flags=0x20 | 0x00000004 | 0x00000008), WebDriver, ) @@ -78,9 +89,10 @@ def test_long_press_keycode_with_flags(self): appium_command('/session/1234567890/appium/device/long_press_keycode'), body='{keycode: 86, metastate: 2097153, flags: 44}', ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync')) # metastate is META_SHIFT_ON and META_NUM_LOCK_ON # flags is CANCELFLAG_CANCELEDED, FLAG_KEEP_TOUCH_MODE, FLAG_FROM_SYSTEM assert isinstance( - driver.long_press_keycode(86, metastate=[0x00000001, 0x00200000], flags=[0x20, 0x00000004, 0x00000008]), + driver.long_press_keycode(86, metastate=0x00000001 | 0x00200000, flags=0x20 | 0x00000004 | 0x00000008), WebDriver, ) diff --git a/test/unit/webdriver/device/location_test.py b/test/unit/webdriver/device/location_test.py index c351c2f4..512bdb95 100644 --- a/test/unit/webdriver/device/location_test.py +++ b/test/unit/webdriver/device/location_test.py @@ -27,6 +27,7 @@ def test_toggle_location_services(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/toggle_location_services') ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync')) assert isinstance(driver.toggle_location_services(), WebDriver) @httpretty.activate diff --git a/test/unit/webdriver/device/lock_test.py b/test/unit/webdriver/device/lock_test.py index cecc5683..1e8ce09c 100644 --- a/test/unit/webdriver/device/lock_test.py +++ b/test/unit/webdriver/device/lock_test.py @@ -25,10 +25,11 @@ def test_lock(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/lock'), body='{"value": ""}' ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": ""}') driver.lock(1) d = get_httpretty_request_body(httpretty.last_request()) - assert d['seconds'] == 1 + assert d.get('seconds', d['args'][0]['seconds']) == 1 @httpretty.activate def test_lock_no_args(self): @@ -36,17 +37,18 @@ def test_lock_no_args(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/lock'), body='{"value": ""}' ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": ""}') driver.lock() - d = get_httpretty_request_body(httpretty.last_request()) - assert d == {} - @httpretty.activate def test_islocked_false(self): driver = android_w3c_driver() httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/is_locked'), body='{"value": false}' ) + httpretty.register_uri( + httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": false}' + ) assert driver.is_locked() is False @httpretty.activate @@ -55,7 +57,9 @@ def test_islocked_true(self): httpretty.register_uri( httpretty.POST, appium_command('/session/1234567890/appium/device/is_locked'), body='{"value": true}' ) - + httpretty.register_uri( + httpretty.POST, appium_command('/session/1234567890/execute/sync'), body='{"value": true}' + ) assert driver.is_locked() is True @httpretty.activate @@ -65,4 +69,5 @@ def test_unlock(self): httpretty.POST, appium_command('/session/1234567890/appium/device/unlock'), ) + httpretty.register_uri(httpretty.POST, appium_command('/session/1234567890/execute/sync')) assert isinstance(driver.unlock(), WebDriver) diff --git a/test/unit/webdriver/device/remote_fs_test.py b/test/unit/webdriver/device/remote_fs_test.py index cab63f9f..4de3a85f 100644 --- a/test/unit/webdriver/device/remote_fs_test.py +++ b/test/unit/webdriver/device/remote_fs_test.py @@ -30,14 +30,18 @@ def test_push_file(self): httpretty.POST, appium_command('/session/1234567890/appium/device/push_file'), ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + ) dest_path = '/path/to/file.txt' data = base64.b64encode(bytes('HelloWorld', 'utf-8')).decode('utf-8') assert isinstance(driver.push_file(dest_path, data), WebDriver) d = get_httpretty_request_body(httpretty.last_request()) - assert d['path'] == dest_path - assert d['data'] == str(data) + assert d.get('path', d['args'][0]['remotePath']) == dest_path + assert d.get('data', d['args'][0]['payload']) == str(data) @httpretty.activate def test_push_file_invalid_arg_exception_without_src_path_and_base64data(self): @@ -46,6 +50,10 @@ def test_push_file_invalid_arg_exception_without_src_path_and_base64data(self): httpretty.POST, appium_command('/session/1234567890/appium/device/push_file'), ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + ) dest_path = '/path/to/file.txt' with pytest.raises(InvalidArgumentException): @@ -58,6 +66,10 @@ def test_push_file_invalid_arg_exception_with_src_file_not_found(self): httpretty.POST, appium_command('/session/1234567890/appium/device/push_file'), ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/appium/device/push_file'), + ) dest_path = '/dest_path/to/file.txt' src_path = '/src_path/to/file.txt' @@ -72,12 +84,17 @@ def test_pull_file(self): appium_command('/session/1234567890/appium/device/pull_file'), body='{"value": "SGVsbG9Xb3JsZA=="}', ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + body='{"value": "SGVsbG9Xb3JsZA=="}', + ) dest_path = '/path/to/file.txt' assert driver.pull_file(dest_path) == str(base64.b64encode(bytes('HelloWorld', 'utf-8')).decode('utf-8')) d = get_httpretty_request_body(httpretty.last_request()) - assert d['path'] == dest_path + assert d.get('path', d['args'][0]['remotePath']) == dest_path @httpretty.activate def test_pull_folder(self): @@ -87,9 +104,14 @@ def test_pull_folder(self): appium_command('/session/1234567890/appium/device/pull_folder'), body='{"value": "base64EncodedZippedFolderData"}', ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + body='{"value": "base64EncodedZippedFolderData"}', + ) dest_path = '/path/to/file.txt' assert driver.pull_folder(dest_path) == 'base64EncodedZippedFolderData' d = get_httpretty_request_body(httpretty.last_request()) - assert d['path'] == dest_path + assert d.get('path', d['args'][0]['remotePath']) == dest_path diff --git a/test/unit/webdriver/device/shake_test.py b/test/unit/webdriver/device/shake_test.py index d60e70b5..0371852f 100644 --- a/test/unit/webdriver/device/shake_test.py +++ b/test/unit/webdriver/device/shake_test.py @@ -27,4 +27,8 @@ def test_shake(self): httpretty.POST, appium_command('/session/1234567890/appium/device/shake'), ) + httpretty.register_uri( + httpretty.POST, + appium_command('/session/1234567890/execute/sync'), + ) assert isinstance(driver.shake(), WebDriver)