blob: 38849eca259b7fbaa06da0963bcf139147eafe78 [file] [log] [blame]
#!/usr/bin/env python
# Copyright 2014 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""
Unit tests for the contents of device_utils.py (mostly DeviceUtils).
"""
# pylint: disable=protected-access
# pylint: disable=unused-argument
import logging
import unittest
from devil import devil_env
from devil.android import device_errors
from devil.android import device_signal
from devil.android import device_utils
from devil.android.sdk import adb_wrapper
from devil.android.sdk import intent
from devil.android.sdk import version_codes
from devil.utils import cmd_helper
from devil.utils import mock_calls
with devil_env.SysPath(devil_env.PYMOCK_PATH):
import mock # pylint: disable=import-error
class _MockApkHelper(object):
def __init__(self, path, package_name, perms=None):
self.path = path
self.package_name = package_name
self.perms = perms
def GetPackageName(self):
return self.package_name
def GetPermissions(self):
return self.perms
class DeviceUtilsInitTest(unittest.TestCase):
def testInitWithStr(self):
serial_as_str = str('0123456789abcdef')
d = device_utils.DeviceUtils('0123456789abcdef')
self.assertEqual(serial_as_str, d.adb.GetDeviceSerial())
def testInitWithUnicode(self):
serial_as_unicode = unicode('fedcba9876543210')
d = device_utils.DeviceUtils(serial_as_unicode)
self.assertEqual(serial_as_unicode, d.adb.GetDeviceSerial())
def testInitWithAdbWrapper(self):
serial = '123456789abcdef0'
a = adb_wrapper.AdbWrapper(serial)
d = device_utils.DeviceUtils(a)
self.assertEqual(serial, d.adb.GetDeviceSerial())
def testInitWithMissing_fails(self):
with self.assertRaises(ValueError):
device_utils.DeviceUtils(None)
with self.assertRaises(ValueError):
device_utils.DeviceUtils('')
class DeviceUtilsGetAVDsTest(mock_calls.TestCase):
def testGetAVDs(self):
mocked_attrs = {
'android_sdk': '/my/sdk/path'
}
with mock.patch('devil.devil_env._Environment.LocalPath',
mock.Mock(side_effect=lambda a: mocked_attrs[a])):
with self.assertCall(
mock.call.devil.utils.cmd_helper.GetCmdOutput(
[mock.ANY, 'list', 'avd']),
'Available Android Virtual Devices:\n'
' Name: my_android5.0\n'
' Path: /some/path/to/.android/avd/my_android5.0.avd\n'
' Target: Android 5.0 (API level 21)\n'
' Tag/ABI: default/x86\n'
' Skin: WVGA800\n'):
self.assertEquals(['my_android5.0'], device_utils.GetAVDs())
class DeviceUtilsRestartServerTest(mock_calls.TestCase):
@mock.patch('time.sleep', mock.Mock())
def testRestartServer_succeeds(self):
with self.assertCalls(
mock.call.devil.android.sdk.adb_wrapper.AdbWrapper.KillServer(),
(mock.call.devil.utils.cmd_helper.GetCmdStatusAndOutput(
['pgrep', 'adb']),
(1, '')),
mock.call.devil.android.sdk.adb_wrapper.AdbWrapper.StartServer(),
(mock.call.devil.utils.cmd_helper.GetCmdStatusAndOutput(
['pgrep', 'adb']),
(1, '')),
(mock.call.devil.utils.cmd_helper.GetCmdStatusAndOutput(
['pgrep', 'adb']),
(0, '123\n'))):
device_utils.RestartServer()
class MockTempFile(object):
def __init__(self, name='/tmp/some/file'):
self.file = mock.MagicMock(spec=file)
self.file.name = name
self.file.name_quoted = cmd_helper.SingleQuote(name)
def __enter__(self):
return self.file
def __exit__(self, exc_type, exc_val, exc_tb):
pass
@property
def name(self):
return self.file.name
class _PatchedFunction(object):
def __init__(self, patched=None, mocked=None):
self.patched = patched
self.mocked = mocked
def _AdbWrapperMock(test_serial, is_ready=True):
adb = mock.Mock(spec=adb_wrapper.AdbWrapper)
adb.__str__ = mock.Mock(return_value=test_serial)
adb.GetDeviceSerial.return_value = test_serial
adb.is_ready = is_ready
return adb
class DeviceUtilsTest(mock_calls.TestCase):
def setUp(self):
self.adb = _AdbWrapperMock('0123456789abcdef')
self.device = device_utils.DeviceUtils(
self.adb, default_timeout=10, default_retries=0)
self.watchMethodCalls(self.call.adb, ignore=['GetDeviceSerial'])
def AdbCommandError(self, args=None, output=None, status=None, msg=None):
if args is None:
args = ['[unspecified]']
return mock.Mock(side_effect=device_errors.AdbCommandFailedError(
args, output, status, msg, str(self.device)))
def CommandError(self, msg=None):
if msg is None:
msg = 'Command failed'
return mock.Mock(side_effect=device_errors.CommandFailedError(
msg, str(self.device)))
def ShellError(self, output=None, status=1):
def action(cmd, *args, **kwargs):
raise device_errors.AdbShellCommandFailedError(
cmd, output, status, str(self.device))
if output is None:
output = 'Permission denied\n'
return action
def TimeoutError(self, msg=None):
if msg is None:
msg = 'Operation timed out'
return mock.Mock(side_effect=device_errors.CommandTimeoutError(
msg, str(self.device)))
class DeviceUtilsEqTest(DeviceUtilsTest):
def testEq_equal_deviceUtils(self):
other = device_utils.DeviceUtils(_AdbWrapperMock('0123456789abcdef'))
self.assertTrue(self.device == other)
self.assertTrue(other == self.device)
def testEq_equal_adbWrapper(self):
other = adb_wrapper.AdbWrapper('0123456789abcdef')
self.assertTrue(self.device == other)
self.assertTrue(other == self.device)
def testEq_equal_string(self):
other = '0123456789abcdef'
self.assertTrue(self.device == other)
self.assertTrue(other == self.device)
def testEq_devicesNotEqual(self):
other = device_utils.DeviceUtils(_AdbWrapperMock('0123456789abcdee'))
self.assertFalse(self.device == other)
self.assertFalse(other == self.device)
def testEq_identity(self):
self.assertTrue(self.device == self.device)
def testEq_serialInList(self):
devices = [self.device]
self.assertTrue('0123456789abcdef' in devices)
class DeviceUtilsLtTest(DeviceUtilsTest):
def testLt_lessThan(self):
other = device_utils.DeviceUtils(_AdbWrapperMock('ffffffffffffffff'))
self.assertTrue(self.device < other)
self.assertTrue(other > self.device)
def testLt_greaterThan_lhs(self):
other = device_utils.DeviceUtils(_AdbWrapperMock('0000000000000000'))
self.assertFalse(self.device < other)
self.assertFalse(other > self.device)
def testLt_equal(self):
other = device_utils.DeviceUtils(_AdbWrapperMock('0123456789abcdef'))
self.assertFalse(self.device < other)
self.assertFalse(other > self.device)
def testLt_sorted(self):
devices = [
device_utils.DeviceUtils(_AdbWrapperMock('ffffffffffffffff')),
device_utils.DeviceUtils(_AdbWrapperMock('0000000000000000')),
]
sorted_devices = sorted(devices)
self.assertEquals('0000000000000000',
sorted_devices[0].adb.GetDeviceSerial())
self.assertEquals('ffffffffffffffff',
sorted_devices[1].adb.GetDeviceSerial())
class DeviceUtilsStrTest(DeviceUtilsTest):
def testStr_returnsSerial(self):
with self.assertCalls(
(self.call.adb.GetDeviceSerial(), '0123456789abcdef')):
self.assertEqual('0123456789abcdef', str(self.device))
class DeviceUtilsIsOnlineTest(DeviceUtilsTest):
def testIsOnline_true(self):
with self.assertCall(self.call.adb.GetState(), 'device'):
self.assertTrue(self.device.IsOnline())
def testIsOnline_false(self):
with self.assertCall(self.call.adb.GetState(), 'offline'):
self.assertFalse(self.device.IsOnline())
def testIsOnline_error(self):
with self.assertCall(self.call.adb.GetState(), self.CommandError()):
self.assertFalse(self.device.IsOnline())
class DeviceUtilsHasRootTest(DeviceUtilsTest):
def testHasRoot_true(self):
with self.assertCall(self.call.adb.Shell('ls /root'), 'foo\n'):
self.assertTrue(self.device.HasRoot())
def testHasRoot_false(self):
with self.assertCall(self.call.adb.Shell('ls /root'), self.ShellError()):
self.assertFalse(self.device.HasRoot())
class DeviceUtilsEnableRootTest(DeviceUtilsTest):
def testEnableRoot_succeeds(self):
with self.assertCalls(
(self.call.device.IsUserBuild(), False),
self.call.adb.Root(),
self.call.device.WaitUntilFullyBooted()):
self.device.EnableRoot()
def testEnableRoot_userBuild(self):
with self.assertCalls(
(self.call.device.IsUserBuild(), True)):
with self.assertRaises(device_errors.CommandFailedError):
self.device.EnableRoot()
def testEnableRoot_rootFails(self):
with self.assertCalls(
(self.call.device.IsUserBuild(), False),
(self.call.adb.Root(), self.CommandError())):
with self.assertRaises(device_errors.CommandFailedError):
self.device.EnableRoot()
class DeviceUtilsIsUserBuildTest(DeviceUtilsTest):
def testIsUserBuild_yes(self):
with self.assertCall(
self.call.device.GetProp('ro.build.type', cache=True), 'user'):
self.assertTrue(self.device.IsUserBuild())
def testIsUserBuild_no(self):
with self.assertCall(
self.call.device.GetProp('ro.build.type', cache=True), 'userdebug'):
self.assertFalse(self.device.IsUserBuild())
class DeviceUtilsGetExternalStoragePathTest(DeviceUtilsTest):
def testGetExternalStoragePath_succeeds(self):
with self.assertCall(
self.call.adb.Shell('echo $EXTERNAL_STORAGE'), '/fake/storage/path\n'):
self.assertEquals('/fake/storage/path',
self.device.GetExternalStoragePath())
def testGetExternalStoragePath_fails(self):
with self.assertCall(self.call.adb.Shell('echo $EXTERNAL_STORAGE'), '\n'):
with self.assertRaises(device_errors.CommandFailedError):
self.device.GetExternalStoragePath()
class DeviceUtilsGetApplicationPathsInternalTest(DeviceUtilsTest):
def testGetApplicationPathsInternal_exists(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '19'),
(self.call.device.RunShellCommand(
['pm', 'path', 'android'], check_return=True),
['package:/path/to/android.apk'])):
self.assertEquals(['/path/to/android.apk'],
self.device._GetApplicationPathsInternal('android'))
def testGetApplicationPathsInternal_notExists(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '19'),
(self.call.device.RunShellCommand(
['pm', 'path', 'not.installed.app'], check_return=True),
'')):
self.assertEquals([],
self.device._GetApplicationPathsInternal('not.installed.app'))
def testGetApplicationPathsInternal_fails(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '19'),
(self.call.device.RunShellCommand(
['pm', 'path', 'android'], check_return=True),
self.CommandError('ERROR. Is package manager running?\n'))):
with self.assertRaises(device_errors.CommandFailedError):
self.device._GetApplicationPathsInternal('android')
class DeviceUtils_GetApplicationVersionTest(DeviceUtilsTest):
def test_GetApplicationVersion_exists(self):
with self.assertCalls(
(self.call.adb.Shell('dumpsys package com.android.chrome'),
'Packages:\n'
' Package [com.android.chrome] (3901ecfb):\n'
' userId=1234 gids=[123, 456, 789]\n'
' pkg=Package{1fecf634 com.android.chrome}\n'
' versionName=45.0.1234.7\n')):
self.assertEquals('45.0.1234.7',
self.device.GetApplicationVersion('com.android.chrome'))
def test_GetApplicationVersion_notExists(self):
with self.assertCalls(
(self.call.adb.Shell('dumpsys package com.android.chrome'), '')):
self.assertEquals(None,
self.device.GetApplicationVersion('com.android.chrome'))
def test_GetApplicationVersion_fails(self):
with self.assertCalls(
(self.call.adb.Shell('dumpsys package com.android.chrome'),
'Packages:\n'
' Package [com.android.chrome] (3901ecfb):\n'
' userId=1234 gids=[123, 456, 789]\n'
' pkg=Package{1fecf634 com.android.chrome}\n')):
with self.assertRaises(device_errors.CommandFailedError):
self.device.GetApplicationVersion('com.android.chrome')
class DeviceUtilsGetApplicationDataDirectoryTest(DeviceUtilsTest):
def testGetApplicationDataDirectory_exists(self):
with self.assertCall(
self.call.device._RunPipedShellCommand(
'pm dump foo.bar.baz | grep dataDir='),
['dataDir=/data/data/foo.bar.baz']):
self.assertEquals(
'/data/data/foo.bar.baz',
self.device.GetApplicationDataDirectory('foo.bar.baz'))
def testGetApplicationDataDirectory_notExists(self):
with self.assertCall(
self.call.device._RunPipedShellCommand(
'pm dump foo.bar.baz | grep dataDir='),
self.ShellError()):
self.assertIsNone(self.device.GetApplicationDataDirectory('foo.bar.baz'))
@mock.patch('time.sleep', mock.Mock())
class DeviceUtilsWaitUntilFullyBootedTest(DeviceUtilsTest):
def testWaitUntilFullyBooted_succeedsNoWifi(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), '/fake/storage/path'),
(self.call.adb.Shell('test -d /fake/storage/path'), ''),
# pm_ready
(self.call.device._GetApplicationPathsInternal('android',
skip_cache=True),
['package:/some/fake/path']),
# boot_completed
(self.call.device.GetProp('sys.boot_completed', cache=False), '1')):
self.device.WaitUntilFullyBooted(wifi=False)
def testWaitUntilFullyBooted_succeedsWithWifi(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), '/fake/storage/path'),
(self.call.adb.Shell('test -d /fake/storage/path'), ''),
# pm_ready
(self.call.device._GetApplicationPathsInternal('android',
skip_cache=True),
['package:/some/fake/path']),
# boot_completed
(self.call.device.GetProp('sys.boot_completed', cache=False), '1'),
# wifi_enabled
(self.call.adb.Shell('dumpsys wifi'),
'stuff\nWi-Fi is enabled\nmore stuff\n')):
self.device.WaitUntilFullyBooted(wifi=True)
def testWaitUntilFullyBooted_deviceNotInitiallyAvailable(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), self.AdbCommandError()),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), self.AdbCommandError()),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), self.AdbCommandError()),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), self.AdbCommandError()),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), '/fake/storage/path'),
(self.call.adb.Shell('test -d /fake/storage/path'), ''),
# pm_ready
(self.call.device._GetApplicationPathsInternal('android',
skip_cache=True),
['package:/some/fake/path']),
# boot_completed
(self.call.device.GetProp('sys.boot_completed', cache=False), '1')):
self.device.WaitUntilFullyBooted(wifi=False)
def testWaitUntilFullyBooted_sdCardReadyFails_noPath(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), self.CommandError())):
with self.assertRaises(device_errors.CommandFailedError):
self.device.WaitUntilFullyBooted(wifi=False)
def testWaitUntilFullyBooted_sdCardReadyFails_notExists(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), '/fake/storage/path'),
(self.call.adb.Shell('test -d /fake/storage/path'), self.ShellError()),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), '/fake/storage/path'),
(self.call.adb.Shell('test -d /fake/storage/path'), self.ShellError()),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), '/fake/storage/path'),
(self.call.adb.Shell('test -d /fake/storage/path'),
self.TimeoutError())):
with self.assertRaises(device_errors.CommandTimeoutError):
self.device.WaitUntilFullyBooted(wifi=False)
def testWaitUntilFullyBooted_devicePmFails(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), '/fake/storage/path'),
(self.call.adb.Shell('test -d /fake/storage/path'), ''),
# pm_ready
(self.call.device._GetApplicationPathsInternal('android',
skip_cache=True),
self.CommandError()),
# pm_ready
(self.call.device._GetApplicationPathsInternal('android',
skip_cache=True),
self.CommandError()),
# pm_ready
(self.call.device._GetApplicationPathsInternal('android',
skip_cache=True),
self.TimeoutError())):
with self.assertRaises(device_errors.CommandTimeoutError):
self.device.WaitUntilFullyBooted(wifi=False)
def testWaitUntilFullyBooted_bootFails(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), '/fake/storage/path'),
(self.call.adb.Shell('test -d /fake/storage/path'), ''),
# pm_ready
(self.call.device._GetApplicationPathsInternal('android',
skip_cache=True),
['package:/some/fake/path']),
# boot_completed
(self.call.device.GetProp('sys.boot_completed', cache=False), '0'),
# boot_completed
(self.call.device.GetProp('sys.boot_completed', cache=False), '0'),
# boot_completed
(self.call.device.GetProp('sys.boot_completed', cache=False),
self.TimeoutError())):
with self.assertRaises(device_errors.CommandTimeoutError):
self.device.WaitUntilFullyBooted(wifi=False)
def testWaitUntilFullyBooted_wifiFails(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), '/fake/storage/path'),
(self.call.adb.Shell('test -d /fake/storage/path'), ''),
# pm_ready
(self.call.device._GetApplicationPathsInternal('android',
skip_cache=True),
['package:/some/fake/path']),
# boot_completed
(self.call.device.GetProp('sys.boot_completed', cache=False), '1'),
# wifi_enabled
(self.call.adb.Shell('dumpsys wifi'), 'stuff\nmore stuff\n'),
# wifi_enabled
(self.call.adb.Shell('dumpsys wifi'), 'stuff\nmore stuff\n'),
# wifi_enabled
(self.call.adb.Shell('dumpsys wifi'), self.TimeoutError())):
with self.assertRaises(device_errors.CommandTimeoutError):
self.device.WaitUntilFullyBooted(wifi=True)
@mock.patch('time.sleep', mock.Mock())
class DeviceUtilsRebootTest(DeviceUtilsTest):
def testReboot_nonBlocking(self):
with self.assertCalls(
self.call.adb.Reboot(),
(self.call.device.IsOnline(), True),
(self.call.device.IsOnline(), False)):
self.device.Reboot(block=False)
def testReboot_blocking(self):
with self.assertCalls(
self.call.adb.Reboot(),
(self.call.device.IsOnline(), True),
(self.call.device.IsOnline(), False),
self.call.device.WaitUntilFullyBooted(wifi=False)):
self.device.Reboot(block=True)
def testReboot_blockUntilWifi(self):
with self.assertCalls(
self.call.adb.Reboot(),
(self.call.device.IsOnline(), True),
(self.call.device.IsOnline(), False),
self.call.device.WaitUntilFullyBooted(wifi=True)):
self.device.Reboot(block=True, wifi=True)
class DeviceUtilsInstallTest(DeviceUtilsTest):
mock_apk = _MockApkHelper('/fake/test/app.apk', 'test.package', ['p1'])
def testInstall_noPriorInstall(self):
with self.patch_call(self.call.device.build_version_sdk, return_value=23):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'), []),
self.call.adb.Install('/fake/test/app.apk', reinstall=False,
allow_downgrade=False),
(self.call.device.GrantPermissions('test.package', ['p1']), [])):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0)
def testInstall_permissionsPreM(self):
with self.patch_call(self.call.device.build_version_sdk, return_value=20):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'), []),
(self.call.adb.Install('/fake/test/app.apk', reinstall=False,
allow_downgrade=False))):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0)
def testInstall_findPermissions(self):
with self.patch_call(self.call.device.build_version_sdk, return_value=23):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'), []),
(self.call.adb.Install('/fake/test/app.apk', reinstall=False,
allow_downgrade=False)),
(self.call.device.GrantPermissions('test.package', ['p1']), [])):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0)
def testInstall_passPermissions(self):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'), []),
(self.call.adb.Install('/fake/test/app.apk', reinstall=False,
allow_downgrade=False)),
(self.call.device.GrantPermissions('test.package', ['p1', 'p2']), [])):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0,
permissions=['p1', 'p2'])
def testInstall_differentPriorInstall(self):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'),
['/fake/data/app/test.package.apk']),
(self.call.device._ComputeStaleApks('test.package',
['/fake/test/app.apk']),
(['/fake/test/app.apk'], None)),
self.call.device.Uninstall('test.package'),
self.call.adb.Install('/fake/test/app.apk', reinstall=False,
allow_downgrade=False)):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0,
permissions=[])
def testInstall_differentPriorInstall_reinstall(self):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'),
['/fake/data/app/test.package.apk']),
(self.call.device._ComputeStaleApks('test.package',
['/fake/test/app.apk']),
(['/fake/test/app.apk'], None)),
self.call.adb.Install('/fake/test/app.apk', reinstall=True,
allow_downgrade=False)):
self.device.Install(DeviceUtilsInstallTest.mock_apk,
reinstall=True, retries=0, permissions=[])
def testInstall_identicalPriorInstall_reinstall(self):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'),
['/fake/data/app/test.package.apk']),
(self.call.device._ComputeStaleApks('test.package',
['/fake/test/app.apk']),
([], None)),
(self.call.device.ForceStop('test.package'))):
self.device.Install(DeviceUtilsInstallTest.mock_apk,
reinstall=True, retries=0, permissions=[])
def testInstall_fails(self):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'), []),
(self.call.adb.Install('/fake/test/app.apk', reinstall=False,
allow_downgrade=False),
self.CommandError('Failure\r\n'))):
with self.assertRaises(device_errors.CommandFailedError):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0)
def testInstall_downgrade(self):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'),
['/fake/data/app/test.package.apk']),
(self.call.device._ComputeStaleApks('test.package',
['/fake/test/app.apk']),
(['/fake/test/app.apk'], None)),
self.call.adb.Install('/fake/test/app.apk', reinstall=True,
allow_downgrade=True)):
self.device.Install(DeviceUtilsInstallTest.mock_apk,
reinstall=True, retries=0, permissions=[], allow_downgrade=True)
class DeviceUtilsInstallSplitApkTest(DeviceUtilsTest):
mock_apk = _MockApkHelper('base.apk', 'test.package', ['p1'])
def testInstallSplitApk_noPriorInstall(self):
with self.assertCalls(
(self.call.device._CheckSdkLevel(21)),
(mock.call.devil.android.sdk.split_select.SelectSplits(
self.device, 'base.apk',
['split1.apk', 'split2.apk', 'split3.apk'],
allow_cached_props=False),
['split2.apk']),
(self.call.device._GetApplicationPathsInternal('test.package'), []),
(self.call.adb.InstallMultiple(
['base.apk', 'split2.apk'], partial=None, reinstall=False,
allow_downgrade=False))):
self.device.InstallSplitApk(DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk', 'split3.apk'], permissions=[], retries=0)
def testInstallSplitApk_partialInstall(self):
with self.assertCalls(
(self.call.device._CheckSdkLevel(21)),
(mock.call.devil.android.sdk.split_select.SelectSplits(
self.device, 'base.apk',
['split1.apk', 'split2.apk', 'split3.apk'],
allow_cached_props=False),
['split2.apk']),
(self.call.device._GetApplicationPathsInternal('test.package'),
['base-on-device.apk', 'split2-on-device.apk']),
(self.call.device._ComputeStaleApks('test.package',
['base.apk', 'split2.apk']),
(['split2.apk'], None)),
(self.call.adb.InstallMultiple(
['split2.apk'], partial='test.package', reinstall=True,
allow_downgrade=False))):
self.device.InstallSplitApk(DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk', 'split3.apk'],
reinstall=True, permissions=[], retries=0)
def testInstallSplitApk_downgrade(self):
with self.assertCalls(
(self.call.device._CheckSdkLevel(21)),
(mock.call.devil.android.sdk.split_select.SelectSplits(
self.device, 'base.apk',
['split1.apk', 'split2.apk', 'split3.apk'],
allow_cached_props=False),
['split2.apk']),
(self.call.device._GetApplicationPathsInternal('test.package'),
['base-on-device.apk', 'split2-on-device.apk']),
(self.call.device._ComputeStaleApks('test.package',
['base.apk', 'split2.apk']),
(['split2.apk'], None)),
(self.call.adb.InstallMultiple(
['split2.apk'], partial='test.package', reinstall=True,
allow_downgrade=True))):
self.device.InstallSplitApk(DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk', 'split3.apk'],
reinstall=True, permissions=[], retries=0,
allow_downgrade=True)
class DeviceUtilsUninstallTest(DeviceUtilsTest):
def testUninstall_callsThrough(self):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'),
['/path.apk']),
self.call.adb.Uninstall('test.package', True)):
self.device.Uninstall('test.package', True)
def testUninstall_noop(self):
with self.assertCalls(
(self.call.device._GetApplicationPathsInternal('test.package'), [])):
self.device.Uninstall('test.package', True)
class DeviceUtilsSuTest(DeviceUtilsTest):
def testSu_preM(self):
with self.patch_call(
self.call.device.build_version_sdk,
return_value=version_codes.LOLLIPOP_MR1):
self.assertEquals('su -c foo', self.device._Su('foo'))
def testSu_mAndAbove(self):
with self.patch_call(
self.call.device.build_version_sdk,
return_value=version_codes.MARSHMALLOW):
self.assertEquals('su 0 foo', self.device._Su('foo'))
class DeviceUtilsRunShellCommandTest(DeviceUtilsTest):
def setUp(self):
super(DeviceUtilsRunShellCommandTest, self).setUp()
self.device.NeedsSU = mock.Mock(return_value=False)
def testRunShellCommand_commandAsList(self):
with self.assertCall(self.call.adb.Shell('pm list packages'), ''):
self.device.RunShellCommand(['pm', 'list', 'packages'])
def testRunShellCommand_commandAsListQuoted(self):
with self.assertCall(self.call.adb.Shell("echo 'hello world' '$10'"), ''):
self.device.RunShellCommand(['echo', 'hello world', '$10'])
def testRunShellCommand_commandAsString(self):
with self.assertCall(self.call.adb.Shell('echo "$VAR"'), ''):
self.device.RunShellCommand('echo "$VAR"')
def testNewRunShellImpl_withEnv(self):
with self.assertCall(
self.call.adb.Shell('VAR=some_string echo "$VAR"'), ''):
self.device.RunShellCommand('echo "$VAR"', env={'VAR': 'some_string'})
def testNewRunShellImpl_withEnvQuoted(self):
with self.assertCall(
self.call.adb.Shell('PATH="$PATH:/other/path" run_this'), ''):
self.device.RunShellCommand('run_this', env={'PATH': '$PATH:/other/path'})
def testNewRunShellImpl_withEnv_failure(self):
with self.assertRaises(KeyError):
self.device.RunShellCommand('some_cmd', env={'INVALID NAME': 'value'})
def testNewRunShellImpl_withCwd(self):
with self.assertCall(self.call.adb.Shell('cd /some/test/path && ls'), ''):
self.device.RunShellCommand('ls', cwd='/some/test/path')
def testNewRunShellImpl_withCwdQuoted(self):
with self.assertCall(
self.call.adb.Shell("cd '/some test/path with/spaces' && ls"), ''):
self.device.RunShellCommand('ls', cwd='/some test/path with/spaces')
def testRunShellCommand_withHugeCmd(self):
payload = 'hi! ' * 1024
expected_cmd = "echo '%s'" % payload
with self.assertCalls(
(mock.call.devil.android.device_temp_file.DeviceTempFile(
self.adb, suffix='.sh'), MockTempFile('/sdcard/temp-123.sh')),
self.call.device._WriteFileWithPush('/sdcard/temp-123.sh', expected_cmd),
(self.call.adb.Shell('sh /sdcard/temp-123.sh'), payload + '\n')):
self.assertEquals([payload],
self.device.RunShellCommand(['echo', payload]))
def testRunShellCommand_withHugeCmdAndSU(self):
payload = 'hi! ' * 1024
expected_cmd_without_su = """sh -c 'echo '"'"'%s'"'"''""" % payload
expected_cmd = 'su -c %s' % expected_cmd_without_su
with self.assertCalls(
(self.call.device.NeedsSU(), True),
(self.call.device._Su(expected_cmd_without_su), expected_cmd),
(mock.call.devil.android.device_temp_file.DeviceTempFile(
self.adb, suffix='.sh'), MockTempFile('/sdcard/temp-123.sh')),
self.call.device._WriteFileWithPush('/sdcard/temp-123.sh', expected_cmd),
(self.call.adb.Shell('sh /sdcard/temp-123.sh'), payload + '\n')):
self.assertEquals(
[payload],
self.device.RunShellCommand(['echo', payload], as_root=True))
def testRunShellCommand_withSu(self):
expected_cmd_without_su = "sh -c 'setprop service.adb.root 0'"
expected_cmd = 'su -c %s' % expected_cmd_without_su
with self.assertCalls(
(self.call.device.NeedsSU(), True),
(self.call.device._Su(expected_cmd_without_su), expected_cmd),
(self.call.adb.Shell(expected_cmd), '')):
self.device.RunShellCommand('setprop service.adb.root 0', as_root=True)
def testRunShellCommand_manyLines(self):
cmd = 'ls /some/path'
with self.assertCall(self.call.adb.Shell(cmd), 'file1\nfile2\nfile3\n'):
self.assertEquals(['file1', 'file2', 'file3'],
self.device.RunShellCommand(cmd))
def testRunShellCommand_singleLine_success(self):
cmd = 'echo $VALUE'
with self.assertCall(self.call.adb.Shell(cmd), 'some value\n'):
self.assertEquals('some value',
self.device.RunShellCommand(cmd, single_line=True))
def testRunShellCommand_singleLine_successEmptyLine(self):
cmd = 'echo $VALUE'
with self.assertCall(self.call.adb.Shell(cmd), '\n'):
self.assertEquals('',
self.device.RunShellCommand(cmd, single_line=True))
def testRunShellCommand_singleLine_successWithoutEndLine(self):
cmd = 'echo -n $VALUE'
with self.assertCall(self.call.adb.Shell(cmd), 'some value'):
self.assertEquals('some value',
self.device.RunShellCommand(cmd, single_line=True))
def testRunShellCommand_singleLine_successNoOutput(self):
cmd = 'echo -n $VALUE'
with self.assertCall(self.call.adb.Shell(cmd), ''):
self.assertEquals('',
self.device.RunShellCommand(cmd, single_line=True))
def testRunShellCommand_singleLine_failTooManyLines(self):
cmd = 'echo $VALUE'
with self.assertCall(self.call.adb.Shell(cmd),
'some value\nanother value\n'):
with self.assertRaises(device_errors.CommandFailedError):
self.device.RunShellCommand(cmd, single_line=True)
def testRunShellCommand_checkReturn_success(self):
cmd = 'echo $ANDROID_DATA'
output = '/data\n'
with self.assertCall(self.call.adb.Shell(cmd), output):
self.assertEquals([output.rstrip()],
self.device.RunShellCommand(cmd, check_return=True))
def testRunShellCommand_checkReturn_failure(self):
cmd = 'ls /root'
output = 'opendir failed, Permission denied\n'
with self.assertCall(self.call.adb.Shell(cmd), self.ShellError(output)):
with self.assertRaises(device_errors.AdbCommandFailedError):
self.device.RunShellCommand(cmd, check_return=True)
def testRunShellCommand_checkReturn_disabled(self):
cmd = 'ls /root'
output = 'opendir failed, Permission denied\n'
with self.assertCall(self.call.adb.Shell(cmd), self.ShellError(output)):
self.assertEquals([output.rstrip()],
self.device.RunShellCommand(cmd, check_return=False))
def testRunShellCommand_largeOutput_enabled(self):
cmd = 'echo $VALUE'
temp_file = MockTempFile('/sdcard/temp-123')
cmd_redirect = '( %s )>%s' % (cmd, temp_file.name)
with self.assertCalls(
(mock.call.devil.android.device_temp_file.DeviceTempFile(self.adb),
temp_file),
(self.call.adb.Shell(cmd_redirect)),
(self.call.device.ReadFile(temp_file.name, force_pull=True),
'something')):
self.assertEquals(
['something'],
self.device.RunShellCommand(
cmd, large_output=True, check_return=True))
def testRunShellCommand_largeOutput_disabledNoTrigger(self):
cmd = 'something'
with self.assertCall(self.call.adb.Shell(cmd), self.ShellError('')):
with self.assertRaises(device_errors.AdbCommandFailedError):
self.device.RunShellCommand(cmd, check_return=True)
def testRunShellCommand_largeOutput_disabledTrigger(self):
cmd = 'echo $VALUE'
temp_file = MockTempFile('/sdcard/temp-123')
cmd_redirect = '( %s )>%s' % (cmd, temp_file.name)
with self.assertCalls(
(self.call.adb.Shell(cmd), self.ShellError('', None)),
(mock.call.devil.android.device_temp_file.DeviceTempFile(self.adb),
temp_file),
(self.call.adb.Shell(cmd_redirect)),
(self.call.device.ReadFile(mock.ANY, force_pull=True),
'something')):
self.assertEquals(['something'],
self.device.RunShellCommand(cmd, check_return=True))
class DeviceUtilsRunPipedShellCommandTest(DeviceUtilsTest):
def testRunPipedShellCommand_success(self):
with self.assertCall(
self.call.device.RunShellCommand(
'ps | grep foo; echo "PIPESTATUS: ${PIPESTATUS[@]}"',
check_return=True),
['This line contains foo', 'PIPESTATUS: 0 0']):
self.assertEquals(['This line contains foo'],
self.device._RunPipedShellCommand('ps | grep foo'))
def testRunPipedShellCommand_firstCommandFails(self):
with self.assertCall(
self.call.device.RunShellCommand(
'ps | grep foo; echo "PIPESTATUS: ${PIPESTATUS[@]}"',
check_return=True),
['PIPESTATUS: 1 0']):
with self.assertRaises(device_errors.AdbShellCommandFailedError) as ec:
self.device._RunPipedShellCommand('ps | grep foo')
self.assertEquals([1, 0], ec.exception.status)
def testRunPipedShellCommand_secondCommandFails(self):
with self.assertCall(
self.call.device.RunShellCommand(
'ps | grep foo; echo "PIPESTATUS: ${PIPESTATUS[@]}"',
check_return=True),
['PIPESTATUS: 0 1']):
with self.assertRaises(device_errors.AdbShellCommandFailedError) as ec:
self.device._RunPipedShellCommand('ps | grep foo')
self.assertEquals([0, 1], ec.exception.status)
def testRunPipedShellCommand_outputCutOff(self):
with self.assertCall(
self.call.device.RunShellCommand(
'ps | grep foo; echo "PIPESTATUS: ${PIPESTATUS[@]}"',
check_return=True),
['foo.bar'] * 256 + ['foo.ba']):
with self.assertRaises(device_errors.AdbShellCommandFailedError) as ec:
self.device._RunPipedShellCommand('ps | grep foo')
self.assertIs(None, ec.exception.status)
@mock.patch('time.sleep', mock.Mock())
class DeviceUtilsKillAllTest(DeviceUtilsTest):
def testKillAll_noMatchingProcessesFailure(self):
with self.assertCall(self.call.device.GetPids('test_process'), {}):
with self.assertRaises(device_errors.CommandFailedError):
self.device.KillAll('test_process')
def testKillAll_noMatchingProcessesQuiet(self):
with self.assertCall(self.call.device.GetPids('test_process'), {}):
self.assertEqual(0, self.device.KillAll('test_process', quiet=True))
def testKillAll_nonblocking(self):
with self.assertCalls(
(self.call.device.GetPids('some.process'),
{'some.process': ['1234'], 'some.processing.thing': ['5678']}),
(self.call.adb.Shell('kill -9 1234 5678'), '')):
self.assertEquals(
2, self.device.KillAll('some.process', blocking=False))
def testKillAll_blocking(self):
with self.assertCalls(
(self.call.device.GetPids('some.process'),
{'some.process': ['1234'], 'some.processing.thing': ['5678']}),
(self.call.adb.Shell('kill -9 1234 5678'), ''),
(self.call.device.GetPids('some.process'),
{'some.processing.thing': ['5678']}),
(self.call.device.GetPids('some.process'),
{'some.process': ['1111']})): # Other instance with different pid.
self.assertEquals(
2, self.device.KillAll('some.process', blocking=True))
def testKillAll_exactNonblocking(self):
with self.assertCalls(
(self.call.device.GetPids('some.process'),
{'some.process': ['1234'], 'some.processing.thing': ['5678']}),
(self.call.adb.Shell('kill -9 1234'), '')):
self.assertEquals(
1, self.device.KillAll('some.process', exact=True, blocking=False))
def testKillAll_exactBlocking(self):
with self.assertCalls(
(self.call.device.GetPids('some.process'),
{'some.process': ['1234'], 'some.processing.thing': ['5678']}),
(self.call.adb.Shell('kill -9 1234'), ''),
(self.call.device.GetPids('some.process'),
{'some.process': ['1234'], 'some.processing.thing': ['5678']}),
(self.call.device.GetPids('some.process'),
{'some.processing.thing': ['5678']})):
self.assertEquals(
1, self.device.KillAll('some.process', exact=True, blocking=True))
def testKillAll_root(self):
with self.assertCalls(
(self.call.device.GetPids('some.process'), {'some.process': ['1234']}),
(self.call.device.NeedsSU(), True),
(self.call.device._Su("sh -c 'kill -9 1234'"),
"su -c sh -c 'kill -9 1234'"),
(self.call.adb.Shell("su -c sh -c 'kill -9 1234'"), '')):
self.assertEquals(
1, self.device.KillAll('some.process', as_root=True))
def testKillAll_sigterm(self):
with self.assertCalls(
(self.call.device.GetPids('some.process'),
{'some.process': ['1234']}),
(self.call.adb.Shell('kill -15 1234'), '')):
self.assertEquals(
1, self.device.KillAll('some.process', signum=device_signal.SIGTERM))
def testKillAll_multipleInstances(self):
with self.assertCalls(
(self.call.device.GetPids('some.process'),
{'some.process': ['1234', '4567']}),
(self.call.adb.Shell('kill -15 1234 4567'), '')):
self.assertEquals(
2, self.device.KillAll('some.process', signum=device_signal.SIGTERM))
class DeviceUtilsStartActivityTest(DeviceUtilsTest):
def testStartActivity_actionOnly(self):
test_intent = intent.Intent(action='android.intent.action.VIEW')
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
def testStartActivity_success(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main')
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-n test.package/.Main'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
def testStartActivity_failure(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main')
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-n test.package/.Main'),
'Error: Failed to start test activity'):
with self.assertRaises(device_errors.CommandFailedError):
self.device.StartActivity(test_intent)
def testStartActivity_blocking(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main')
with self.assertCall(
self.call.adb.Shell('am start '
'-W '
'-a android.intent.action.VIEW '
'-n test.package/.Main'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent, blocking=True)
def testStartActivity_withCategory(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main',
category='android.intent.category.HOME')
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-c android.intent.category.HOME '
'-n test.package/.Main'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
def testStartActivity_withMultipleCategories(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main',
category=['android.intent.category.HOME',
'android.intent.category.BROWSABLE'])
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-c android.intent.category.HOME '
'-c android.intent.category.BROWSABLE '
'-n test.package/.Main'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
def testStartActivity_withData(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main',
data='http://www.google.com/')
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-d http://www.google.com/ '
'-n test.package/.Main'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
def testStartActivity_withStringExtra(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main',
extras={'foo': 'test'})
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-n test.package/.Main '
'--es foo test'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
def testStartActivity_withBoolExtra(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main',
extras={'foo': True})
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-n test.package/.Main '
'--ez foo True'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
def testStartActivity_withIntExtra(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main',
extras={'foo': 123})
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-n test.package/.Main '
'--ei foo 123'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
def testStartActivity_withTraceFile(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main')
with self.assertCall(
self.call.adb.Shell('am start '
'--start-profiler test_trace_file.out '
'-a android.intent.action.VIEW '
'-n test.package/.Main'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent,
trace_file_name='test_trace_file.out')
def testStartActivity_withForceStop(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main')
with self.assertCall(
self.call.adb.Shell('am start '
'-S '
'-a android.intent.action.VIEW '
'-n test.package/.Main'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent, force_stop=True)
def testStartActivity_withFlags(self):
test_intent = intent.Intent(action='android.intent.action.VIEW',
package='test.package',
activity='.Main',
flags='0x10000000')
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-n test.package/.Main '
'-f 0x10000000'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
class DeviceUtilsStartInstrumentationTest(DeviceUtilsTest):
def testStartInstrumentation_nothing(self):
with self.assertCalls(
self.call.device.RunShellCommand(
'p=test.package;am instrument "$p"/.TestInstrumentation',
check_return=True, large_output=True)):
self.device.StartInstrumentation(
'test.package/.TestInstrumentation',
finish=False, raw=False, extras=None)
def testStartInstrumentation_finish(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
'p=test.package;am instrument -w "$p"/.TestInstrumentation',
check_return=True, large_output=True),
['OK (1 test)'])):
output = self.device.StartInstrumentation(
'test.package/.TestInstrumentation',
finish=True, raw=False, extras=None)
self.assertEquals(['OK (1 test)'], output)
def testStartInstrumentation_raw(self):
with self.assertCalls(
self.call.device.RunShellCommand(
'p=test.package;am instrument -r "$p"/.TestInstrumentation',
check_return=True, large_output=True)):
self.device.StartInstrumentation(
'test.package/.TestInstrumentation',
finish=False, raw=True, extras=None)
def testStartInstrumentation_extras(self):
with self.assertCalls(
self.call.device.RunShellCommand(
'p=test.package;am instrument -e "$p".foo Foo -e bar \'Val \'"$p" '
'"$p"/.TestInstrumentation',
check_return=True, large_output=True)):
self.device.StartInstrumentation(
'test.package/.TestInstrumentation',
finish=False, raw=False, extras={'test.package.foo': 'Foo',
'bar': 'Val test.package'})
class DeviceUtilsBroadcastIntentTest(DeviceUtilsTest):
def testBroadcastIntent_noExtras(self):
test_intent = intent.Intent(action='test.package.with.an.INTENT')
with self.assertCall(
self.call.adb.Shell('am broadcast -a test.package.with.an.INTENT'),
'Broadcasting: Intent { act=test.package.with.an.INTENT } '):
self.device.BroadcastIntent(test_intent)
def testBroadcastIntent_withExtra(self):
test_intent = intent.Intent(action='test.package.with.an.INTENT',
extras={'foo': 'bar value'})
with self.assertCall(
self.call.adb.Shell(
"am broadcast -a test.package.with.an.INTENT --es foo 'bar value'"),
'Broadcasting: Intent { act=test.package.with.an.INTENT } '):
self.device.BroadcastIntent(test_intent)
def testBroadcastIntent_withExtra_noValue(self):
test_intent = intent.Intent(action='test.package.with.an.INTENT',
extras={'foo': None})
with self.assertCall(
self.call.adb.Shell(
'am broadcast -a test.package.with.an.INTENT --esn foo'),
'Broadcasting: Intent { act=test.package.with.an.INTENT } '):
self.device.BroadcastIntent(test_intent)
class DeviceUtilsGoHomeTest(DeviceUtilsTest):
def testGoHome_popupsExist(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True), []),
(self.call.device.RunShellCommand(
['am', 'start', '-W', '-a', 'android.intent.action.MAIN',
'-c', 'android.intent.category.HOME'], check_return=True),
'Starting: Intent { act=android.intent.action.MAIN }\r\n'''),
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True), []),
(self.call.device.RunShellCommand(
['input', 'keyevent', '66'], check_return=True)),
(self.call.device.RunShellCommand(
['input', 'keyevent', '4'], check_return=True)),
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True),
['mCurrentFocus Launcher'])):
self.device.GoHome()
def testGoHome_willRetry(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True), []),
(self.call.device.RunShellCommand(
['am', 'start', '-W', '-a', 'android.intent.action.MAIN',
'-c', 'android.intent.category.HOME'], check_return=True),
'Starting: Intent { act=android.intent.action.MAIN }\r\n'''),
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True), []),
(self.call.device.RunShellCommand(
['input', 'keyevent', '66'], check_return=True,)),
(self.call.device.RunShellCommand(
['input', 'keyevent', '4'], check_return=True)),
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True), []),
(self.call.device.RunShellCommand(
['input', 'keyevent', '66'], check_return=True)),
(self.call.device.RunShellCommand(
['input', 'keyevent', '4'], check_return=True)),
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True),
self.TimeoutError())):
with self.assertRaises(device_errors.CommandTimeoutError):
self.device.GoHome()
def testGoHome_alreadyFocused(self):
with self.assertCall(
self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True),
['mCurrentFocus Launcher']):
self.device.GoHome()
def testGoHome_alreadyFocusedAlternateCase(self):
with self.assertCall(
self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True),
[' mCurrentFocus .launcher/.']):
self.device.GoHome()
def testGoHome_obtainsFocusAfterGoingHome(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True), []),
(self.call.device.RunShellCommand(
['am', 'start', '-W', '-a', 'android.intent.action.MAIN',
'-c', 'android.intent.category.HOME'], check_return=True),
'Starting: Intent { act=android.intent.action.MAIN }\r\n'''),
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True),
['mCurrentFocus Launcher'])):
self.device.GoHome()
class DeviceUtilsForceStopTest(DeviceUtilsTest):
def testForceStop(self):
with self.assertCall(
self.call.adb.Shell('p=test.package;if [[ "$(ps)" = *$p* ]]; then '
'am force-stop $p; fi'),
''):
self.device.ForceStop('test.package')
class DeviceUtilsClearApplicationStateTest(DeviceUtilsTest):
def testClearApplicationState_setPermissions(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '17'),
(self.call.device._GetApplicationPathsInternal('this.package.exists'),
['/data/app/this.package.exists.apk']),
(self.call.device.RunShellCommand(
['pm', 'clear', 'this.package.exists'],
check_return=True),
['Success']),
(self.call.device.GrantPermissions(
'this.package.exists', ['p1']), [])):
self.device.ClearApplicationState(
'this.package.exists', permissions=['p1'])
def testClearApplicationState_packageDoesntExist(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '11'),
(self.call.device._GetApplicationPathsInternal('does.not.exist'),
[])):
self.device.ClearApplicationState('does.not.exist')
def testClearApplicationState_packageDoesntExistOnAndroidJBMR2OrAbove(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '18'),
(self.call.device.RunShellCommand(
['pm', 'clear', 'this.package.does.not.exist'],
check_return=True),
['Failed'])):
self.device.ClearApplicationState('this.package.does.not.exist')
def testClearApplicationState_packageExists(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '17'),
(self.call.device._GetApplicationPathsInternal('this.package.exists'),
['/data/app/this.package.exists.apk']),
(self.call.device.RunShellCommand(
['pm', 'clear', 'this.package.exists'],
check_return=True),
['Success'])):
self.device.ClearApplicationState('this.package.exists')
def testClearApplicationState_packageExistsOnAndroidJBMR2OrAbove(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '18'),
(self.call.device.RunShellCommand(
['pm', 'clear', 'this.package.exists'],
check_return=True),
['Success'])):
self.device.ClearApplicationState('this.package.exists')
class DeviceUtilsSendKeyEventTest(DeviceUtilsTest):
def testSendKeyEvent(self):
with self.assertCall(self.call.adb.Shell('input keyevent 66'), ''):
self.device.SendKeyEvent(66)
class DeviceUtilsPushChangedFilesIndividuallyTest(DeviceUtilsTest):
def testPushChangedFilesIndividually_empty(self):
test_files = []
with self.assertCalls():
self.device._PushChangedFilesIndividually(test_files)
def testPushChangedFilesIndividually_single(self):
test_files = [('/test/host/path', '/test/device/path')]
with self.assertCalls(self.call.adb.Push(*test_files[0])):
self.device._PushChangedFilesIndividually(test_files)
def testPushChangedFilesIndividually_multiple(self):
test_files = [
('/test/host/path/file1', '/test/device/path/file1'),
('/test/host/path/file2', '/test/device/path/file2')]
with self.assertCalls(
self.call.adb.Push(*test_files[0]),
self.call.adb.Push(*test_files[1])):
self.device._PushChangedFilesIndividually(test_files)
class DeviceUtilsPushChangedFilesZippedTest(DeviceUtilsTest):
def testPushChangedFilesZipped_noUnzipCommand(self):
test_files = [('/test/host/path/file1', '/test/device/path/file1')]
mock_zip_temp = mock.mock_open()
mock_zip_temp.return_value.name = '/test/temp/file/tmp.zip'
with self.assertCalls(
(mock.call.tempfile.NamedTemporaryFile(suffix='.zip'), mock_zip_temp),
(mock.call.multiprocessing.Process(
target=device_utils.DeviceUtils._CreateDeviceZip,
args=('/test/temp/file/tmp.zip', test_files)), mock.Mock()),
(self.call.device._MaybeInstallCommands(), False)):
self.assertFalse(self.device._PushChangedFilesZipped(test_files,
['/test/dir']))
def _testPushChangedFilesZipped_spec(self, test_files):
mock_zip_temp = mock.mock_open()
mock_zip_temp.return_value.name = '/test/temp/file/tmp.zip'
with self.assertCalls(
(mock.call.tempfile.NamedTemporaryFile(suffix='.zip'), mock_zip_temp),
(mock.call.multiprocessing.Process(
target=device_utils.DeviceUtils._CreateDeviceZip,
args=('/test/temp/file/tmp.zip', test_files)), mock.Mock()),
(self.call.device._MaybeInstallCommands(), True),
(self.call.device.NeedsSU(), True),
(mock.call.devil.android.device_temp_file.DeviceTempFile(self.adb,
suffix='.zip'),
MockTempFile('/test/sdcard/foo123.zip')),
self.call.adb.Push(
'/test/temp/file/tmp.zip', '/test/sdcard/foo123.zip'),
self.call.device.RunShellCommand(
'unzip /test/sdcard/foo123.zip&&chmod -R 777 /test/dir',
as_root=True,
env={'PATH': '/data/local/tmp/bin:$PATH'},
check_return=True)):
self.assertTrue(self.device._PushChangedFilesZipped(test_files,
['/test/dir']))
def testPushChangedFilesZipped_single(self):
self._testPushChangedFilesZipped_spec(
[('/test/host/path/file1', '/test/device/path/file1')])
def testPushChangedFilesZipped_multiple(self):
self._testPushChangedFilesZipped_spec(
[('/test/host/path/file1', '/test/device/path/file1'),
('/test/host/path/file2', '/test/device/path/file2')])
class DeviceUtilsPathExistsTest(DeviceUtilsTest):
def testPathExists_pathExists(self):
with self.assertCall(
self.call.device.RunShellCommand(
"test -e '/path/file exists'",
as_root=False, check_return=True, timeout=10, retries=0),
[]):
self.assertTrue(self.device.PathExists('/path/file exists'))
def testPathExists_multiplePathExists(self):
with self.assertCall(
self.call.device.RunShellCommand(
"test -e '/path 1' -a -e /path2",
as_root=False, check_return=True, timeout=10, retries=0),
[]):
self.assertTrue(self.device.PathExists(('/path 1', '/path2')))
def testPathExists_pathDoesntExist(self):
with self.assertCall(
self.call.device.RunShellCommand(
"test -e /path/file.not.exists",
as_root=False, check_return=True, timeout=10, retries=0),
self.ShellError()):
self.assertFalse(self.device.PathExists('/path/file.not.exists'))
def testPathExists_asRoot(self):
with self.assertCall(
self.call.device.RunShellCommand(
"test -e /root/path/exists",
as_root=True, check_return=True, timeout=10, retries=0),
self.ShellError()):
self.assertFalse(
self.device.PathExists('/root/path/exists', as_root=True))
def testFileExists_pathDoesntExist(self):
with self.assertCall(
self.call.device.RunShellCommand(
"test -e /path/file.not.exists",
as_root=False, check_return=True, timeout=10, retries=0),
self.ShellError()):
self.assertFalse(self.device.FileExists('/path/file.not.exists'))
class DeviceUtilsPullFileTest(DeviceUtilsTest):
def testPullFile_existsOnDevice(self):
with mock.patch('os.path.exists', return_value=True):
with self.assertCall(
self.call.adb.Pull('/data/app/test.file.exists',
'/test/file/host/path')):
self.device.PullFile('/data/app/test.file.exists',
'/test/file/host/path')
def testPullFile_doesntExistOnDevice(self):
with mock.patch('os.path.exists', return_value=True):
with self.assertCall(
self.call.adb.Pull('/data/app/test.file.does.not.exist',
'/test/file/host/path'),
self.CommandError('remote object does not exist')):
with self.assertRaises(device_errors.CommandFailedError):
self.device.PullFile('/data/app/test.file.does.not.exist',
'/test/file/host/path')
class DeviceUtilsReadFileTest(DeviceUtilsTest):
def testReadFileWithPull_success(self):
tmp_host_dir = '/tmp/dir/on.host/'
tmp_host = MockTempFile('/tmp/dir/on.host/tmp_ReadFileWithPull')
tmp_host.file.read.return_value = 'some interesting contents'
with self.assertCalls(
(mock.call.tempfile.mkdtemp(), tmp_host_dir),
(self.call.adb.Pull('/path/to/device/file', mock.ANY)),
(mock.call.__builtin__.open(mock.ANY, 'r'), tmp_host),
(mock.call.os.path.exists(tmp_host_dir), True),
(mock.call.shutil.rmtree(tmp_host_dir), None)):
self.assertEquals('some interesting contents',
self.device._ReadFileWithPull('/path/to/device/file'))
tmp_host.file.read.assert_called_once_with()
def testReadFileWithPull_rejected(self):
tmp_host_dir = '/tmp/dir/on.host/'
with self.assertCalls(
(mock.call.tempfile.mkdtemp(), tmp_host_dir),
(self.call.adb.Pull('/path/to/device/file', mock.ANY),
self.CommandError()),
(mock.call.os.path.exists(tmp_host_dir), True),
(mock.call.shutil.rmtree(tmp_host_dir), None)):
with self.assertRaises(device_errors.CommandFailedError):
self.device._ReadFileWithPull('/path/to/device/file')
def testReadFile_exists(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['ls', '-l', '/read/this/test/file'],
as_root=False, check_return=True),
['-rw-rw---- root foo 256 1970-01-01 00:00 file']),
(self.call.device.RunShellCommand(
['cat', '/read/this/test/file'],
as_root=False, check_return=True),
['this is a test file'])):
self.assertEqual('this is a test file\n',
self.device.ReadFile('/read/this/test/file'))
def testReadFile_exists2(self):
# Same as testReadFile_exists, but uses Android N ls output.
with self.assertCalls(
(self.call.device.RunShellCommand(
['ls', '-l', '/read/this/test/file'],
as_root=False, check_return=True),
['-rw-rw-rw- 1 root root 256 2016-03-15 03:27 /read/this/test/file']),
(self.call.device.RunShellCommand(
['cat', '/read/this/test/file'],
as_root=False, check_return=True),
['this is a test file'])):
self.assertEqual('this is a test file\n',
self.device.ReadFile('/read/this/test/file'))
def testReadFile_doesNotExist(self):
with self.assertCall(
self.call.device.RunShellCommand(
['ls', '-l', '/this/file/does.not.exist'],
as_root=False, check_return=True),
self.CommandError('File does not exist')):
with self.assertRaises(device_errors.CommandFailedError):
self.device.ReadFile('/this/file/does.not.exist')
def testReadFile_zeroSize(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['ls', '-l', '/this/file/has/zero/size'],
as_root=False, check_return=True),
['-r--r--r-- root foo 0 1970-01-01 00:00 zero_size_file']),
(self.call.device._ReadFileWithPull('/this/file/has/zero/size'),
'but it has contents\n')):
self.assertEqual('but it has contents\n',
self.device.ReadFile('/this/file/has/zero/size'))
def testReadFile_withSU(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['ls', '-l', '/this/file/can.be.read.with.su'],
as_root=True, check_return=True),
['-rw------- root root 256 1970-01-01 00:00 can.be.read.with.su']),
(self.call.device.RunShellCommand(
['cat', '/this/file/can.be.read.with.su'],
as_root=True, check_return=True),
['this is a test file', 'read with su'])):
self.assertEqual(
'this is a test file\nread with su\n',
self.device.ReadFile('/this/file/can.be.read.with.su',
as_root=True))
def testReadFile_withPull(self):
contents = 'a' * 123456
with self.assertCalls(
(self.call.device.RunShellCommand(
['ls', '-l', '/read/this/big/test/file'],
as_root=False, check_return=True),
['-rw-rw---- root foo 123456 1970-01-01 00:00 file']),
(self.call.device._ReadFileWithPull('/read/this/big/test/file'),
contents)):
self.assertEqual(
contents, self.device.ReadFile('/read/this/big/test/file'))
def testReadFile_withPullAndSU(self):
contents = 'b' * 123456
with self.assertCalls(
(self.call.device.RunShellCommand(
['ls', '-l', '/this/big/file/can.be.read.with.su'],
as_root=True, check_return=True),
['-rw------- root root 123456 1970-01-01 00:00 can.be.read.with.su']),
(self.call.device.NeedsSU(), True),
(mock.call.devil.android.device_temp_file.DeviceTempFile(self.adb),
MockTempFile('/sdcard/tmp/on.device')),
self.call.device.RunShellCommand(
'SRC=/this/big/file/can.be.read.with.su DEST=/sdcard/tmp/on.device;'
'cp "$SRC" "$DEST" && chmod 666 "$DEST"',
as_root=True, check_return=True),
(self.call.device._ReadFileWithPull('/sdcard/tmp/on.device'),
contents)):
self.assertEqual(
contents,
self.device.ReadFile('/this/big/file/can.be.read.with.su',
as_root=True))
def testReadFile_forcePull(self):
contents = 'a' * 123456
with self.assertCall(
self.call.device._ReadFileWithPull('/read/this/big/test/file'),
contents):
self.assertEqual(
contents,
self.device.ReadFile('/read/this/big/test/file', force_pull=True))
class DeviceUtilsWriteFileTest(DeviceUtilsTest):
def testWriteFileWithPush_success(self):
tmp_host = MockTempFile('/tmp/file/on.host')
contents = 'some interesting contents'
with self.assertCalls(
(mock.call.tempfile.NamedTemporaryFile(), tmp_host),
self.call.adb.Push('/tmp/file/on.host', '/path/to/device/file')):
self.device._WriteFileWithPush('/path/to/device/file', contents)
tmp_host.file.write.assert_called_once_with(contents)
def testWriteFileWithPush_rejected(self):
tmp_host = MockTempFile('/tmp/file/on.host')
contents = 'some interesting contents'
with self.assertCalls(
(mock.call.tempfile.NamedTemporaryFile(), tmp_host),
(self.call.adb.Push('/tmp/file/on.host', '/path/to/device/file'),
self.CommandError())):
with self.assertRaises(device_errors.CommandFailedError):
self.device._WriteFileWithPush('/path/to/device/file', contents)
def testWriteFile_withPush(self):
contents = 'some large contents ' * 26 # 20 * 26 = 520 chars
with self.assertCalls(
self.call.device._WriteFileWithPush('/path/to/device/file', contents)):
self.device.WriteFile('/path/to/device/file', contents)
def testWriteFile_withPushForced(self):
contents = 'tiny contents'
with self.assertCalls(
self.call.device._WriteFileWithPush('/path/to/device/file', contents)):
self.device.WriteFile('/path/to/device/file', contents, force_push=True)
def testWriteFile_withPushAndSU(self):
contents = 'some large contents ' * 26 # 20 * 26 = 520 chars
with self.assertCalls(
(self.call.device.NeedsSU(), True),
(mock.call.devil.android.device_temp_file.DeviceTempFile(self.adb),
MockTempFile('/sdcard/tmp/on.device')),
self.call.device._WriteFileWithPush('/sdcard/tmp/on.device', contents),
self.call.device.RunShellCommand(
['cp', '/sdcard/tmp/on.device', '/path/to/device/file'],
as_root=True, check_return=True)):
self.device.WriteFile('/path/to/device/file', contents, as_root=True)
def testWriteFile_withEcho(self):
with self.assertCall(self.call.adb.Shell(
"echo -n the.contents > /test/file/to.write"), ''):
self.device.WriteFile('/test/file/to.write', 'the.contents')
def testWriteFile_withEchoAndQuotes(self):
with self.assertCall(self.call.adb.Shell(
"echo -n 'the contents' > '/test/file/to write'"), ''):
self.device.WriteFile('/test/file/to write', 'the contents')
def testWriteFile_withEchoAndSU(self):
expected_cmd_without_su = "sh -c 'echo -n contents > /test/file'"
expected_cmd = 'su -c %s' % expected_cmd_without_su
with self.assertCalls(
(self.call.device.NeedsSU(), True),
(self.call.device._Su(expected_cmd_without_su), expected_cmd),
(self.call.adb.Shell(expected_cmd),
'')):
self.device.WriteFile('/test/file', 'contents', as_root=True)
class DeviceUtilsLsTest(DeviceUtilsTest):
def testLs_directory(self):
result = [('.', adb_wrapper.DeviceStat(16889, 4096, 1417436123)),
('..', adb_wrapper.DeviceStat(16873, 4096, 12382237)),
('testfile.txt', adb_wrapper.DeviceStat(33206, 3, 1417436122))]
with self.assertCalls(
(self.call.adb.Ls('/data/local/tmp'), result)):
self.assertEquals(result,
self.device.Ls('/data/local/tmp'))
def testLs_nothing(self):
with self.assertCalls(
(self.call.adb.Ls('/data/local/tmp/testfile.txt'), [])):
self.assertEquals([],
self.device.Ls('/data/local/tmp/testfile.txt'))
class DeviceUtilsStatTest(DeviceUtilsTest):
def testStat_file(self):
result = [('.', adb_wrapper.DeviceStat(16889, 4096, 1417436123)),
('..', adb_wrapper.DeviceStat(16873, 4096, 12382237)),
('testfile.txt', adb_wrapper.DeviceStat(33206, 3, 1417436122))]
with self.assertCalls(
(self.call.adb.Ls('/data/local/tmp'), result)):
self.assertEquals(adb_wrapper.DeviceStat(33206, 3, 1417436122),
self.device.Stat('/data/local/tmp/testfile.txt'))
def testStat_directory(self):
result = [('.', adb_wrapper.DeviceStat(16873, 4096, 12382237)),
('..', adb_wrapper.DeviceStat(16873, 4096, 12382237)),
('tmp', adb_wrapper.DeviceStat(16889, 4096, 1417436123))]
with self.assertCalls(
(self.call.adb.Ls('/data/local'), result)):
self.assertEquals(adb_wrapper.DeviceStat(16889, 4096, 1417436123),
self.device.Stat('/data/local/tmp'))
def testStat_doesNotExist(self):
result = [('.', adb_wrapper.DeviceStat(16889, 4096, 1417436123)),
('..', adb_wrapper.DeviceStat(16873, 4096, 12382237)),
('testfile.txt', adb_wrapper.DeviceStat(33206, 3, 1417436122))]
with self.assertCalls(
(self.call.adb.Ls('/data/local/tmp'), result)):
with self.assertRaises(device_errors.CommandFailedError):
self.device.Stat('/data/local/tmp/does.not.exist.txt')
class DeviceUtilsSetJavaAssertsTest(DeviceUtilsTest):
def testSetJavaAsserts_enable(self):
with self.assertCalls(
(self.call.device.ReadFile(self.device.LOCAL_PROPERTIES_PATH),
'some.example.prop=with an example value\n'
'some.other.prop=value_ok\n'),
self.call.device.WriteFile(
self.device.LOCAL_PROPERTIES_PATH,
'some.example.prop=with an example value\n'
'some.other.prop=value_ok\n'
'dalvik.vm.enableassertions=all\n'),
(self.call.device.GetProp('dalvik.vm.enableassertions'), ''),
self.call.device.SetProp('dalvik.vm.enableassertions', 'all')):
self.assertTrue(self.device.SetJavaAsserts(True))
def testSetJavaAsserts_disable(self):
with self.assertCalls(
(self.call.device.ReadFile(self.device.LOCAL_PROPERTIES_PATH),
'some.example.prop=with an example value\n'
'dalvik.vm.enableassertions=all\n'
'some.other.prop=value_ok\n'),
self.call.device.WriteFile(
self.device.LOCAL_PROPERTIES_PATH,
'some.example.prop=with an example value\n'
'some.other.prop=value_ok\n'),
(self.call.device.GetProp('dalvik.vm.enableassertions'), 'all'),
self.call.device.SetProp('dalvik.vm.enableassertions', '')):
self.assertTrue(self.device.SetJavaAsserts(False))
def testSetJavaAsserts_alreadyEnabled(self):
with self.assertCalls(
(self.call.device.ReadFile(self.device.LOCAL_PROPERTIES_PATH),
'some.example.prop=with an example value\n'
'dalvik.vm.enableassertions=all\n'
'some.other.prop=value_ok\n'),
(self.call.device.GetProp('dalvik.vm.enableassertions'), 'all')):
self.assertFalse(self.device.SetJavaAsserts(True))
def testSetJavaAsserts_malformedLocalProp(self):
with self.assertCalls(
(self.call.device.ReadFile(self.device.LOCAL_PROPERTIES_PATH),
'some.example.prop=with an example value\n'
'malformed_property\n'
'dalvik.vm.enableassertions=all\n'
'some.other.prop=value_ok\n'),
(self.call.device.GetProp('dalvik.vm.enableassertions'), 'all')):
self.assertFalse(self.device.SetJavaAsserts(True))
class DeviceUtilsGetPropTest(DeviceUtilsTest):
def testGetProp_exists(self):
with self.assertCall(
self.call.device.RunShellCommand(
['getprop', 'test.property'], check_return=True, single_line=True,
timeout=self.device._default_timeout,
retries=self.device._default_retries),
'property_value'):
self.assertEqual('property_value',
self.device.GetProp('test.property'))
def testGetProp_doesNotExist(self):
with self.assertCall(
self.call.device.RunShellCommand(
['getprop', 'property.does.not.exist'],
check_return=True, single_line=True,
timeout=self.device._default_timeout,
retries=self.device._default_retries),
''):
self.assertEqual('', self.device.GetProp('property.does.not.exist'))
def testGetProp_cachedRoProp(self):
with self.assertCall(
self.call.device.RunShellCommand(
['getprop'], check_return=True, large_output=True,
timeout=self.device._default_timeout,
retries=self.device._default_retries),
['[ro.build.type]: [userdebug]']):
self.assertEqual('userdebug',
self.device.GetProp('ro.build.type', cache=True))
self.assertEqual('userdebug',
self.device.GetProp('ro.build.type', cache=True))
def testGetProp_retryAndCache(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['getprop'], check_return=True, large_output=True,
timeout=self.device._default_timeout,
retries=3),
['[ro.build.type]: [userdebug]'])):
self.assertEqual('userdebug',
self.device.GetProp('ro.build.type',
cache=True, retries=3))
self.assertEqual('userdebug',
self.device.GetProp('ro.build.type',
cache=True, retries=3))
class DeviceUtilsSetPropTest(DeviceUtilsTest):
def testSetProp(self):
with self.assertCall(
self.call.device.RunShellCommand(
['setprop', 'test.property', 'test value'], check_return=True)):
self.device.SetProp('test.property', 'test value')
def testSetProp_check_succeeds(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['setprop', 'test.property', 'new_value'], check_return=True)),
(self.call.device.GetProp('test.property', cache=False), 'new_value')):
self.device.SetProp('test.property', 'new_value', check=True)
def testSetProp_check_fails(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['setprop', 'test.property', 'new_value'], check_return=True)),
(self.call.device.GetProp('test.property', cache=False), 'old_value')):
with self.assertRaises(device_errors.CommandFailedError):
self.device.SetProp('test.property', 'new_value', check=True)
class DeviceUtilsGetPidsTest(DeviceUtilsTest):
def testGetPids_noMatches(self):
with self.assertCall(
self.call.device._RunPipedShellCommand('ps | grep -F does.not.match'),
[]):
self.assertEqual({}, self.device.GetPids('does.not.match'))
def testGetPids_oneMatch(self):
with self.assertCall(
self.call.device._RunPipedShellCommand('ps | grep -F one.match'),
['user 1001 100 1024 1024 ffffffff 00000000 one.match']):
self.assertEqual(
{'one.match': ['1001']},
self.device.GetPids('one.match'))
def testGetPids_multipleMatches(self):
with self.assertCall(
self.call.device._RunPipedShellCommand('ps | grep -F match'),
['user 1001 100 1024 1024 ffffffff 00000000 one.match',
'user 1002 100 1024 1024 ffffffff 00000000 two.match',
'user 1003 100 1024 1024 ffffffff 00000000 three.match']):
self.assertEqual(
{'one.match': ['1001'],
'two.match': ['1002'],
'three.match': ['1003']},
self.device.GetPids('match'))
def testGetPids_exactMatch(self):
with self.assertCall(
self.call.device._RunPipedShellCommand('ps | grep -F exact.match'),
['user 1000 100 1024 1024 ffffffff 00000000 not.exact.match',
'user 1234 100 1024 1024 ffffffff 00000000 exact.match']):
self.assertEqual(
{'not.exact.match': ['1000'], 'exact.match': ['1234']},
self.device.GetPids('exact.match'))
def testGetPids_quotable(self):
with self.assertCall(
self.call.device._RunPipedShellCommand("ps | grep -F 'my$process'"),
['user 1234 100 1024 1024 ffffffff 00000000 my$process']):
self.assertEqual(
{'my$process': ['1234']}, self.device.GetPids('my$process'))
def testGetPids_multipleInstances(self):
with self.assertCall(
self.call.device._RunPipedShellCommand('ps | grep -F foo'),
['user 1000 100 1024 1024 ffffffff 00000000 foo',
'user 1234 100 1024 1024 ffffffff 00000000 foo']):
self.assertEqual(
{'foo': ['1000', '1234']},
self.device.GetPids('foo'))
class DeviceUtilsTakeScreenshotTest(DeviceUtilsTest):
def testTakeScreenshot_fileNameProvided(self):
with self.assertCalls(
(mock.call.devil.android.device_temp_file.DeviceTempFile(
self.adb, suffix='.png'),
MockTempFile('/tmp/path/temp-123.png')),
(self.call.adb.Shell('/system/bin/screencap -p /tmp/path/temp-123.png'),
''),
self.call.device.PullFile('/tmp/path/temp-123.png',
'/test/host/screenshot.png')):
self.device.TakeScreenshot('/test/host/screenshot.png')
class DeviceUtilsGetMemoryUsageForPidTest(DeviceUtilsTest):
def setUp(self):
super(DeviceUtilsGetMemoryUsageForPidTest, self).setUp()
def testGetMemoryUsageForPid_validPid(self):
with self.assertCalls(
(self.call.device._RunPipedShellCommand(
'showmap 1234 | grep TOTAL', as_root=True),
['100 101 102 103 104 105 106 107 TOTAL']),
(self.call.device.ReadFile('/proc/1234/status', as_root=True),
'VmHWM: 1024 kB\n')):
self.assertEqual(
{
'Size': 100,
'Rss': 101,
'Pss': 102,
'Shared_Clean': 103,
'Shared_Dirty': 104,
'Private_Clean': 105,
'Private_Dirty': 106,
'VmHWM': 1024
},
self.device.GetMemoryUsageForPid(1234))
def testGetMemoryUsageForPid_noSmaps(self):
with self.assertCalls(
(self.call.device._RunPipedShellCommand(
'showmap 4321 | grep TOTAL', as_root=True),
['cannot open /proc/4321/smaps: No such file or directory']),
(self.call.device.ReadFile('/proc/4321/status', as_root=True),
'VmHWM: 1024 kb\n')):
self.assertEquals({'VmHWM': 1024}, self.device.GetMemoryUsageForPid(4321))
def testGetMemoryUsageForPid_noStatus(self):
with self.assertCalls(
(self.call.device._RunPipedShellCommand(
'showmap 4321 | grep TOTAL', as_root=True),
['100 101 102 103 104 105 106 107 TOTAL']),
(self.call.device.ReadFile('/proc/4321/status', as_root=True),
self.CommandError())):
self.assertEquals(
{
'Size': 100,
'Rss': 101,
'Pss': 102,
'Shared_Clean': 103,
'Shared_Dirty': 104,
'Private_Clean': 105,
'Private_Dirty': 106,
},
self.device.GetMemoryUsageForPid(4321))
class DeviceUtilsDismissCrashDialogIfNeededTest(DeviceUtilsTest):
def testDismissCrashDialogIfNeeded_crashedPageckageNotFound(self):
sample_dumpsys_output = '''
WINDOW MANAGER WINDOWS (dumpsys window windows)
Window #11 Window{f8b647a u0 SearchPanel}:
mDisplayId=0 mSession=Session{8 94:122} mClient=android.os.BinderProxy@1ba5
mOwnerUid=100 mShowToOwnerOnly=false package=com.android.systemui appop=NONE
mAttrs=WM.LayoutParams{(0,0)(fillxfill) gr=#53 sim=#31 ty=2024 fl=100
Requested w=1080 h=1920 mLayoutSeq=426
mBaseLayer=211000 mSubLayer=0 mAnimLayer=211000+0=211000 mLastLayer=211000
'''
with self.assertCalls(
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True), sample_dumpsys_output.split('\n'))):
package_name = self.device.DismissCrashDialogIfNeeded()
self.assertIsNone(package_name)
def testDismissCrashDialogIfNeeded_crashedPageckageFound(self):
sample_dumpsys_output = '''
WINDOW MANAGER WINDOWS (dumpsys window windows)
Window #11 Window{f8b647a u0 SearchPanel}:
mDisplayId=0 mSession=Session{8 94:122} mClient=android.os.BinderProxy@1ba5
mOwnerUid=102 mShowToOwnerOnly=false package=com.android.systemui appop=NONE
mAttrs=WM.LayoutParams{(0,0)(fillxfill) gr=#53 sim=#31 ty=2024 fl=100
Requested w=1080 h=1920 mLayoutSeq=426
mBaseLayer=211000 mSubLayer=0 mAnimLayer=211000+0=211000 mLastLayer=211000
mHasPermanentDpad=false
mCurrentFocus=Window{3a27740f u0 Application Error: com.android.chrome}
mFocusedApp=AppWindowToken{470af6f token=Token{272ec24e ActivityRecord{t894}}}
'''
with self.assertCalls(
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True), sample_dumpsys_output.split('\n')),
(self.call.device.RunShellCommand(
['input', 'keyevent', '22'], check_return=True)),
(self.call.device.RunShellCommand(
['input', 'keyevent', '22'], check_return=True)),
(self.call.device.RunShellCommand(
['input', 'keyevent', '66'], check_return=True)),
(self.call.device.RunShellCommand(
['dumpsys', 'window', 'windows'], check_return=True,
large_output=True), [])):
package_name = self.device.DismissCrashDialogIfNeeded()
self.assertEqual(package_name, 'com.android.chrome')
class DeviceUtilsClientCache(DeviceUtilsTest):
def testClientCache_twoCaches(self):
self.device._cache['test'] = 0
client_cache_one = self.device.GetClientCache('ClientOne')
client_cache_one['test'] = 1
client_cache_two = self.device.GetClientCache('ClientTwo')
client_cache_two['test'] = 2
self.assertEqual(self.device._cache['test'], 0)
self.assertEqual(client_cache_one, {'test': 1})
self.assertEqual(client_cache_two, {'test': 2})
self.device._ClearCache()
self.assertTrue('test' not in self.device._cache)
self.assertEqual(client_cache_one, {})
self.assertEqual(client_cache_two, {})
def testClientCache_multipleInstances(self):
client_cache_one = self.device.GetClientCache('ClientOne')
client_cache_one['test'] = 1
client_cache_two = self.device.GetClientCache('ClientOne')
self.assertEqual(client_cache_one, {'test': 1})
self.assertEqual(client_cache_two, {'test': 1})
self.device._ClearCache()
self.assertEqual(client_cache_one, {})
self.assertEqual(client_cache_two, {})
class DeviceUtilsHealthyDevicesTest(mock_calls.TestCase):
def testHealthyDevices_emptyBlacklist(self):
test_serials = ['0123456789abcdef', 'fedcba9876543210']
with self.assertCalls(
(mock.call.devil.android.sdk.adb_wrapper.AdbWrapper.Devices(),
[_AdbWrapperMock(s) for s in test_serials])):
blacklist = mock.NonCallableMock(**{'Read.return_value': []})
devices = device_utils.DeviceUtils.HealthyDevices(blacklist)
for serial, device in zip(test_serials, devices):
self.assertTrue(isinstance(device, device_utils.DeviceUtils))
self.assertEquals(serial, device.adb.GetDeviceSerial())
def testHealthyDevices_blacklist(self):
test_serials = ['0123456789abcdef', 'fedcba9876543210']
with self.assertCalls(
(mock.call.devil.android.sdk.adb_wrapper.AdbWrapper.Devices(),
[_AdbWrapperMock(s) for s in test_serials])):
blacklist = mock.NonCallableMock(
**{'Read.return_value': ['fedcba9876543210']})
devices = device_utils.DeviceUtils.HealthyDevices(blacklist)
self.assertEquals(1, len(devices))
self.assertTrue(isinstance(devices[0], device_utils.DeviceUtils))
self.assertEquals('0123456789abcdef', devices[0].adb.GetDeviceSerial())
class DeviceUtilsRestartAdbdTest(DeviceUtilsTest):
def testAdbdRestart(self):
mock_temp_file = '/sdcard/temp-123.sh'
with self.assertCalls(
(mock.call.devil.android.device_temp_file.DeviceTempFile(
self.adb, suffix='.sh'), MockTempFile(mock_temp_file)),
self.call.device.WriteFile(mock.ANY, mock.ANY),
(self.call.device.RunShellCommand(
['source', mock_temp_file], as_root=True)),
self.call.adb.WaitForDevice()):
self.device.RestartAdbd()
class DeviceUtilsGrantPermissionsTest(DeviceUtilsTest):
def testGrantPermissions_none(self):
self.device.GrantPermissions('package', [])
def testGrantPermissions_underM(self):
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.LOLLIPOP):
self.device.GrantPermissions('package', ['p1'])
def testGrantPermissions_one(self):
permissions_cmd = 'pm grant package p1'
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.MARSHMALLOW):
with self.assertCalls(
(self.call.device.RunShellCommand(
permissions_cmd, check_return=True), [])):
self.device.GrantPermissions('package', ['p1'])
def testGrantPermissions_multiple(self):
permissions_cmd = 'pm grant package p1&&pm grant package p2'
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.MARSHMALLOW):
with self.assertCalls(
(self.call.device.RunShellCommand(
permissions_cmd, check_return=True), [])):
self.device.GrantPermissions('package', ['p1', 'p2'])
def testGrantPermissions_WriteExtrnalStorage(self):
permissions_cmd = (
'pm grant package android.permission.WRITE_EXTERNAL_STORAGE&&'
'pm grant package android.permission.READ_EXTERNAL_STORAGE')
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.MARSHMALLOW):
with self.assertCalls(
(self.call.device.RunShellCommand(
permissions_cmd, check_return=True), [])):
self.device.GrantPermissions(
'package', ['android.permission.WRITE_EXTERNAL_STORAGE'])
def testGrantPermissions_BlackList(self):
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.MARSHMALLOW):
self.device.GrantPermissions(
'package', ['android.permission.ACCESS_MOCK_LOCATION'])
class DeviecUtilsIsScreenOn(DeviceUtilsTest):
_L_SCREEN_ON = ['test=test mInteractive=true']
_K_SCREEN_ON = ['test=test mScreenOn=true']
_L_SCREEN_OFF = ['mInteractive=false']
_K_SCREEN_OFF = ['mScreenOn=false']
def testIsScreenOn_onPreL(self):
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.KITKAT):
with self.assertCalls(
(self.call.device._RunPipedShellCommand(
'dumpsys input_method | grep mScreenOn'), self._K_SCREEN_ON)):
self.assertTrue(self.device.IsScreenOn())
def testIsScreenOn_onL(self):
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.LOLLIPOP):
with self.assertCalls(
(self.call.device._RunPipedShellCommand(
'dumpsys input_method | grep mInteractive'), self._L_SCREEN_ON)):
self.assertTrue(self.device.IsScreenOn())
def testIsScreenOn_offPreL(self):
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.KITKAT):
with self.assertCalls(
(self.call.device._RunPipedShellCommand(
'dumpsys input_method | grep mScreenOn'), self._K_SCREEN_OFF)):
self.assertFalse(self.device.IsScreenOn())
def testIsScreenOn_offL(self):
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.LOLLIPOP):
with self.assertCalls(
(self.call.device._RunPipedShellCommand(
'dumpsys input_method | grep mInteractive'), self._L_SCREEN_OFF)):
self.assertFalse(self.device.IsScreenOn())
def testIsScreenOn_noOutput(self):
with self.patch_call(self.call.device.build_version_sdk,
return_value=version_codes.LOLLIPOP):
with self.assertCalls(
(self.call.device._RunPipedShellCommand(
'dumpsys input_method | grep mInteractive'), [])):
with self.assertRaises(device_errors.CommandFailedError):
self.device.IsScreenOn()
class DeviecUtilsSetScreen(DeviceUtilsTest):
@mock.patch('time.sleep', mock.Mock())
def testSetScren_alreadySet(self):
with self.assertCalls(
(self.call.device.IsScreenOn(), False)):
self.device.SetScreen(False)
@mock.patch('time.sleep', mock.Mock())
def testSetScreen_on(self):
with self.assertCalls(
(self.call.device.IsScreenOn(), False),
(self.call.device.RunShellCommand('input keyevent 26'), []),
(self.call.device.IsScreenOn(), True)):
self.device.SetScreen(True)
@mock.patch('time.sleep', mock.Mock())
def testSetScreen_off(self):
with self.assertCalls(
(self.call.device.IsScreenOn(), True),
(self.call.device.RunShellCommand('input keyevent 26'), []),
(self.call.device.IsScreenOn(), False)):
self.device.SetScreen(False)
@mock.patch('time.sleep', mock.Mock())
def testSetScreen_slow(self):
with self.assertCalls(
(self.call.device.IsScreenOn(), True),
(self.call.device.RunShellCommand('input keyevent 26'), []),
(self.call.device.IsScreenOn(), True),
(self.call.device.IsScreenOn(), True),
(self.call.device.IsScreenOn(), False)):
self.device.SetScreen(False)
if __name__ == '__main__':
logging.getLogger().setLevel(logging.DEBUG)
unittest.main(verbosity=2)