diff --git a/python-packages/adb/__init__.py b/python-packages/adb/__init__.py deleted file mode 100644 index 6b509c643..000000000 --- a/python-packages/adb/__init__.py +++ /dev/null @@ -1,17 +0,0 @@ -# -# Copyright (C) 2015 The Android Open Source Project -# -# 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 __future__ import absolute_import -from .device import * # pylint: disable=wildcard-import diff --git a/python-packages/adb/device.py b/python-packages/adb/adb/__init__.py similarity index 71% rename from python-packages/adb/device.py rename to python-packages/adb/adb/__init__.py index 2a6ebf79a..473c6cae2 100644 --- a/python-packages/adb/device.py +++ b/python-packages/adb/adb/__init__.py @@ -13,12 +13,15 @@ # See the License for the specific language governing permissions and # limitations under the License. # +from __future__ import annotations + import atexit import base64 import logging import os import re import subprocess +from typing import Any, Callable class FindDeviceError(RuntimeError): @@ -26,19 +29,21 @@ class FindDeviceError(RuntimeError): class DeviceNotFoundError(FindDeviceError): - def __init__(self, serial): + def __init__(self, serial: str) -> None: self.serial = serial super(DeviceNotFoundError, self).__init__( 'No device with serial {}'.format(serial)) class NoUniqueDeviceError(FindDeviceError): - def __init__(self): + def __init__(self) -> None: super(NoUniqueDeviceError, self).__init__('No unique device') class ShellError(RuntimeError): - def __init__(self, cmd, stdout, stderr, exit_code): + def __init__( + self, cmd: list[str], stdout: str, stderr: str, exit_code: int + ) -> None: super(ShellError, self).__init__( '`{0}` exited with code {1}'.format(cmd, exit_code)) self.cmd = cmd @@ -47,7 +52,7 @@ class ShellError(RuntimeError): self.exit_code = exit_code -def get_devices(adb_path='adb'): +def get_devices(adb_path: str = 'adb') -> list[str]: with open(os.devnull, 'wb') as devnull: subprocess.check_call([adb_path, 'start-server'], stdout=devnull, stderr=devnull) @@ -68,21 +73,27 @@ def get_devices(adb_path='adb'): return devices -def _get_unique_device(product=None, adb_path='adb'): +def _get_unique_device( + product: str | None = None, adb_path: str = 'adb' +) -> AndroidDevice: devices = get_devices(adb_path=adb_path) if len(devices) != 1: raise NoUniqueDeviceError() return AndroidDevice(devices[0], product, adb_path) -def _get_device_by_serial(serial, product=None, adb_path='adb'): +def _get_device_by_serial( + serial: str, product: str | None = None, adb_path: str = 'adb' +) -> AndroidDevice: for device in get_devices(adb_path=adb_path): if device == serial: return AndroidDevice(serial, product, adb_path) raise DeviceNotFoundError(serial) -def get_device(serial=None, product=None, adb_path='adb'): +def get_device( + serial: str | None = None, product: str | None = None, adb_path: str = 'adb' +) -> AndroidDevice: """Get a uniquely identified AndroidDevice if one is available. Raises: @@ -113,7 +124,7 @@ def get_device(serial=None, product=None, adb_path='adb'): return _get_unique_device(product, adb_path=adb_path) -def _get_device_by_type(flag, adb_path): +def _get_device_by_type(flag: str, adb_path: str) -> AndroidDevice: with open(os.devnull, 'wb') as devnull: subprocess.check_call([adb_path, 'start-server'], stdout=devnull, stderr=devnull) @@ -127,7 +138,7 @@ def _get_device_by_type(flag, adb_path): return _get_device_by_serial(serial, adb_path=adb_path) -def get_usb_device(adb_path='adb'): +def get_usb_device(adb_path: str = 'adb') -> AndroidDevice: """Get the unique USB-connected AndroidDevice if it is available. Raises: @@ -140,7 +151,7 @@ def get_usb_device(adb_path='adb'): return _get_device_by_type('-d', adb_path=adb_path) -def get_emulator_device(adb_path='adb'): +def get_emulator_device(adb_path: str = 'adb') -> AndroidDevice: """Get the unique emulator AndroidDevice if it is available. Raises: @@ -153,74 +164,13 @@ def get_emulator_device(adb_path='adb'): return _get_device_by_type('-e', adb_path=adb_path) -# If necessary, modifies subprocess.check_output() or subprocess.Popen() args -# to run the subprocess via Windows PowerShell to work-around an issue in -# Python 2's subprocess class on Windows where it doesn't support Unicode. -def _get_subprocess_args(args): - # Only do this slow work-around if Unicode is in the cmd line on Windows. - # PowerShell takes 600-700ms to startup on a 2013-2014 machine, which is - # very slow. - if os.name != 'nt' or all(not isinstance(arg, unicode) for arg in args[0]): - return args - - def escape_arg(arg): - # Escape for the parsing that the C Runtime does in Windows apps. In - # particular, this will take care of double-quotes. - arg = subprocess.list2cmdline([arg]) - # Escape single-quote with another single-quote because we're about - # to... - arg = arg.replace(u"'", u"''") - # ...put the arg in a single-quoted string for PowerShell to parse. - arg = u"'" + arg + u"'" - return arg - - # Escape command line args. - argv = map(escape_arg, args[0]) - # Cause script errors (such as adb not found) to stop script immediately - # with an error. - ps_code = u'$ErrorActionPreference = "Stop"\r\n' - # Add current directory to the PATH var, to match cmd.exe/CreateProcess() - # behavior. - ps_code += u'$env:Path = ".;" + $env:Path\r\n' - # Precede by &, the PowerShell call operator, and separate args by space. - ps_code += u'& ' + u' '.join(argv) - # Make the PowerShell exit code the exit code of the subprocess. - ps_code += u'\r\nExit $LastExitCode' - # Encode as UTF-16LE (without Byte-Order-Mark) which Windows natively - # understands. - ps_code = ps_code.encode('utf-16le') - - # Encode the PowerShell command as base64 and use the special - # -EncodedCommand option that base64 decodes. Base64 is just plain ASCII, - # so it should have no problem passing through Win32 CreateProcessA() - # (which python erroneously calls instead of CreateProcessW()). - return (['powershell.exe', '-NoProfile', '-NonInteractive', - '-EncodedCommand', base64.b64encode(ps_code)],) + args[1:] - - -# Call this instead of subprocess.check_output() to work-around issue in Python -# 2's subprocess class on Windows where it doesn't support Unicode. -def _subprocess_check_output(*args, **kwargs): - try: - return subprocess.check_output(*_get_subprocess_args(args), **kwargs) - except subprocess.CalledProcessError as e: - # Show real command line instead of the powershell.exe command line. - raise subprocess.CalledProcessError(e.returncode, args[0], - output=e.output) - - -# Call this instead of subprocess.Popen(). Like _subprocess_check_output(). -def _subprocess_Popen(*args, **kwargs): - return subprocess.Popen(*_get_subprocess_args(args), **kwargs) - - -def split_lines(s): +def split_lines(s: str) -> list[str]: """Splits lines in a way that works even on Windows and old devices. Windows will see \r\n instead of \n, old devices do the same, old devices on Windows will see \r\r\n. """ - # rstrip is used here to workaround a difference between splineslines and + # rstrip is used here to workaround a difference between splitlines and # re.split: # >>> 'foo\n'.splitlines() # ['foo'] @@ -229,12 +179,11 @@ def split_lines(s): return re.split(r'[\r\n]+', s.rstrip()) -def version(adb_path=None): +def version(adb_path: list[str] | None = None) -> int: """Get the version of adb (in terms of ADB_SERVER_VERSION).""" adb_path = adb_path if adb_path is not None else ['adb'] - version_output = subprocess.check_output(adb_path + ['version']) - version_output = version_output.decode('utf-8') + version_output = subprocess.check_output(adb_path + ['version'], encoding='utf-8') pattern = r'^Android Debug Bridge version 1.0.(\d+)$' result = re.match(pattern, version_output.splitlines()[0]) if not result: @@ -259,28 +208,30 @@ class AndroidDevice(object): _RETURN_CODE_SEARCH_LENGTH = len( '{0}255\r\r\n'.format(_RETURN_CODE_DELIMITER)) - def __init__(self, serial, product=None, adb_path='adb'): + def __init__( + self, serial: str | None, product: str | None = None, adb_path: str = 'adb' + ) -> None: self.serial = serial self.product = product self.adb_path = adb_path self.adb_cmd = [adb_path] if self.serial is not None: - self.adb_cmd.extend(['-s', serial]) + self.adb_cmd.extend(['-s', self.serial]) if self.product is not None: - self.adb_cmd.extend(['-p', product]) - self._linesep = None - self._features = None + self.adb_cmd.extend(['-p', self.product]) + self._linesep: str | None = None + self._features: list[str] | None = None @property - def linesep(self): + def linesep(self) -> str: if self._linesep is None: self._linesep = subprocess.check_output( - self.adb_cmd + ['shell', 'echo']).decode('utf-8') + self.adb_cmd + ['shell', 'echo'], encoding='utf-8') return self._linesep @property - def features(self): + def features(self) -> list[str]: if self._features is None: try: self._features = split_lines(self._simple_call(['features'])) @@ -288,16 +239,16 @@ class AndroidDevice(object): self._features = [] return self._features - def has_shell_protocol(self): + def has_shell_protocol(self) -> bool: return version(self.adb_cmd) >= 35 and 'shell_v2' in self.features - def _make_shell_cmd(self, user_cmd): + def _make_shell_cmd(self, user_cmd: list[str]) -> list[str]: command = self.adb_cmd + ['shell'] + user_cmd if not self.has_shell_protocol(): command += self._RETURN_CODE_PROBE return command - def _parse_shell_output(self, out): + def _parse_shell_output(self, out: str) -> tuple[int, str]: """Finds the exit code string from shell output. Args: @@ -325,12 +276,12 @@ class AndroidDevice(object): out = out[:-len(partition[1]) - len(partition[2])] return result, out - def _simple_call(self, cmd): + def _simple_call(self, cmd: list[str]) -> str: logging.info(' '.join(self.adb_cmd + cmd)) - return _subprocess_check_output( + return subprocess.check_output( self.adb_cmd + cmd, stderr=subprocess.STDOUT).decode('utf-8') - def shell(self, cmd): + def shell(self, cmd: list[str]) -> tuple[str, str]: """Calls `adb shell` Args: @@ -348,7 +299,7 @@ class AndroidDevice(object): raise ShellError(cmd, stdout, stderr, exit_code) return stdout, stderr - def shell_nocheck(self, cmd): + def shell_nocheck(self, cmd: list[str]) -> tuple[int, str, str]: """Calls `adb shell` Args: @@ -360,19 +311,23 @@ class AndroidDevice(object): """ cmd = self._make_shell_cmd(cmd) logging.info(' '.join(cmd)) - p = _subprocess_Popen( - cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + p = subprocess.Popen( + cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, encoding='utf-8') stdout, stderr = p.communicate() - stdout = stdout.decode('utf-8') - stderr = stderr.decode('utf-8') if self.has_shell_protocol(): exit_code = p.returncode else: exit_code, stdout = self._parse_shell_output(stdout) return exit_code, stdout, stderr - def shell_popen(self, cmd, kill_atexit=True, preexec_fn=None, - creationflags=0, **kwargs): + def shell_popen( + self, + cmd: list[str], + kill_atexit: bool = True, + preexec_fn: Callable[[], None] | None = None, + creationflags: int = 0, + **kwargs: Any, + ) -> subprocess.Popen[Any]: """Calls `adb shell` and returns a handle to the adb process. This function provides direct access to the subprocess used to run the @@ -400,27 +355,27 @@ class AndroidDevice(object): preexec_fn = os.setpgrp elif preexec_fn is not os.setpgrp: fn = preexec_fn - def _wrapper(): + def _wrapper() -> None: fn() os.setpgrp() preexec_fn = _wrapper - p = _subprocess_Popen(command, creationflags=creationflags, - preexec_fn=preexec_fn, **kwargs) + p = subprocess.Popen(command, creationflags=creationflags, + preexec_fn=preexec_fn, **kwargs) if kill_atexit: atexit.register(p.kill) return p - def install(self, filename, replace=False): + def install(self, filename: str, replace: bool = False) -> str: cmd = ['install'] if replace: cmd.append('-r') cmd.append(filename) return self._simple_call(cmd) - def push(self, local, remote, sync=False): + def push(self, local: str | list[str], remote: str, sync: bool = False) -> str: """Transfer a local file or directory to the device. Args: @@ -430,7 +385,7 @@ class AndroidDevice(object): those on the device. If False, transfers all files. Returns: - Exit status of the push command. + Output of the command. """ cmd = ['push'] if sync: @@ -444,73 +399,73 @@ class AndroidDevice(object): return self._simple_call(cmd) - def pull(self, remote, local): + def pull(self, remote: str, local: str) -> str: return self._simple_call(['pull', remote, local]) - def sync(self, directory=None): + def sync(self, directory: str | None = None) -> str: cmd = ['sync'] if directory is not None: cmd.append(directory) return self._simple_call(cmd) - def tcpip(self, port): + def tcpip(self, port: str) -> str: return self._simple_call(['tcpip', port]) - def usb(self): + def usb(self) -> str: return self._simple_call(['usb']) - def reboot(self): + def reboot(self) -> str: return self._simple_call(['reboot']) - def remount(self): + def remount(self) -> str: return self._simple_call(['remount']) - def root(self): + def root(self) -> str: return self._simple_call(['root']) - def unroot(self): + def unroot(self) -> str: return self._simple_call(['unroot']) - def connect(self, host): + def connect(self, host: str) -> str: return self._simple_call(['connect', host]) - def disconnect(self, host): + def disconnect(self, host: str) -> str: return self._simple_call(['disconnect', host]) - def forward(self, local, remote): + def forward(self, local: str, remote: str) -> str: return self._simple_call(['forward', local, remote]) - def forward_list(self): + def forward_list(self) -> str: return self._simple_call(['forward', '--list']) - def forward_no_rebind(self, local, remote): + def forward_no_rebind(self, local: str, remote: str) -> str: return self._simple_call(['forward', '--no-rebind', local, remote]) - def forward_remove(self, local): + def forward_remove(self, local: str) -> str: return self._simple_call(['forward', '--remove', local]) - def forward_remove_all(self): + def forward_remove_all(self) -> str: return self._simple_call(['forward', '--remove-all']) - def reverse(self, remote, local): + def reverse(self, remote: str, local: str) -> str: return self._simple_call(['reverse', remote, local]) - def reverse_list(self): + def reverse_list(self) -> str: return self._simple_call(['reverse', '--list']) - def reverse_no_rebind(self, local, remote): + def reverse_no_rebind(self, local: str, remote: str) -> str: return self._simple_call(['reverse', '--no-rebind', local, remote]) - def reverse_remove_all(self): + def reverse_remove_all(self) -> str: return self._simple_call(['reverse', '--remove-all']) - def reverse_remove(self, remote): + def reverse_remove(self, remote: str) -> str: return self._simple_call(['reverse', '--remove', remote]) - def wait(self): + def wait(self) -> str: return self._simple_call(['wait-for-device']) - def get_prop(self, prop_name): + def get_prop(self, prop_name: str) -> str | None: output = split_lines(self.shell(['getprop', prop_name])[0]) if len(output) != 1: raise RuntimeError('Too many lines in getprop output:\n' + @@ -520,7 +475,7 @@ class AndroidDevice(object): return None return value - def set_prop(self, prop_name, value): + def set_prop(self, prop_name: str, value: str) -> None: self.shell(['setprop', prop_name, value]) def logcat(self) -> str: diff --git a/python-packages/adb/py.typed b/python-packages/adb/adb/py.typed similarity index 100% rename from python-packages/adb/py.typed rename to python-packages/adb/adb/py.typed diff --git a/python-packages/adb/mypy.ini b/python-packages/adb/mypy.ini new file mode 100644 index 000000000..dd92b0269 --- /dev/null +++ b/python-packages/adb/mypy.ini @@ -0,0 +1,19 @@ +[mypy] +check_untyped_defs = true +disallow_any_generics = true +disallow_any_unimported = true +disallow_subclassing_any = true +disallow_untyped_decorators = true +disallow_untyped_defs = true +follow_imports = silent +implicit_reexport = false +namespace_packages = true +no_implicit_optional = true +show_error_codes = true +strict_equality = true +warn_redundant_casts = true +# Can't enable because mypy cannot reason about _get_subprocess_args. +# warn_return_any = true +warn_unreachable = true +warn_unused_configs = true +warn_unused_ignores = true diff --git a/python-packages/adb/setup.py b/python-packages/adb/setup.py index 4322fb785..93a2e1794 100644 --- a/python-packages/adb/setup.py +++ b/python-packages/adb/setup.py @@ -22,7 +22,6 @@ setup( description='A Python interface to the Android Debug Bridge.', license='Apache 2.0', keywords='adb android', - package_dir={'adb': ''}, packages=['adb'], package_data={'adb': ['py.typed']}, classifiers=[ diff --git a/python-packages/adb/test.py b/python-packages/adb/tests/test_adb.py similarity index 76% rename from python-packages/adb/test.py rename to python-packages/adb/tests/test_adb.py index c44688377..c3e45891f 100644 --- a/python-packages/adb/test.py +++ b/python-packages/adb/tests/test_adb.py @@ -15,64 +15,64 @@ # import os import unittest -import mock +from unittest.mock import Mock, patch import adb class GetDeviceTest(unittest.TestCase): - def setUp(self): + def setUp(self) -> None: self.android_serial = os.getenv('ANDROID_SERIAL') if 'ANDROID_SERIAL' in os.environ: del os.environ['ANDROID_SERIAL'] - def tearDown(self): + def tearDown(self) -> None: if self.android_serial is not None: os.environ['ANDROID_SERIAL'] = self.android_serial else: if 'ANDROID_SERIAL' in os.environ: del os.environ['ANDROID_SERIAL'] - @mock.patch('adb.device.get_devices') - def test_explicit(self, mock_get_devices): + @patch('adb.get_devices') + def test_explicit(self, mock_get_devices: Mock) -> None: mock_get_devices.return_value = ['foo', 'bar'] device = adb.get_device('foo') self.assertEqual(device.serial, 'foo') - @mock.patch('adb.device.get_devices') - def test_from_env(self, mock_get_devices): + @patch('adb.get_devices') + def test_from_env(self, mock_get_devices: Mock) -> None: mock_get_devices.return_value = ['foo', 'bar'] os.environ['ANDROID_SERIAL'] = 'foo' device = adb.get_device() self.assertEqual(device.serial, 'foo') - @mock.patch('adb.device.get_devices') - def test_arg_beats_env(self, mock_get_devices): + @patch('adb.get_devices') + def test_arg_beats_env(self, mock_get_devices: Mock) -> None: mock_get_devices.return_value = ['foo', 'bar'] os.environ['ANDROID_SERIAL'] = 'bar' device = adb.get_device('foo') self.assertEqual(device.serial, 'foo') - @mock.patch('adb.device.get_devices') - def test_no_such_device(self, mock_get_devices): + @patch('adb.get_devices') + def test_no_such_device(self, mock_get_devices: Mock) -> None: mock_get_devices.return_value = ['foo', 'bar'] self.assertRaises(adb.DeviceNotFoundError, adb.get_device, ['baz']) os.environ['ANDROID_SERIAL'] = 'baz' self.assertRaises(adb.DeviceNotFoundError, adb.get_device) - @mock.patch('adb.device.get_devices') - def test_unique_device(self, mock_get_devices): + @patch('adb.get_devices') + def test_unique_device(self, mock_get_devices: Mock) -> None: mock_get_devices.return_value = ['foo'] device = adb.get_device() self.assertEqual(device.serial, 'foo') - @mock.patch('adb.device.get_devices') - def test_no_unique_device(self, mock_get_devices): + @patch('adb.get_devices') + def test_no_unique_device(self, mock_get_devices: Mock) -> None: mock_get_devices.return_value = ['foo', 'bar'] self.assertRaises(adb.NoUniqueDeviceError, adb.get_device) -def main(): +def main() -> None: suite = unittest.TestLoader().loadTestsFromName(__name__) unittest.TextTestRunner(verbosity=3).run(suite)