Add explicit type declarations

appium/common/helper.py

@@ -17,7 +17,7 @@
 from appium import version as appium_version
 
 
-def appium_bytes(value, encoding):
+def appium_bytes(value: str, encoding: str) -> str:
     """Return a bytes-like object
 
     Has _appium_ prefix to avoid overriding built-in bytes.
@@ -36,7 +36,7 @@ def appium_bytes(value, encoding):
         return value  # Python 2
 
 
-def extract_const_attributes(cls):
+def extract_const_attributes(cls) -> OrderedDict:
     """Return dict with constants attributes and values in the class(e.g. {'VAL1': 1, 'VAL2': 2})
 
     Args:

appium/saucetestcase.py

@@ -19,6 +19,7 @@
 import os
 import sys
 import unittest
+from typing import Any, List, TypeVar
 
 from sauceclient import SauceClient
 
@@ -29,8 +30,8 @@
 sauce = SauceClient(SAUCE_USERNAME, SAUCE_ACCESS_KEY)
 
 
-def on_platforms(platforms):
-    def decorator(base_class):
+def on_platforms(platforms: List) -> Any:
+    def decorator(base_class: Any) -> None:
         module = sys.modules[base_class.__module__].__dict__
         for i, platform in enumerate(platforms):
             name = "%s_%s" % (base_class.__name__, i + 1)
@@ -40,7 +41,7 @@ def decorator(base_class):
 
 
 class SauceTestCase(unittest.TestCase):
-    def setUp(self):
+    def setUp(self) -> None:
         self.desired_capabilities['name'] = self.id()
         sauce_url = "http://%s:%[email protected]:80/wd/hub"
         self.driver = webdriver.Remote(
@@ -49,7 +50,7 @@ def setUp(self):
         )
         self.driver.implicitly_wait(30)
 
-    def tearDown(self):
+    def tearDown(self) -> None:
         print("Link to your job: https://saucelabs.com/jobs/%s" % self.driver.session_id)
         try:
             if sys.exc_info() == (None, None, None):

appium/webdriver/appium_connection.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import Dict
+
 from selenium.webdriver.remote.remote_connection import RemoteConnection
 
 from appium.common.helper import library_version
@@ -20,7 +22,7 @@
 class AppiumConnection(RemoteConnection):
 
     @classmethod
-    def get_remote_connection_headers(cls, parsed_url, keep_alive=True):
+    def get_remote_connection_headers(cls, parsed_url: str, keep_alive: bool = True) -> Dict:
         """Override get_remote_connection_headers in RemoteConnection"""
         headers = RemoteConnection.get_remote_connection_headers(parsed_url, keep_alive=keep_alive)
         headers['User-Agent'] = 'appium/python {} ({})'.format(library_version(), headers['User-Agent'])

appium/webdriver/appium_service.py

@@ -12,11 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
 import os
-import subprocess
+import subprocess as sp
 import sys
 import time
+from typing import Dict, List, Union
 
 import urllib3
 
@@ -27,7 +27,7 @@
 STATUS_URL = '/wd/hub/status'
 
 
-def find_executable(executable):
+def find_executable(executable: str) -> Union[str, None]:
     path = os.environ['PATH']
     paths = path.split(os.pathsep)
     base, ext = os.path.splitext(executable)
@@ -45,7 +45,7 @@ def find_executable(executable):
     return None
 
 
-def poll_url(host, port, path, timeout_ms):
+def poll_url(host: str, port: str, path: str, timeout_ms: int) -> bool:
     time_started_sec = time.time()
     while time.time() < time_started_sec + timeout_ms / 1000.0:
         try:
@@ -89,19 +89,19 @@ def _get_main_script(self):
         if not hasattr(self, '_main_script'):
             for args in [['root', '-g'], ['root']]:
                 try:
-                    modules_root = subprocess.check_output([self._get_npm()] + args).strip().decode('utf-8')
+                    modules_root = sp.check_output([self._get_npm()] + args).strip().decode('utf-8')
                     if os.path.exists(os.path.join(modules_root, MAIN_SCRIPT_PATH)):
                         self._main_script = os.path.join(modules_root, MAIN_SCRIPT_PATH)
                         break
-                except subprocess.CalledProcessError:
+                except sp.CalledProcessError:
                     continue
             if not hasattr(self, '_main_script'):
                 try:
-                    self._main_script = subprocess.check_output(
+                    self._main_script = sp.check_output(
                         [self._get_node(),
                          '-e',
                          'console.log(require.resolve("{}"))'.format(MAIN_SCRIPT_PATH)]).strip()
-                except subprocess.CalledProcessError as e:
+                except sp.CalledProcessError as e:
                     raise AppiumServiceError(e.output)
         return self._main_script
 
@@ -119,7 +119,7 @@ def _parse_host(args):
                 return args[idx + 1]
         return DEFAULT_HOST
 
-    def start(self, **kwargs):
+    def start(self, **kwargs: Union[Dict, str, int]) -> sp.Popen:
         """Starts Appium service with given arguments.
 
         The service will be forcefully restarted if it is already running.
@@ -153,23 +153,23 @@ def start(self, **kwargs):
 
         env = kwargs['env'] if 'env' in kwargs else None
         node = kwargs['node'] if 'node' in kwargs else self._get_node()
-        stdout = kwargs['stdout'] if 'stdout' in kwargs else subprocess.PIPE
-        stderr = kwargs['stderr'] if 'stderr' in kwargs else subprocess.PIPE
+        stdout = kwargs['stdout'] if 'stdout' in kwargs else sp.PIPE
+        stderr = kwargs['stderr'] if 'stderr' in kwargs else sp.PIPE
         timeout_ms = int(kwargs['timeout_ms']) if 'timeout_ms' in kwargs else STARTUP_TIMEOUT_MS
         main_script = kwargs['main_script'] if 'main_script' in kwargs else self._get_main_script()
         args = [node, main_script]
         if 'args' in kwargs:
             args.extend(kwargs['args'])
         self._cmd = args
-        self._process = subprocess.Popen(args=args, stdout=stdout, stderr=stderr, env=env)
+        self._process = sp.Popen(args=args, stdout=stdout, stderr=stderr, env=env)
         host = self._parse_host(args)
         port = self._parse_port(args)
         error_msg = None
         if not self.is_running or (timeout_ms > 0 and not poll_url(host, port, STATUS_URL, timeout_ms)):
             error_msg = 'Appium has failed to start on {}:{} within {}ms timeout'\
                         .format(host, port, timeout_ms)
         if error_msg is not None:
-            if stderr == subprocess.PIPE:
+            if stderr == sp.PIPE:
                 err_output = self._process.stderr.read()
                 if err_output:
                     error_msg += '\nOriginal error: {}'.format(err_output)

appium/webdriver/common/multi_action.py

@@ -19,17 +19,25 @@
 # chaining as the spec requires.
 
 import copy
+from typing import Dict, List, Optional, TypeVar
 
 from appium.webdriver.mobilecommand import MobileCommand as Command
 
+if False:  # For mypy
+    from appium.webdriver.common.touch_action import TouchAction
+    from appium.webdriver.webdriver import WebDriver
+    from appium.webdriver.webelement import WebElement
+
+T = TypeVar('T', bound='MultiAction')
+
 
 class MultiAction(object):
-    def __init__(self, driver, element=None):
+    def __init__(self, driver, element=None) -> None:
         self._driver = driver
         self._element = element
         self._touch_actions = []
 
-    def add(self, *touch_actions):
+    def add(self, *touch_actions) -> None:
         """Add TouchAction objects to the MultiAction, to be performed later.
 
         Args:
@@ -49,7 +57,7 @@ def add(self, *touch_actions):
 
             self._touch_actions.append(copy.copy(touch_action))
 
-    def perform(self):
+    def perform(self: T) -> T:
         """Perform the actions stored in the object.
 
         Usage:
@@ -68,7 +76,7 @@ def perform(self):
         return self
 
     @property
-    def json_wire_gestures(self):
+    def json_wire_gestures(self) -> Dict:
         actions = []
         for action in self._touch_actions:
             actions.append(action.json_wire_gestures)

appium/webdriver/common/touch_action.py

@@ -24,16 +24,23 @@
 # pylint: disable=no-self-use
 
 import copy
+from typing import Dict, List, TypeVar
 
 from appium.webdriver.mobilecommand import MobileCommand as Command
 
+# from appium.webdriver.webelement import WebElement as WebElement
+# from appium.webdriver.webdriver import WebDriver as WebDriver
+
+T = TypeVar('T', bound='TouchAction')
+
 
 class TouchAction(object):
+
     def __init__(self, driver=None):
         self._driver = driver
-        self._actions = []
+        self._actions: List = []
 
-    def tap(self, element=None, x=None, y=None, count=1):
+    def tap(self: T, element=None, x: int = None, y: int = None, count: int = 1) -> T:
         """Perform a tap action on the element
 
         Args:
@@ -50,7 +57,7 @@ def tap(self, element=None, x=None, y=None, count=1):
 
         return self
 
-    def press(self, el=None, x=None, y=None, pressure=None):
+    def press(self: T, el=None, x: int = None, y: int = None, pressure: float = None) -> T:
         """Begin a chain with a press down action at a particular element or point
 
         Args:
@@ -67,7 +74,7 @@ def press(self, el=None, x=None, y=None, pressure=None):
 
         return self
 
-    def long_press(self, el=None, x=None, y=None, duration=1000):
+    def long_press(self: T, el=None, x: int = None, y: int = None, duration: int = 1000) -> T:
         """Begin a chain with a press down that lasts `duration` milliseconds
 
         Args:
@@ -83,7 +90,7 @@ def long_press(self, el=None, x=None, y=None, duration=1000):
 
         return self
 
-    def wait(self, ms=0):
+    def wait(self: T, ms: int = 0) -> T:
         """Pause for `ms` milliseconds.
 
         Args:
@@ -101,7 +108,7 @@ def wait(self, ms=0):
 
         return self
 
-    def move_to(self, el=None, x=None, y=None):
+    def move_to(self: T, el=None, x: int = None, y: int = None) -> T:
         """Move the pointer from the previous point to the element or point specified
 
         Args:
@@ -116,7 +123,7 @@ def move_to(self, el=None, x=None, y=None):
 
         return self
 
-    def release(self):
+    def release(self: T) -> T:
         """End the action by lifting the pointer off the screen
 
         Returns:
@@ -126,7 +133,7 @@ def release(self):
 
         return self
 
-    def perform(self):
+    def perform(self: T) -> T:
         """Perform the action by sending the commands to the server to be operated upon
 
         Returns:
@@ -141,23 +148,24 @@ def perform(self):
         return self
 
     @property
-    def json_wire_gestures(self):
+    def json_wire_gestures(self) -> List[Dict]:
         gestures = []
         for action in self._actions:
             gestures.append(copy.deepcopy(action))
         return gestures
 
-    def _add_action(self, action, options):
+    def _add_action(self, action: str, options: Dict) -> None:
         gesture = {
             'action': action,
             'options': options,
         }
         self._actions.append(gesture)
 
-    def _get_opts(self, element, x, y, duration=None, pressure=None):
+    def _get_opts(self, el=None, x: int = None, y: int = None,
+                  duration: int = None, pressure: float = None) -> Dict:
         opts = {}
-        if element is not None:
-            opts['element'] = element.id
+        if el is not None:
+            opts['element'] = el.id
 
         # it makes no sense to have x but no y, or vice versa.
         if x is not None and y is not None:

appium/webdriver/extensions/action_helpers.py

@@ -12,15 +12,20 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from typing import List, Tuple, TypeVar
+
 from selenium import webdriver
 
 from appium.webdriver.common.multi_action import MultiAction
 from appium.webdriver.common.touch_action import TouchAction
+from appium.webdriver.webelement import WebElement
+
+T = TypeVar('T', bound=webdriver.Remote)
 
 
 class ActionHelpers(webdriver.Remote):
 
-    def scroll(self, origin_el, destination_el, duration=None):
+    def scroll(self: T, origin_el: WebElement, destination_el: WebElement, duration: int = None) -> T:
         """Scrolls from one element to another
 
         Args:
@@ -47,7 +52,7 @@ def scroll(self, origin_el, destination_el, duration=None):
             action.press(origin_el).wait(duration).move_to(destination_el).release().perform()
         return self
 
-    def drag_and_drop(self, origin_el, destination_el):
+    def drag_and_drop(self: T, origin_el: WebElement, destination_el: WebElement) -> T:
         """Drag the origin element to the destination element
 
         Args:
@@ -61,7 +66,7 @@ def drag_and_drop(self, origin_el, destination_el):
         action.long_press(origin_el).move_to(destination_el).release().perform()
         return self
 
-    def tap(self, positions, duration=None):
+    def tap(self: T, positions: List[Tuple], duration: int = None) -> T:
         """Taps on an particular place with up to five fingers, holding for a
         certain time
 
@@ -100,7 +105,7 @@ def tap(self, positions, duration=None):
             ma.perform()
         return self
 
-    def swipe(self, start_x, start_y, end_x, end_y, duration=None):
+    def swipe(self: T, start_x: int, start_y: int, end_x: int, end_y: int, duration: int = 0) -> T:
         """Swipe from one point to another point, for an optional duration.
 
         Args:
@@ -127,7 +132,7 @@ def swipe(self, start_x, start_y, end_x, end_y, duration=None):
         action.perform()
         return self
 
-    def flick(self, start_x, start_y, end_x, end_y):
+    def flick(self: T, start_x: int, start_y: int, end_x: int, end_y: int) -> T:
         """Flick from one point to another point.
 
         Args: