tequilaOS/platform_system_core
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

am 20cf8702: Merge "Create adb Python package."

Browse source 
* commit '20cf8702a37c2517cb1225babaa81747b8443376':
  Create adb Python package.
This commit is contained in:
Dan Albert 2015-07-27 23:13:19 +00:00 committed by Android Git Automerger
commit 05432f07eb
5 changed files with 751 additions and 496 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
adb/__init__.py Normal file
View file

@ -0,0 +1,17 @@
#
# 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

233
adb/device.py Normal file
View file

@ -0,0 +1,233 @@
#
# 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.
#
import os
import re
import subprocess
class FindDeviceError(RuntimeError):
pass
class DeviceNotFoundError(FindDeviceError):
def __init__(self, serial):
self.serial = serial
super(DeviceNotFoundError, self).__init__(
'No device with serial {}'.format(serial))
class NoUniqueDeviceError(FindDeviceError):
def __init__(self):
super(NoUniqueDeviceError, self).__init__('No unique device')
def get_devices():
with open(os.devnull, 'wb') as devnull:
subprocess.check_call(['adb', 'start-server'], stdout=devnull,
stderr=devnull)
out = subprocess.check_output(['adb', 'devices']).splitlines()
# The first line of `adb devices` just says "List of attached devices", so
# skip that.
devices = []
for line in out[1:]:
if not line.strip():
continue
if 'offline' in line:
continue
serial, _ = re.split(r'\s+', line, maxsplit=1)
devices.append(serial)
return devices
def _get_unique_device(product=None):
devices = get_devices()
if len(devices) != 1:
raise NoUniqueDeviceError()
return AndroidDevice(devices[0], product)
def _get_device_by_serial(serial, product=None):
for device in get_devices():
if device == serial:
return AndroidDevice(serial, product)
raise DeviceNotFoundError(serial)
def get_device(serial=None, product=None):
"""Get a uniquely identified AndroidDevice if one is available.
Raises:
DeviceNotFoundError:
The serial specified by `serial` or $ANDROID_SERIAL is not
connected.
NoUniqueDeviceError:
Neither `serial` nor $ANDROID_SERIAL was set, and the number of
devices connected to the system is not 1. Having 0 connected
devices will also result in this error.
Returns:
An AndroidDevice associated with the first non-None identifier in the
following order of preference:
1) The `serial` argument.
2) The environment variable $ANDROID_SERIAL.
3) The single device connnected to the system.
"""
if serial is not None:
return _get_device_by_serial(serial, product)
android_serial = os.getenv('ANDROID_SERIAL')
if android_serial is not None:
return _get_device_by_serial(android_serial, product)
return _get_unique_device(product)
class AndroidDevice(object):
def __init__(self, serial, product=None):
self.serial = serial
self.product = product
self.adb_cmd = ['adb']
if self.serial is not None:
self.adb_cmd.extend(['-s', serial])
if self.product is not None:
self.adb_cmd.extend(['-p', product])
self._linesep = None
self._shell_result_pattern = None
@property
def linesep(self):
if self._linesep is None:
self._linesep = subprocess.check_output(['adb', 'shell', 'echo'])
return self._linesep
def _make_shell_cmd(self, user_cmd):
# Follow any shell command with `; echo; echo $?` to get the exit
# status of a program since this isn't propagated by adb.
#
# The leading newline is needed because `printf 1; echo $?` would print
# "10", and we wouldn't be able to distinguish the exit code.
rc_probe = '; echo "\n$?"'
return self.adb_cmd + ['shell'] + user_cmd + [rc_probe]
def _parse_shell_output(self, out): # pylint: disable=no-self-use
search_text = out
max_result_len = len('{0}255{0}'.format(self.linesep))
if len(search_text) > max_result_len:
# We don't want to regex match over massive amounts of data when we
# know the part we want is right at the end.
search_text = search_text[-max_result_len:]
if self._shell_result_pattern is None:
self._shell_result_pattern = re.compile(
r'({0}\d+{0})$'.format(self.linesep), re.MULTILINE)
m = self._shell_result_pattern.search(search_text)
if m is None:
raise RuntimeError('Could not find exit status in shell output.')
result_text = m.group(1)
result = int(result_text.strip())
out = out[:-len(result_text)] # Trim the result text from the output.
return result, out
def _simple_call(self, cmd):
return subprocess.check_output(
self.adb_cmd + cmd, stderr=subprocess.STDOUT)
def shell(self, cmd):
cmd = self._make_shell_cmd(cmd)
out = subprocess.check_output(cmd)
rc, out = self._parse_shell_output(out)
if rc != 0:
error = subprocess.CalledProcessError(rc, cmd)
error.out = out
raise error
return out
def shell_nocheck(self, cmd):
cmd = self._make_shell_cmd(cmd)
p = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
out, _ = p.communicate()
return self._parse_shell_output(out)
def install(self, filename):
return self._simple_call(['install', filename])
def push(self, local, remote):
return self._simple_call(['push', local, remote])
def pull(self, remote, local):
return self._simple_call(['pull', remote, local])
def sync(self, directory=None):
cmd = ['sync']
if directory is not None:
cmd.append(directory)
return self._simple_call(cmd)
def forward(self, local, remote):
return self._simple_call(['forward', local, remote])
def tcpip(self, port):
return self._simple_call(['tcpip', port])
def usb(self):
return self._simple_call(['usb'])
def root(self):
return self._simple_call(['root'])
def unroot(self):
return self._simple_call(['unroot'])
def forward_remove(self, local):
return self._simple_call(['forward', '--remove', local])
def forward_remove_all(self):
return self._simple_call(['forward', '--remove-all'])
def connect(self, host):
return self._simple_call(['connect', host])
def disconnect(self, host):
return self._simple_call(['disconnect', host])
def reverse(self, remote, local):
return self._simple_call(['reverse', remote, local])
def reverse_remove_all(self):
return self._simple_call(['reverse', '--remove-all'])
def reverse_remove(self, remote):
return self._simple_call(['reverse', '--remove', remote])
def wait(self):
return self._simple_call(['wait-for-device'])
def get_prop(self, prop_name):
output = self.shell(['getprop', prop_name])
if len(output) != 1:
raise RuntimeError('Too many lines in getprop output:\n' +
'\n'.join(output))
value = output[0]
if not value.strip():
return None
return value
def set_prop(self, prop_name, value):
self.shell(['setprop', prop_name, value])

77
adb/test_adb.py Normal file
View file

@ -0,0 +1,77 @@
#!/usr/bin/env python
#
# 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.
#
"""Tests for the adb program itself.
This differs from things in test_device.py in that there is no API for these
things. Most of these tests involve specific error messages or the help text.
"""
from __future__ import print_function
import random
import subprocess
import unittest
import adb
class NonApiTest(unittest.TestCase):
"""Tests for ADB that aren't a part of the AndroidDevice API."""
def test_help(self):
"""Make sure we get _something_ out of help."""
out = subprocess.check_output(
['adb', 'help'], stderr=subprocess.STDOUT)
self.assertGreater(len(out), 0)
def test_version(self):
"""Get a version number out of the output of adb."""
lines = subprocess.check_output(['adb', 'version']).splitlines()
version_line = lines[0]
self.assertRegexpMatches(
version_line, r'^Android Debug Bridge version \d+\.\d+\.\d+$')
if len(lines) == 2:
# Newer versions of ADB have a second line of output for the
# version that includes a specific revision (git SHA).
revision_line = lines[1]
self.assertRegexpMatches(
revision_line, r'^Revision [0-9a-f]{12}-android$')
def test_tcpip_error_messages(self):
p = subprocess.Popen(['adb', 'tcpip'], stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
out, _ = p.communicate()
self.assertEqual(1, p.returncode)
self.assertIn('help message', out)
p = subprocess.Popen(['adb', 'tcpip', 'foo'], stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
out, _ = p.communicate()
self.assertEqual(1, p.returncode)
self.assertIn('error', out)
def main():
random.seed(0)
if len(adb.get_devices()) > 0:
suite = unittest.TestLoader().loadTestsFromName(__name__)
unittest.TextTestRunner(verbosity=3).run(suite)
else:
print('Test suite must be run with attached devices')
if __name__ == '__main__':
main()

424
adb/test_device.py Normal file
View file

@ -0,0 +1,424 @@
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# 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 print_function
import hashlib
import os
import posixpath
import random
import shlex
import shutil
import subprocess
import tempfile
import unittest
import mock
import adb
class GetDeviceTest(unittest.TestCase):
def setUp(self):
self.android_serial = os.getenv('ANDROID_SERIAL')
del os.environ['ANDROID_SERIAL']
def tearDown(self):
os.environ['ANDROID_SERIAL'] = self.android_serial
@mock.patch('adb.device.get_devices')
def test_explicit(self, mock_get_devices):
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):
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):
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):
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):
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):
mock_get_devices.return_value = ['foo', 'bar']
self.assertRaises(adb.NoUniqueDeviceError, adb.get_device)
class DeviceTest(unittest.TestCase):
def setUp(self):
self.device = adb.get_device()
class ShellTest(DeviceTest):
def test_cat(self):
"""Check that we can at least cat a file."""
out = self.device.shell(['cat', '/proc/uptime']).strip()
elements = out.split()
self.assertEqual(len(elements), 2)
uptime, idle = elements
self.assertGreater(float(uptime), 0.0)
self.assertGreater(float(idle), 0.0)
def test_throws_on_failure(self):
self.assertRaises(subprocess.CalledProcessError,
self.device.shell, ['false'])
def test_output_not_stripped(self):
out = self.device.shell(['echo', 'foo'])
self.assertEqual(out, 'foo' + self.device.linesep)
def test_shell_nocheck_failure(self):
rc, out = self.device.shell_nocheck(['false'])
self.assertNotEqual(rc, 0)
self.assertEqual(out, '')
def test_shell_nocheck_output_not_stripped(self):
rc, out = self.device.shell_nocheck(['echo', 'foo'])
self.assertEqual(rc, 0)
self.assertEqual(out, 'foo' + self.device.linesep)
def test_can_distinguish_tricky_results(self):
# If result checking on ADB shell is naively implemented as
# `adb shell <cmd>; echo $?`, we would be unable to distinguish the
# output from the result for a cmd of `echo -n 1`.
rc, out = self.device.shell_nocheck(['echo', '-n', '1'])
self.assertEqual(rc, 0)
self.assertEqual(out, '1')
def test_line_endings(self):
"""Ensure that line ending translation is not happening in the pty.
Bug: http://b/19735063
"""
output = self.device.shell(['uname'])
self.assertEqual(output, 'Linux' + self.device.linesep)
class ArgumentEscapingTest(DeviceTest):
def test_shell_escaping(self):
"""Make sure that argument escaping is somewhat sane."""
# http://b/19734868
# Note that this actually matches ssh(1)'s behavior --- it's
# converted to `sh -c echo hello; echo world` which sh interprets
# as `sh -c echo` (with an argument to that shell of "hello"),
# and then `echo world` back in the first shell.
result = self.device.shell(
shlex.split("sh -c 'echo hello; echo world'"))
result = result.splitlines()
self.assertEqual(['', 'world'], result)
# If you really wanted "hello" and "world", here's what you'd do:
result = self.device.shell(
shlex.split(r'echo hello\;echo world')).splitlines()
self.assertEqual(['hello', 'world'], result)
# http://b/15479704
result = self.device.shell(shlex.split("'true && echo t'")).strip()
self.assertEqual('t', result)
result = self.device.shell(
shlex.split("sh -c 'true && echo t'")).strip()
self.assertEqual('t', result)
# http://b/20564385
result = self.device.shell(shlex.split('FOO=a BAR=b echo t')).strip()
self.assertEqual('t', result)
result = self.device.shell(shlex.split(r'echo -n 123\;uname')).strip()
self.assertEqual('123Linux', result)
def test_install_argument_escaping(self):
"""Make sure that install argument escaping works."""
# http://b/20323053
tf = tempfile.NamedTemporaryFile('wb', suffix='-text;ls;1.apk')
self.assertIn("-text;ls;1.apk", self.device.install(tf.name))
# http://b/3090932
tf = tempfile.NamedTemporaryFile('wb', suffix="-Live Hold'em.apk")
self.assertIn("-Live Hold'em.apk", self.device.install(tf.name))
class RootUnrootTest(DeviceTest):
def _test_root(self):
message = self.device.root()
if 'adbd cannot run as root in production builds' in message:
return
self.device.wait()
self.assertEqual('root', self.device.shell(['id', '-un']).strip())
def _test_unroot(self):
self.device.unroot()
self.device.wait()
self.assertEqual('shell', self.device.shell(['id', '-un']).strip())
def test_root_unroot(self):
"""Make sure that adb root and adb unroot work, using id(1)."""
original_user = self.device.shell(['id', '-un']).strip()
try:
if original_user == 'root':
self._test_unroot()
self._test_root()
elif original_user == 'shell':
self._test_root()
self._test_unroot()
finally:
if original_user == 'root':
self.device.root()
else:
self.device.unroot()
self.device.wait()
class TcpIpTest(DeviceTest):
def test_tcpip_failure_raises(self):
"""adb tcpip requires a port.
Bug: http://b/22636927
"""
self.assertRaises(
subprocess.CalledProcessError, self.device.tcpip, '')
self.assertRaises(
subprocess.CalledProcessError, self.device.tcpip, 'foo')
def compute_md5(string):
hsh = hashlib.md5()
hsh.update(string)
return hsh.hexdigest()
def get_md5_prog(device):
"""Older platforms (pre-L) had the name md5 rather than md5sum."""
try:
device.shell(['md5sum', '/proc/uptime'])
return 'md5sum'
except subprocess.CalledProcessError:
return 'md5'
class HostFile(object):
def __init__(self, handle, checksum):
self.handle = handle
self.checksum = checksum
self.full_path = handle.name
self.base_name = os.path.basename(self.full_path)
class DeviceFile(object):
def __init__(self, checksum, full_path):
self.checksum = checksum
self.full_path = full_path
self.base_name = posixpath.basename(self.full_path)
def make_random_host_files(in_dir, num_files):
min_size = 1 * (1 << 10)
max_size = 16 * (1 << 10)
files = []
for _ in xrange(num_files):
file_handle = tempfile.NamedTemporaryFile(dir=in_dir, delete=False)
size = random.randrange(min_size, max_size, 1024)
rand_str = os.urandom(size)
file_handle.write(rand_str)
file_handle.flush()
file_handle.close()
md5 = compute_md5(rand_str)
files.append(HostFile(file_handle, md5))
return files
def make_random_device_files(device, in_dir, num_files):
min_size = 1 * (1 << 10)
max_size = 16 * (1 << 10)
files = []
for file_num in xrange(num_files):
size = random.randrange(min_size, max_size, 1024)
base_name = 'device_tmpfile' + str(file_num)
full_path = os.path.join(in_dir, base_name)
device.shell(['dd', 'if=/dev/urandom', 'of={}'.format(full_path),
'bs={}'.format(size), 'count=1'])
dev_md5, _ = device.shell([get_md5_prog(device), full_path]).split()
files.append(DeviceFile(dev_md5, full_path))
return files
class FileOperationsTest(DeviceTest):
SCRATCH_DIR = '/data/local/tmp'
DEVICE_TEMP_FILE = SCRATCH_DIR + '/adb_test_file'
DEVICE_TEMP_DIR = SCRATCH_DIR + '/adb_test_dir'
def _test_push(self, local_file, checksum):
self.device.shell(['rm', '-rf', self.DEVICE_TEMP_FILE])
try:
self.device.push(
local=local_file, remote=self.DEVICE_TEMP_FILE)
dev_md5, _ = self.device.shell(
[get_md5_prog(self.device), self.DEVICE_TEMP_FILE]).split()
self.assertEqual(checksum, dev_md5)
finally:
self.device.shell(['rm', '-f', self.DEVICE_TEMP_FILE])
def test_push(self):
"""Push a randomly generated file to specified device."""
kbytes = 512
tmp = tempfile.NamedTemporaryFile(mode='wb', delete=False)
try:
rand_str = os.urandom(1024 * kbytes)
tmp.write(rand_str)
tmp.close()
self._test_push(tmp.name, compute_md5(rand_str))
finally:
os.remove(tmp.name)
# TODO: write push directory test.
def _test_pull(self, remote_file, checksum):
tmp_write = tempfile.NamedTemporaryFile(mode='wb', delete=False)
try:
tmp_write.close()
self.device.pull(remote=remote_file, local=tmp_write.name)
with open(tmp_write.name, 'rb') as tmp_read:
host_contents = tmp_read.read()
host_md5 = compute_md5(host_contents)
self.assertEqual(checksum, host_md5)
finally:
os.remove(tmp_write.name)
def test_pull(self):
"""Pull a randomly generated file from specified device."""
kbytes = 512
self.device.shell(['rm', '-rf', self.DEVICE_TEMP_FILE])
try:
cmd = ['dd', 'if=/dev/urandom',
'of={}'.format(self.DEVICE_TEMP_FILE), 'bs=1024',
'count={}'.format(kbytes)]
self.device.shell(cmd)
dev_md5, _ = self.device.shell(
[get_md5_prog(self.device), self.DEVICE_TEMP_FILE]).split()
self._test_pull(self.DEVICE_TEMP_FILE, dev_md5)
finally:
self.device.shell_nocheck(['rm', self.DEVICE_TEMP_FILE])
def test_pull_dir(self):
"""Pull a randomly generated directory of files from the device."""
host_dir = tempfile.mkdtemp()
try:
self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
# Populate device directory with random files.
temp_files = make_random_device_files(
self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
self.device.pull(remote=self.DEVICE_TEMP_DIR, local=host_dir)
for temp_file in temp_files:
host_path = os.path.join(host_dir, temp_file.base_name)
with open(host_path, 'rb') as host_file:
host_md5 = compute_md5(host_file.read())
self.assertEqual(host_md5, temp_file.checksum)
finally:
self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
if host_dir is not None:
shutil.rmtree(host_dir)
def test_sync(self):
"""Sync a randomly generated directory of files to specified device."""
base_dir = tempfile.mkdtemp()
try:
# Create mirror device directory hierarchy within base_dir.
full_dir_path = base_dir + self.DEVICE_TEMP_DIR
os.makedirs(full_dir_path)
# Create 32 random files within the host mirror.
temp_files = make_random_host_files(in_dir=full_dir_path,
num_files=32)
# Clean up any trash on the device.
device = adb.get_device(product=base_dir)
device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
device.sync('data')
# Confirm that every file on the device mirrors that on the host.
for temp_file in temp_files:
device_full_path = posixpath.join(
self.DEVICE_TEMP_DIR, temp_file.base_name)
dev_md5, _ = device.shell(
[get_md5_prog(self.device), device_full_path]).split()
self.assertEqual(temp_file.checksum, dev_md5)
finally:
self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
shutil.rmtree(base_dir + self.DEVICE_TEMP_DIR)
def test_unicode_paths(self):
"""Ensure that we can support non-ASCII paths, even on Windows."""
name = u'로보카 폴리'.encode('utf-8')
## push.
tf = tempfile.NamedTemporaryFile('wb', suffix=name)
self.device.push(tf.name, '/data/local/tmp/adb-test-{}'.format(name))
self.device.shell(['rm', '-f', '/data/local/tmp/adb-test-*'])
# pull.
cmd = ['touch', '"/data/local/tmp/adb-test-{}"'.format(name)]
self.device.shell(cmd)
tf = tempfile.NamedTemporaryFile('wb', suffix=name)
self.device.pull('/data/local/tmp/adb-test-{}'.format(name), tf.name)
def main():
random.seed(0)
if len(adb.get_devices()) > 0:
suite = unittest.TestLoader().loadTestsFromName(__name__)
unittest.TextTestRunner(verbosity=3).run(suite)
else:
print('Test suite must be run with attached devices')
if __name__ == '__main__':
main()

496
adb/tests/test_adb.py
View file

@ -1,496 +0,0 @@
#!/usr/bin/env python2
# -*- coding: utf-8 -*-
"""Simple conformance test for adb.
This script will use the available adb in path and run simple
tests that attempt to touch all accessible attached devices.
"""
import hashlib
import os
import pipes
import posixpath
import random
import re
import shlex
import subprocess
import sys
import tempfile
import unittest
def trace(cmd):
"""Print debug message if tracing enabled."""
if False:
print >> sys.stderr, cmd
def call(cmd_str):
"""Run process and return output tuple (stdout, stderr, ret code)."""
trace(cmd_str)
process = subprocess.Popen(shlex.split(cmd_str),
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout, stderr = process.communicate()
return stdout, stderr, process.returncode
def call_combined(cmd_str):
"""Run process and return output tuple (stdout+stderr, ret code)."""
trace(cmd_str)
process = subprocess.Popen(shlex.split(cmd_str),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT)
stdout, _ = process.communicate()
return stdout, process.returncode
def call_checked(cmd_str):
"""Run process and get stdout+stderr, raise an exception on trouble."""
trace(cmd_str)
return subprocess.check_output(shlex.split(cmd_str),
stderr=subprocess.STDOUT)
def call_checked_list(cmd_str):
return call_checked(cmd_str).split('\n')
def call_checked_list_skip(cmd_str):
out_list = call_checked_list(cmd_str)
def is_init_line(line):
if (len(line) >= 3) and (line[0] == "*") and (line[-2] == "*"):
return True
else:
return False
return [line for line in out_list if not is_init_line(line)]
def get_device_list():
output = call_checked_list_skip("adb devices")
dev_list = []
for line in output[1:]:
if line.strip() == "":
continue
device, _ = line.split()
dev_list.append(device)
return dev_list
def get_attached_device_count():
return len(get_device_list())
def compute_md5(string):
hsh = hashlib.md5()
hsh.update(string)
return hsh.hexdigest()
class HostFile(object):
def __init__(self, handle, md5):
self.handle = handle
self.md5 = md5
self.full_path = handle.name
self.base_name = os.path.basename(self.full_path)
class DeviceFile(object):
def __init__(self, md5, full_path):
self.md5 = md5
self.full_path = full_path
self.base_name = posixpath.basename(self.full_path)
def make_random_host_files(in_dir, num_files, rand_size=True):
files = {}
min_size = 1 * (1 << 10)
max_size = 16 * (1 << 10)
fixed_size = min_size
for _ in range(num_files):
file_handle = tempfile.NamedTemporaryFile(dir=in_dir, delete=False)
if rand_size:
size = random.randrange(min_size, max_size, 1024)
else:
size = fixed_size
rand_str = os.urandom(size)
file_handle.write(rand_str)
file_handle.flush()
file_handle.close()
md5 = compute_md5(rand_str)
files[file_handle.name] = HostFile(file_handle, md5)
return files
def make_random_device_files(adb, in_dir, num_files, rand_size=True):
files = {}
min_size = 1 * (1 << 10)
max_size = 16 * (1 << 10)
fixed_size = min_size
for i in range(num_files):
if rand_size:
size = random.randrange(min_size, max_size, 1024)
else:
size = fixed_size
base_name = "device_tmpfile" + str(i)
full_path = in_dir + "/" + base_name
adb.shell("dd if=/dev/urandom of={} bs={} count=1".format(full_path,
size))
dev_md5, _ = adb.shell("md5sum {}".format(full_path)).split()
files[full_path] = DeviceFile(dev_md5, full_path)
return files
class AdbWrapper(object):
"""Convenience wrapper object for the adb command."""
def __init__(self, device=None, out_dir=None):
self.device = device
self.out_dir = out_dir
self.adb_cmd = "adb "
if self.device:
self.adb_cmd += "-s {} ".format(pipes.quote(device))
if self.out_dir:
self.adb_cmd += "-p {} ".format(pipes.quote(out_dir))
def shell(self, cmd):
return call_checked(self.adb_cmd + "shell " + cmd)
def shell_nocheck(self, cmd):
return call_combined(self.adb_cmd + "shell " + cmd)
def install(self, filename):
return call_checked(
self.adb_cmd + "install {}".format(pipes.quote(filename)))
def push(self, local, remote):
return call_checked(self.adb_cmd + "push {} {}".format(
pipes.quote(local), pipes.quote(remote)))
def pull(self, remote, local):
return call_checked(self.adb_cmd + "pull {} {}".format(
pipes.quote(remote), pipes.quote(local)))
def sync(self, directory=""):
return call_checked(self.adb_cmd + "sync {}".format(
pipes.quote(directory) if directory else directory))
def forward(self, local, remote):
return call_checked(self.adb_cmd + "forward {} {}".format(local,
remote))
def tcpip(self, port):
return call_combined(self.adb_cmd + "tcpip {}".format(port))
def usb(self):
return call_checked(self.adb_cmd + "usb")
def root(self):
return call_checked(self.adb_cmd + "root")
def unroot(self):
return call_checked(self.adb_cmd + "unroot")
def forward_remove(self, local):
return call_checked(self.adb_cmd + "forward --remove {}".format(local))
def forward_remove_all(self):
return call_checked(self.adb_cmd + "forward --remove-all")
def connect(self, host):
return call_checked(self.adb_cmd + "connect {}".format(host))
def disconnect(self, host):
return call_checked(self.adb_cmd + "disconnect {}".format(host))
def reverse(self, remote, local):
return call_checked(self.adb_cmd + "reverse {} {}".format(remote,
local))
def reverse_remove_all(self):
return call_checked(self.adb_cmd + "reverse --remove-all")
def reverse_remove(self, remote):
return call_checked(
self.adb_cmd + "reverse --remove {}".format(remote))
def wait(self):
return call_checked(self.adb_cmd + "wait-for-device")
class AdbBasic(unittest.TestCase):
def test_shell(self):
"""Check that we can at least cat a file."""
adb = AdbWrapper()
out = adb.shell("cat /proc/uptime")
self.assertEqual(len(out.split()), 2)
self.assertGreater(float(out.split()[0]), 0.0)
self.assertGreater(float(out.split()[1]), 0.0)
def test_help(self):
"""Make sure we get _something_ out of help."""
out = call_checked("adb help")
self.assertTrue(len(out) > 0)
def test_version(self):
"""Get a version number out of the output of adb."""
out = call_checked("adb version").split()
version_num = False
for item in out:
if re.match(r"[\d+\.]*\d", item):
version_num = True
self.assertTrue(version_num)
def _test_root(self):
adb = AdbWrapper()
if "adbd cannot run as root in production builds" in adb.root():
return
adb.wait()
self.assertEqual("root", adb.shell("id -un").strip())
def _test_unroot(self):
adb = AdbWrapper()
adb.unroot()
adb.wait()
self.assertEqual("shell", adb.shell("id -un").strip())
def test_root_unroot(self):
"""Make sure that adb root and adb unroot work, using id(1)."""
adb = AdbWrapper()
original_user = adb.shell("id -un").strip()
try:
if original_user == "root":
self._test_unroot()
self._test_root()
elif original_user == "shell":
self._test_root()
self._test_unroot()
finally:
if original_user == "root":
adb.root()
else:
adb.unroot()
adb.wait()
def test_argument_escaping(self):
"""Make sure that argument escaping is somewhat sane."""
adb = AdbWrapper()
# http://b/19734868
# Note that this actually matches ssh(1)'s behavior --- it's
# converted to "sh -c echo hello; echo world" which sh interprets
# as "sh -c echo" (with an argument to that shell of "hello"),
# and then "echo world" back in the first shell.
result = adb.shell("sh -c 'echo hello; echo world'").splitlines()
self.assertEqual(["", "world"], result)
# If you really wanted "hello" and "world", here's what you'd do:
result = adb.shell(r"echo hello\;echo world").splitlines()
self.assertEqual(["hello", "world"], result)
# http://b/15479704
self.assertEqual('t', adb.shell("'true && echo t'").strip())
self.assertEqual('t', adb.shell("sh -c 'true && echo t'").strip())
# http://b/20564385
self.assertEqual('t', adb.shell("FOO=a BAR=b echo t").strip())
self.assertEqual('123Linux', adb.shell(r"echo -n 123\;uname").strip())
def test_install_argument_escaping(self):
"""Make sure that install argument escaping works."""
adb = AdbWrapper()
# http://b/20323053
tf = tempfile.NamedTemporaryFile("wb", suffix="-text;ls;1.apk")
self.assertIn("-text;ls;1.apk", adb.install(tf.name))
# http://b/3090932
tf = tempfile.NamedTemporaryFile("wb", suffix="-Live Hold'em.apk")
self.assertIn("-Live Hold'em.apk", adb.install(tf.name))
def test_line_endings(self):
"""Ensure that line ending translation is not happening in the pty.
Bug: http://b/19735063
"""
output = AdbWrapper().shell("uname")
if sys.platform == 'win32':
# adb.exe running on Windows does translation to the Windows \r\n
# convention, so we should expect those chars.
self.assertEqual(output, "Linux\r\n")
else:
self.assertEqual(output, "Linux\n")
def test_tcpip(self):
"""adb tcpip requires a port. http://b/22636927"""
output, status_code = AdbWrapper().tcpip("")
self.assertEqual(1, status_code)
self.assertIn("help message", output)
output, status_code = AdbWrapper().tcpip("blah")
self.assertEqual(1, status_code)
self.assertIn("error", output)
def test_unicode_paths(self):
"""Ensure that we can support non-ASCII paths, even on Windows."""
adb = AdbWrapper()
name = u'로보카 폴리'.encode('utf-8')
# push.
tf = tempfile.NamedTemporaryFile("wb", suffix=name)
adb.push(tf.name, "/data/local/tmp/adb-test-{}".format(name))
# pull.
adb.shell("rm \"'/data/local/tmp/adb-test-*'\"".format(name))
adb.shell("touch \"'/data/local/tmp/adb-test-{}'\"".format(name))
tf = tempfile.NamedTemporaryFile("wb", suffix=name)
adb.pull("/data/local/tmp/adb-test-{}".format(name), tf.name)
class AdbFile(unittest.TestCase):
SCRATCH_DIR = "/data/local/tmp"
DEVICE_TEMP_FILE = SCRATCH_DIR + "/adb_test_file"
DEVICE_TEMP_DIR = SCRATCH_DIR + "/adb_test_dir"
def test_push(self):
"""Push a randomly generated file to specified device."""
kbytes = 512
adb = AdbWrapper()
with tempfile.NamedTemporaryFile(mode="wb", delete=False) as tmp:
try:
rand_str = os.urandom(1024 * kbytes)
tmp.write(rand_str)
tmp.flush()
tmp.close()
host_md5 = compute_md5(rand_str)
adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_FILE))
try:
adb.push(local=tmp.name, remote=AdbFile.DEVICE_TEMP_FILE)
dev_md5, _ = adb.shell(
"md5sum {}".format(AdbFile.DEVICE_TEMP_FILE)).split()
self.assertEqual(host_md5, dev_md5)
finally:
adb.shell_nocheck("rm {}".format(AdbFile.DEVICE_TEMP_FILE))
finally:
os.remove(tmp.name)
# TODO: write push directory test.
def test_pull(self):
"""Pull a randomly generated file from specified device."""
kbytes = 512
adb = AdbWrapper()
adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_FILE))
try:
adb.shell("dd if=/dev/urandom of={} bs=1024 count={}".format(
AdbFile.DEVICE_TEMP_FILE, kbytes))
dev_md5, _ = adb.shell(
"md5sum {}".format(AdbFile.DEVICE_TEMP_FILE)).split()
with tempfile.NamedTemporaryFile(mode="wb", delete=False) \
as tmp_write:
try:
tmp_write.close()
adb.pull(remote=AdbFile.DEVICE_TEMP_FILE,
local=tmp_write.name)
with open(tmp_write.name, "rb") as tmp_read:
host_contents = tmp_read.read()
host_md5 = compute_md5(host_contents)
self.assertEqual(dev_md5, host_md5)
finally:
os.remove(tmp_write.name)
finally:
adb.shell_nocheck("rm {}".format(AdbFile.DEVICE_TEMP_FILE))
def test_pull_dir(self):
"""Pull a randomly generated directory of files from the device."""
adb = AdbWrapper()
temp_files = {}
host_dir = None
try:
# create temporary host directory
host_dir = tempfile.mkdtemp()
# create temporary dir on device
adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))
adb.shell("mkdir -p {}".format(AdbFile.DEVICE_TEMP_DIR))
# populate device dir with random files
temp_files = make_random_device_files(
adb, in_dir=AdbFile.DEVICE_TEMP_DIR, num_files=32)
adb.pull(remote=AdbFile.DEVICE_TEMP_DIR, local=host_dir)
for device_full_path in temp_files:
host_path = os.path.join(
host_dir, temp_files[device_full_path].base_name)
with open(host_path, "rb") as host_file:
host_md5 = compute_md5(host_file.read())
self.assertEqual(host_md5,
temp_files[device_full_path].md5)
finally:
for dev_file in temp_files.values():
host_path = os.path.join(host_dir, dev_file.base_name)
os.remove(host_path)
adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))
if host_dir:
os.removedirs(host_dir)
def test_sync(self):
"""Sync a randomly generated directory of files to specified device."""
try:
adb = AdbWrapper()
temp_files = {}
# create temporary host directory
base_dir = tempfile.mkdtemp()
# create mirror device directory hierarchy within base_dir
full_dir_path = base_dir + AdbFile.DEVICE_TEMP_DIR
os.makedirs(full_dir_path)
# create 32 random files within the host mirror
temp_files = make_random_host_files(in_dir=full_dir_path,
num_files=32)
# clean up any trash on the device
adb = AdbWrapper(out_dir=base_dir)
adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))
# issue the sync
adb.sync("data")
# confirm that every file on the device mirrors that on the host
for host_full_path in temp_files.keys():
device_full_path = posixpath.join(
AdbFile.DEVICE_TEMP_DIR,
temp_files[host_full_path].base_name)
dev_md5, _ = adb.shell(
"md5sum {}".format(device_full_path)).split()
self.assertEqual(temp_files[host_full_path].md5, dev_md5)
finally:
adb.shell_nocheck("rm -r {}".format(AdbFile.DEVICE_TEMP_DIR))
if temp_files:
for tf in temp_files.values():
os.remove(tf.full_path)
if base_dir:
os.removedirs(base_dir + AdbFile.DEVICE_TEMP_DIR)
if __name__ == '__main__':
random.seed(0)
dev_count = get_attached_device_count()
if dev_count:
suite = unittest.TestLoader().loadTestsFromName(__name__)
unittest.TextTestRunner(verbosity=3).run(suite)
else:
print "Test suite must be run with attached devices"