Add explicit type declarations

Pipfile

@@ -24,3 +24,5 @@ mock = "~=3.0"
 pylint = "~=2.4"
 astroid = "~=2.3"
 isort = "~=4.3"
+
+mypy = "==0.761"

README.md

@@ -45,14 +45,18 @@ download and unarchive the source tarball (Appium-Python-Client-X.X.tar.gz).
 
 - Style Guide: https://www.python.org/dev/peps/pep-0008/
     - `autopep8` helps to format code automatically
-        ```
+        ```shell
         $ python -m autopep8 -r --global-config .config-pep8 -i .
         ```
     - `isort` helps to order imports automatically
-        ```
+        ```shell
         $ python -m isort -rc .
         ```
         - When you use newly 3rd party modules, add it to [.isort.cfg](.isort.cfg) to keep import order correct
+    - `mypy` helps to check explicit type declarations
+        ```shell
+        $ python -m mypy appium
+        ```
 - Docstring style: Google Style
     - Refer [link](https://sphinxcontrib-napoleon.readthedocs.io/en/latest/example_google.html)
 - You can customise `CHANGELOG.rst` with commit messages following [.gitchangelog.rc](.gitchangelog.rc)

appium/common/helper.py

@@ -12,44 +12,25 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from collections import OrderedDict
+from typing import Dict
 
 from appium import version as appium_version
 
 
-def appium_bytes(value, encoding):
-    """Return a bytes-like object
-
-    Has _appium_ prefix to avoid overriding built-in bytes.
-
-    Args:
-        value (str): A value to convert
-        encoding (str): A encoding which will convert to
-
-    Returns:
-        str: A bytes-like object
-    """
-
-    try:
-        return bytes(value, encoding)  # Python 3
-    except TypeError:
-        return value  # Python 2
-
-
-def extract_const_attributes(cls):
+def extract_const_attributes(cls: type) -> Dict:
     """Return dict with constants attributes and values in the class(e.g. {'VAL1': 1, 'VAL2': 2})
 
     Args:
         cls (type): Class to be extracted constants
 
     Returns:
-        OrderedDict: dict with constants attributes and values in the class
+        dict: dict with constants attributes and values in the class
     """
-    return OrderedDict(
-        [(attr, value) for attr, value in vars(cls).items() if not callable(getattr(cls, attr)) and attr.isupper()])
+    return dict([(attr, value) for attr, value in vars(cls).items()
+                 if not callable(getattr(cls, attr)) and attr.isupper()])
 
 
-def library_version():
+def library_version() -> str:
     """Return a version of this python library
     """
 

appium/common/logger.py

@@ -16,7 +16,7 @@
 import sys
 
 
-def setup_logger(level=logging.NOTSET):
+def setup_logger(level: int = logging.NOTSET) -> None:
     logger.propagate = False
     logger.setLevel(level)
     handler = logging.StreamHandler(stream=sys.stderr)

appium/saucetestcase.py

@@ -19,6 +19,7 @@
 import os
 import sys
 import unittest
+from typing import Any, Callable, List
 
 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[str]) -> Callable[[type], None]:
+    def decorator(base_class: type) -> None:
         module = sys.modules[base_class.__module__].__dict__
         for i, platform in enumerate(platforms):
             name = "%s_%s" % (base_class.__name__, i + 1)
@@ -40,16 +41,16 @@ def decorator(base_class):
 
 
 class SauceTestCase(unittest.TestCase):
-    def setUp(self):
-        self.desired_capabilities['name'] = self.id()
+    def setUp(self) -> None:
+        self.desired_capabilities['name'] = self.id()  # type: ignore
         sauce_url = "http://%s:%s@ondemand.saucelabs.com:80/wd/hub"
         self.driver = webdriver.Remote(
-            desired_capabilities=self.desired_capabilities,
+            desired_capabilities=self.desired_capabilities,  # type: ignore
             command_executor=sauce_url % (SAUCE_USERNAME, SAUCE_ACCESS_KEY)
         )
         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 Any, 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[str, Any]:
         """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 Any, List, Optional, TypeVar, Union
 
 import urllib3
 
@@ -27,7 +27,7 @@
 STATUS_URL = '/wd/hub/status'
 
 
-def find_executable(executable):
+def find_executable(executable: str) -> Optional[str]:
     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: int, path: str, timeout_ms: int) -> bool:
     time_started_sec = time.time()
     while time.time() < time_started_sec + timeout_ms / 1000.0:
         try:
@@ -64,62 +64,65 @@ class AppiumServiceError(RuntimeError):
     pass
 
 
+T = TypeVar('T', bound='AppiumService')
+
+
 class AppiumService(object):
-    def __init__(self):
-        self._process = None
-        self._cmd = None
+    def __init__(self) -> None:
+        self._process: Optional[sp.Popen] = None
+        self._cmd: Optional[List] = None
 
-    def _get_node(self):
+    def _get_node(self) -> str:
         if not hasattr(self, '_node_executable'):
             self._node_executable = find_executable('node')
         if self._node_executable is None:
             raise AppiumServiceError('NodeJS main executable cannot be found. ' +
                                      'Make sure it is installed and present in PATH')
         return self._node_executable
 
-    def _get_npm(self):
+    def _get_npm(self) -> str:
         if not hasattr(self, '_npm_executable'):
             self._npm_executable = find_executable('npm.cmd' if sys.platform == 'win32' else 'npm')
         if self._npm_executable is None:
             raise AppiumServiceError('Node Package Manager executable cannot be found. ' +
                                      'Make sure it is installed and present in PATH')
         return self._npm_executable
 
-    def _get_main_script(self):
+    def _get_main_script(self) -> Union[str, bytes]:
         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)
+                        self._main_script: Union[str, bytes] = 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
 
     @staticmethod
-    def _parse_port(args):
+    def _parse_port(args: List[str]) -> int:
         for idx, arg in enumerate(args or []):
             if arg in ('--port', '-p') and idx < len(args) - 1:
                 return int(args[idx + 1])
         return DEFAULT_PORT
 
     @staticmethod
-    def _parse_host(args):
+    def _parse_host(args: List[str]) -> str:
         for idx, arg in enumerate(args or []):
             if arg in ('--address', '-a') and idx < len(args) - 1:
                 return args[idx + 1]
         return DEFAULT_HOST
 
-    def start(self, **kwargs):
+    def start(self, **kwargs: Any) -> sp.Popen:
         """Starts Appium service with given arguments.
 
         The service will be forcefully restarted if it is already running.
@@ -153,31 +156,31 @@ 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
+        error_msg: Optional[str] = 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)
+                    error_msg += '\nOriginal error: {}'.format(str(err_output))
             self.stop()
             raise AppiumServiceError(error_msg)
         return self._process
 
-    def stop(self):
+    def stop(self) -> bool:
         """Stops Appium service if it is running.
 
         The call will be ignored if the service is not running
@@ -188,14 +191,14 @@ def stop(self):
         """
         is_terminated = False
         if self.is_running:
-            self._process.terminate()
+            self._process.terminate()  # type: ignore
             is_terminated = True
         self._process = None
         self._cmd = None
         return is_terminated
 
     @property
-    def is_running(self):
+    def is_running(self) -> bool:
         """Check if the service is running.
 
         Returns:
@@ -204,7 +207,7 @@ def is_running(self):
         return self._process is not None and self._process.poll() is None
 
     @property
-    def is_listening(self):
+    def is_listening(self) -> bool:
         """Check if the service is listening on the given/default host/port.
 
         The fact, that the service is running, does not always mean it is listening.

appium/webdriver/common/multi_action.py

@@ -19,17 +19,25 @@
 # chaining as the spec requires.
 
 import copy
+from typing import TYPE_CHECKING, Dict, List, Optional, TypeVar, Union
 
 from appium.webdriver.mobilecommand import MobileCommand as Command
 
+if TYPE_CHECKING:
+    from appium.webdriver.webdriver import WebDriver
+    from appium.webdriver.webelement import WebElement
+    from appium.webdriver.common.touch_action import TouchAction
+
+T = TypeVar('T', bound='MultiAction')
+
 
 class MultiAction(object):
-    def __init__(self, driver, element=None):
+    def __init__(self, driver: 'WebDriver', element: Optional['WebElement'] = None) -> None:
         self._driver = driver
         self._element = element
-        self._touch_actions = []
+        self._touch_actions: List['TouchAction'] = []
 
-    def add(self, *touch_actions):
+    def add(self, *touch_actions: 'TouchAction') -> 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[str, Union[List, str]]:
         actions = []
         for action in self._touch_actions:
             actions.append(action.json_wire_gestures)