blob: 2e2ee2409ef5f38df48c34f45fd5d7a435df140d [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 collections
import contextlib
import io
import json
import logging
import os
import posixpath
import stat
import sys
import unittest
import six
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.ndk import abis
from devil.android.sdk import adb_wrapper
from devil.android.sdk import intent
from devil.android.sdk import keyevent
from devil.android.sdk import version_codes
from devil.utils import cmd_helper
from devil.utils import mock_calls
with devil_env.SysPath(os.path.join(devil_env.PY_UTILS_PATH)):
from py_utils import tempfile_ext
with devil_env.SysPath(devil_env.PYMOCK_PATH):
import mock # pylint: disable=import-error
TEST_APK_PATH = '/fake/test/app.apk'
TEST_PACKAGE = 'test.package'
def Process(name, pid, ppid='1'):
return device_utils.ProcessInfo(name=name, pid=pid, ppid=ppid)
def Processes(*args):
return [Process(*arg) for arg in args]
class AnyStringWith(object):
def __init__(self, value):
self._value = value
def __eq__(self, other):
return self._value in other
def __repr__(self):
return '<AnyStringWith: %s>' % self._value
class _FakeContextManager(object):
def __init__(self, obj):
self._obj = obj
def __enter__(self):
return self._obj
def __exit__(self, type_, value, traceback):
pass
class _MockApkHelper(object):
def __init__(self, path, package_name, perms=None, splits=None):
self.path = path
self.is_bundle = path.endswith('_bundle')
self.package_name = package_name
self.perms = perms
self.splits = splits if splits else []
self.abis = [abis.ARM]
self.version_code = None
def GetPackageName(self):
return self.package_name
def GetPermissions(self):
return self.perms
def GetVersionCode(self):
return self.version_code
def GetAbis(self):
return self.abis
def GetApkPaths(self,
device,
modules=None,
allow_cached_props=False,
additional_locales=None):
return _FakeContextManager([self.path] + self.splits)
#override
@staticmethod
def SupportsSplits():
return True
class _MockMultipleDevicesError(Exception):
pass
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):
if six.PY2:
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.assertEqual(['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'))):
adb_wrapper.RestartServer()
class MockTempFile(object):
def __init__(self, name='/tmp/some/file'):
self.file = mock.MagicMock(spec=io.BufferedIOBase)
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 MockLogger(mock.Mock):
def __init__(self, *args, **kwargs):
super(MockLogger, self).__init__(*args, **kwargs)
self.warnings = []
def warning(self, message, *args):
self.warnings.append(message % args)
def PatchLogger():
return mock.patch(
'devil.android.device_utils.logger', new_callable=MockLogger)
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 safeAssertItemsEqual(self, expected, actual):
if six.PY2:
self.assertItemsEqual(expected, actual)
else:
self.assertCountEqual(expected, actual) # pylint: disable=no-member
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)))
def EnsureCacheInitialized(self, props=None, sdcard='/sdcard'):
props = props or []
ret = [sdcard, 'TOKEN'] + props
return (self.call.device.RunShellCommand(
AnyStringWith('getprop'),
shell=True,
check_return=True,
large_output=True), ret)
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.assertEqual('0000000000000000',
sorted_devices[0].adb.GetDeviceSerial())
self.assertEqual('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.patch_call(self.call.device.build_type,
return_value='userdebug'), (self.assertCall(
self.call.adb.Shell('id'), 'uid=0(root)\n')):
self.assertTrue(self.device.HasRoot())
def testHasRootEngBuild_true(self):
with self.patch_call(self.call.device.build_type, return_value='eng'):
self.assertTrue(self.device.HasRoot())
def testHasRoot_false(self):
with self.patch_call(self.call.device.build_type,
return_value='userdebug'), (self.assertCall(
self.call.adb.Shell('id'), 'uid=2000(shell)\n')):
self.assertFalse(self.device.HasRoot())
class DeviceUtilsEnableRootTest(DeviceUtilsTest):
def testEnableRoot_succeeds(self):
with self.assertCalls(self.call.adb.Root(), self.call.adb.WaitForDevice(),
(self.call.device.HasRoot(), True)):
self.device.EnableRoot()
def testEnableRoot_userBuild(self):
with self.assertCalls((self.call.adb.Root(), self.AdbCommandError()),
(self.call.device.IsUserBuild(), True)):
with self.assertRaises(device_errors.CommandFailedError):
self.device.EnableRoot()
def testEnableRoot_rootFails(self):
with self.assertCalls((self.call.adb.Root(), self.AdbCommandError()),
(self.call.device.IsUserBuild(), False)):
with self.assertRaises(device_errors.AdbCommandFailedError):
self.device.EnableRoot()
def testEnableRoot_timeoutInWaitForDevice(self):
with self.assertCalls(
(self.call.adb.Root(),
self.AdbCommandError(
output='timeout expired while waiting for device')),
(self.call.device.IsUserBuild(), False), self.call.adb.WaitForDevice(),
(self.call.device.HasRoot(), True)):
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.assertCalls(
self.EnsureCacheInitialized(sdcard='/fake/storage/path')):
self.assertEqual('/fake/storage/path',
self.device.GetExternalStoragePath())
def testGetExternalStoragePath_fails(self):
with self.assertCalls(self.EnsureCacheInitialized(sdcard='')):
with self.assertRaises(device_errors.CommandFailedError):
self.device.GetExternalStoragePath()
class DeviceUtilsGetAppWritablePathTest(DeviceUtilsTest):
def testGetAppWritablePath_succeeds_sdk_pre_q(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '28'),
self.EnsureCacheInitialized(sdcard='/fake/storage/path')):
self.assertEqual('/fake/storage/path', self.device.GetAppWritablePath())
def testGetAppWritablePath_succeeds_sdk_q(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '29'),
self.EnsureCacheInitialized(sdcard='/fake/storage/path')):
self.assertEqual('/fake/storage/path/Download',
self.device.GetAppWritablePath())
def testGetAppWritablePath_fails(self):
with self.assertCalls(
(self.call.device.GetProp('ro.build.version.sdk', cache=True), '29'),
self.EnsureCacheInitialized(sdcard='')):
with self.assertRaises(device_errors.CommandFailedError):
self.device.GetAppWritablePath()
class DeviceUtilsIsApplicationInstalledTest(DeviceUtilsTest):
def testIsApplicationInstalled_installed(self):
with self.assertCalls((self.call.device.RunShellCommand(
['pm', 'list', 'packages', 'some.installed.app'], check_return=True),
['package:some.installed.app'])):
self.assertTrue(self.device.IsApplicationInstalled('some.installed.app'))
def testIsApplicationInstalled_notInstalled(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['pm', 'list', 'packages', 'not.installed.app'], check_return=True),
''),
(self.call.device.RunShellCommand(
['dumpsys', 'package'], check_return=True, large_output=True), [])):
self.assertFalse(self.device.IsApplicationInstalled('not.installed.app'))
def testIsApplicationInstalled_substringMatch(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['pm', 'list', 'packages', 'substring.of.package'],
check_return=True),
[
'package:first.substring.of.package',
'package:second.substring.of.package',
]),
(self.call.device.RunShellCommand(
['dumpsys', 'package'], check_return=True, large_output=True), [])):
self.assertFalse(
self.device.IsApplicationInstalled('substring.of.package'))
def testIsApplicationInstalled_dumpsysFallback(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['pm', 'list', 'packages', 'some.installed.app'],
check_return=True), []),
(self.call.device.RunShellCommand(
['dumpsys', 'package'], check_return=True, large_output=True),
['Package [some.installed.app] (a12345):'])):
self.assertTrue(self.device.IsApplicationInstalled('some.installed.app'))
def testIsApplicationInstalled_dumpsysFallbackVersioned(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['dumpsys', 'package'], check_return=True, large_output=True),
['Package [some.installed.app_1234] (a12345):'])):
self.assertTrue(
self.device.IsApplicationInstalled('some.installed.app', 1234))
def testIsApplicationInstalled_dumpsysFallbackVersionNotNeeded(self):
with self.assertCalls(
(self.call.device.RunShellCommand(
['dumpsys', 'package'], check_return=True, large_output=True),
['Package [some.installed.app] (a12345):'])):
self.assertTrue(
self.device.IsApplicationInstalled('some.installed.app', 1234))
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.assertEqual(['/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.assertEqual(
[], self.device._GetApplicationPathsInternal('not.installed.app'))
def testGetApplicationPathsInternal_garbageOutputRaises(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),
['garbage first line'])):
with self.assertRaises(device_errors.CommandFailedError):
self.device._GetApplicationPathsInternal('android')
def testGetApplicationPathsInternal_outputWarningsIgnored(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),
['WARNING: some warning message from pm'])):
self.assertEqual(
[], 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.assertEqual('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.assertEqual(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 DeviceUtils_GetApplicationTargetSdkTest(DeviceUtilsTest):
def test_GetApplicationTargetSdk_exists(self):
with self.assertCalls(
(self.call.device.IsApplicationInstalled('com.android.chrome'), True),
(self.call.device._GetDumpsysOutput(['package', 'com.android.chrome'],
'targetSdk='),
[' versionCode=413200001 minSdk=21 targetSdk=29'])):
self.assertEqual(
'29', self.device.GetApplicationTargetSdk('com.android.chrome'))
def test_GetApplicationTargetSdk_notExists(self):
with self.assertCalls(
(self.call.device.IsApplicationInstalled('com.android.chrome'), False)):
self.assertIsNone(
self.device.GetApplicationTargetSdk('com.android.chrome'))
def test_GetApplicationTargetSdk_fails(self):
with self.assertCalls(
(self.call.device.IsApplicationInstalled('com.android.chrome'), True),
(self.call.device._GetDumpsysOutput(['package', 'com.android.chrome'],
'targetSdk='), [])):
with self.assertRaises(device_errors.CommandFailedError):
self.device.GetApplicationTargetSdk('com.android.chrome')
def test_GetApplicationTargetSdk_prefinalizedSdk(self):
with self.assertCalls(
(self.call.device.IsApplicationInstalled('com.android.chrome'), True),
(self.call.device._GetDumpsysOutput(['package', 'com.android.chrome'],
'targetSdk='),
[' versionCode=410301483 minSdk=10000 targetSdk=10000']),
(self.call.device.GetProp('ro.build.version.codename',
cache=True), 'R')):
self.assertEqual(
'R', self.device.GetApplicationTargetSdk('com.android.chrome'))
class DeviceUtils_GetUidForPackageTest(DeviceUtilsTest):
def test_GetUidForPackage_Exists(self):
with self.assertCall(
self.call.device._GetDumpsysOutput(
['package', 'com.android.chrome'], 'userId='),
[' userId=1001']):
self.assertEquals('1001',
self.device.GetUidForPackage('com.android.chrome'))
def test_GetUidForPackage_notInstalled(self):
with self.assertCall(
self.call.device._GetDumpsysOutput(
['package', 'com.android.chrome'], 'userId='),
['']):
self.assertEquals(None,
self.device.GetUidForPackage('com.android.chrome'))
def test_GetUidForPackage_fails(self):
with self.assertCall(
self.call.device._GetDumpsysOutput(
['package', 'com.android.chrome'], 'userId='),
[]):
with self.assertRaises(device_errors.CommandFailedError):
self.device.GetUidForPackage('com.android.chrome')
class DeviceUtils_GetPackageArchitectureTest(DeviceUtilsTest):
def test_GetPackageArchitecture_exists(self):
with self.assertCall(
self.call.device._RunPipedShellCommand(
'dumpsys package com.android.chrome | grep -F primaryCpuAbi'),
[' primaryCpuAbi=armeabi-v7a']):
self.assertEqual(abis.ARM,
self.device.GetPackageArchitecture('com.android.chrome'))
def test_GetPackageArchitecture_notExists(self):
with self.assertCall(
self.call.device._RunPipedShellCommand(
'dumpsys package com.android.chrome | grep -F primaryCpuAbi'), []):
self.assertEqual(None,
self.device.GetPackageArchitecture('com.android.chrome'))
class DeviceUtilsGetApplicationDataDirectoryTest(DeviceUtilsTest):
def testGetApplicationDataDirectory_exists(self):
with self.assertCalls(
(self.call.device.IsApplicationInstalled('foo.bar.baz'), True),
(self.call.device._RunPipedShellCommand(
'pm dump foo.bar.baz | grep dataDir='),
['dataDir=/data/data/foo.bar.baz'])):
self.assertEqual('/data/data/foo.bar.baz',
self.device.GetApplicationDataDirectory('foo.bar.baz'))
def testGetApplicationDataDirectory_notInstalled(self):
with self.assertCalls(
(self.call.device.IsApplicationInstalled('foo.bar.baz'), False)):
with self.assertRaises(device_errors.CommandFailedError):
self.device.GetApplicationDataDirectory('foo.bar.baz')
def testGetApplicationDataDirectory_notExists(self):
with self.assertCalls(
(self.call.device.IsApplicationInstalled('foo.bar.baz'), True),
(self.call.device._RunPipedShellCommand(
'pm dump foo.bar.baz | grep dataDir='), self.ShellError())):
with self.assertRaises(device_errors.CommandFailedError):
self.device.GetApplicationDataDirectory('foo.bar.baz')
@mock.patch('time.sleep', mock.Mock())
class DeviceUtilsWaitUntilFullyBootedTest(DeviceUtilsTest):
def testWaitUntilFullyBooted_succeedsWithDefaults(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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, decrypt=False)
def testWaitUntilFullyBooted_succeedsWithWifi(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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, decrypt=False)
def testWaitUntilFullyBooted_succeedsWithDecryptFDE(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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'),
# decryption_completed
(self.call.device.GetProp('vold.decrypt', cache=False),
'trigger_restart_framework')):
self.device.WaitUntilFullyBooted(wifi=False, decrypt=True)
def testWaitUntilFullyBooted_succeedsWithDecryptNotFDE(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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'),
# decryption_completed
(self.call.device.GetProp('vold.decrypt', cache=False), '')):
self.device.WaitUntilFullyBooted(wifi=False, decrypt=True)
def testWaitUntilFullyBooted_deviceIsRock960(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), 'rk3399'),
(self.call.device.GetProp('sys.usb.config'), 'mtp,adb'),
(self.call.device.GetProp('ro.product.model'), 'rk3399'),
(self.call.device.GetProp('sys.usb.config'), 'adb'),
# 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, decrypt=False)
def testWaitUntilFullyBooted_deviceNotInitiallyAvailable(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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, decrypt=False)
def testWaitUntilFullyBooted_deviceBrieflyOffline(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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),
self.AdbCommandError()),
# boot_completed
(self.call.device.GetProp('sys.boot_completed', cache=False), '1')):
self.device.WaitUntilFullyBooted(wifi=False, decrypt=False)
def testWaitUntilFullyBooted_sdCardReadyFails_noPath(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# sd_card_ready
(self.call.device.GetExternalStoragePath(), self.CommandError())):
with self.assertRaises(device_errors.CommandFailedError):
self.device.WaitUntilFullyBooted(wifi=False, decrypt=False)
def testWaitUntilFullyBooted_sdCardReadyFails_notExists(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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, decrypt=False)
def testWaitUntilFullyBooted_devicePmFails(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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, decrypt=False)
def testWaitUntilFullyBooted_bootFails(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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, decrypt=False)
def testWaitUntilFullyBooted_wifiFails(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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, decrypt=False)
def testWaitUntilFullyBooted_decryptFails(self):
with self.assertCalls(
self.call.adb.WaitForDevice(),
# is_device_connection_ready
(self.call.device.GetProp('ro.product.model'), ''),
# 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'),
# decryption_completed
(self.call.device.GetProp('vold.decrypt', cache=False),
'trigger_restart_min_framework'),
# decryption_completed
(self.call.device.GetProp('vold.decrypt', cache=False),
'trigger_restart_min_framework'),
# decryption_completed
(self.call.device.GetProp('vold.decrypt', cache=False),
self.TimeoutError())):
with self.assertRaises(device_errors.CommandTimeoutError):
self.device.WaitUntilFullyBooted(wifi=False, decrypt=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.device.HasRoot(), False),
self.call.adb.Reboot(), (self.call.device.IsOnline(), True),
(self.call.device.IsOnline(), False),
self.call.device.WaitUntilFullyBooted(wifi=False, decrypt=False)):
self.device.Reboot(block=True)
def testReboot_blockingWithRoot(self):
with self.assertCalls(
(self.call.device.HasRoot(), True),
self.call.adb.Reboot(), (self.call.device.IsOnline(), True),
(self.call.device.IsOnline(), False),
self.call.device.WaitUntilFullyBooted(wifi=False, decrypt=False),
self.call.device.EnableRoot()):
self.device.Reboot(block=True)
def testReboot_blockUntilWifi(self):
with self.assertCalls(
(self.call.device.HasRoot(), False),
self.call.adb.Reboot(), (self.call.device.IsOnline(), True),
(self.call.device.IsOnline(), False),
self.call.device.WaitUntilFullyBooted(wifi=True, decrypt=False)):
self.device.Reboot(block=True, wifi=True, decrypt=False)
def testReboot_blockUntilDecrypt(self):
with self.assertCalls(
(self.call.device.HasRoot(), False),
self.call.adb.Reboot(), (self.call.device.IsOnline(), True),
(self.call.device.IsOnline(), False),
self.call.device.WaitUntilFullyBooted(wifi=False, decrypt=True)):
self.device.Reboot(block=True, wifi=False, decrypt=True)
class DeviceUtilsInstallTest(DeviceUtilsTest):
mock_apk = _MockApkHelper(TEST_APK_PATH, TEST_PACKAGE, ['p1'])
def testInstall_noPriorInstall(self):
with self.patch_call(
self.call.device.product_name,
return_value='notflounder'), (self.patch_call(
self.call.device.build_version_sdk, return_value=23)):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
self.call.adb.Install(TEST_APK_PATH,
reinstall=False,
streaming=None,
allow_downgrade=False),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True),
(self.call.device.GrantPermissions(TEST_PACKAGE, ['p1']), [])):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0)
def testInstall_noStreaming(self):
with self.patch_call(
self.call.device.product_name,
return_value='flounder'), (self.patch_call(
self.call.device.build_version_sdk, return_value=23)):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
self.call.adb.Install(TEST_APK_PATH,
reinstall=False,
streaming=False,
allow_downgrade=False),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True),
(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.product_name,
return_value='notflounder'), (self.patch_call(
self.call.device.build_version_sdk, return_value=20)):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
(self.call.adb.Install(TEST_APK_PATH,
reinstall=False,
streaming=None,
allow_downgrade=False)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0)
def testInstall_findPermissions(self):
with self.patch_call(
self.call.device.product_name,
return_value='notflounder'), (self.patch_call(
self.call.device.build_version_sdk, return_value=23)):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
(self.call.adb.Install(TEST_APK_PATH,
reinstall=False,
streaming=None,
allow_downgrade=False)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True),
(self.call.device.GrantPermissions(TEST_PACKAGE, ['p1']), [])):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0)
def testInstall_passPermissions(self):
with self.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
(self.call.adb.Install(TEST_APK_PATH,
reinstall=False,
streaming=None,
allow_downgrade=False)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True),
(self.call.device.GrantPermissions(TEST_PACKAGE, ['p1', 'p2']), [])):
self.device.Install(
DeviceUtilsInstallTest.mock_apk,
retries=0,
permissions=['p1', 'p2'])
def testInstall_identicalPriorInstall(self):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE),
['/fake/data/app/test.package.apk']),
(self.call.device._ComputeStaleApks(TEST_PACKAGE, [TEST_APK_PATH]),
([], None)), (self.call.device.ForceStop(TEST_PACKAGE)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.Install(
DeviceUtilsInstallTest.mock_apk, retries=0, permissions=[])
def testInstall_differentPriorInstall(self):
with self.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE),
['/fake/data/app/test.package.apk']),
(self.call.device._ComputeStaleApks(TEST_PACKAGE, [TEST_APK_PATH]),
([TEST_APK_PATH], None)), self.call.device.Uninstall(TEST_PACKAGE),
self.call.adb.Install(TEST_APK_PATH,
reinstall=False,
streaming=None,
allow_downgrade=False),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.Install(
DeviceUtilsInstallTest.mock_apk, retries=0, permissions=[])
def testInstall_differentPriorInstallSplitApk(self):
with self.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), [
'/fake/data/app/test.package.apk',
'/fake/data/app/test.package2.apk'
]), self.call.device.Uninstall(TEST_PACKAGE),
self.call.adb.Install(TEST_APK_PATH,
reinstall=False,
streaming=None,
allow_downgrade=False),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.Install(
DeviceUtilsInstallTest.mock_apk, retries=0, permissions=[])
def testInstall_differentPriorInstall_reinstall(self):
with self.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE),
['/fake/data/app/test.package.apk']),
(self.call.device._ComputeStaleApks(TEST_PACKAGE, [TEST_APK_PATH]),
([TEST_APK_PATH], None)),
self.call.adb.Install(TEST_APK_PATH,
reinstall=True,
streaming=None,
allow_downgrade=False),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.Install(
DeviceUtilsInstallTest.mock_apk,
reinstall=True,
retries=0,
permissions=[])
def testInstall_identicalPriorInstall_reinstall(self):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE),
['/fake/data/app/test.package.apk']),
(self.call.device._ComputeStaleApks(TEST_PACKAGE, [TEST_APK_PATH]),
([], None)), (self.call.device.ForceStop(TEST_PACKAGE)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.Install(
DeviceUtilsInstallTest.mock_apk,
reinstall=True,
retries=0,
permissions=[])
def testInstall_missingApk(self):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), False)):
with self.assertRaises(device_errors.CommandFailedError):
self.device.Install(DeviceUtilsInstallTest.mock_apk, retries=0)
def testInstall_fails(self):
with self.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
(self.call.adb.Install(
TEST_APK_PATH,
reinstall=False,
streaming=None,
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.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE),
['/fake/data/app/test.package.apk']),
(self.call.device._ComputeStaleApks(TEST_PACKAGE, [TEST_APK_PATH]),
([TEST_APK_PATH], None)),
self.call.adb.Install(TEST_APK_PATH,
reinstall=True,
streaming=None,
allow_downgrade=True),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.Install(
DeviceUtilsInstallTest.mock_apk,
reinstall=True,
retries=0,
permissions=[],
allow_downgrade=True)
def testInstall_pushesFakeModulesToDevice(self):
@contextlib.contextmanager
def mock_zip_temp_dir():
yield '/test/tmp/dir'
mock_apk_with_fake = _MockApkHelper(
TEST_APK_PATH, TEST_PACKAGE, splits=['fake1-master.apk'])
fake_modules = ['fake1']
with self.patch_call(
self.call.device.product_name,
return_value='notflounder'), (self.patch_call(
self.call.device.build_version_sdk, return_value=23)):
with self.assertCalls(
(mock.call.py_utils.tempfile_ext.NamedTemporaryDirectory(),
mock_zip_temp_dir),
self.call.device.RunShellCommand([
'rm', '-rf',
'/sdcard/Android/data/test.package/files/local_testing'
],
as_root=True),
(mock.call.os.rename('fake1-master.apk', '/test/tmp/dir/fake1.apk')),
(self.call.device.PushChangedFiles(
[('/test/tmp/dir', '/data/local/tmp/modules/test.package')],
delete_device_stale=True)),
self.call.device.RunShellCommand([
'mkdir', '-p',
'/sdcard/Android/data/test.package/files/local_testing'
],
as_root=True),
self.call.device.RunShellCommand(
'cp -a /data/local/tmp/modules/test.package/* ' +
'/sdcard/Android/data/test.package/files/local_testing/',
as_root=True,
shell=True),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
self.call.adb.Install(TEST_APK_PATH,
reinstall=False,
streaming=None,
allow_downgrade=False),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True),
(self.call.device.GrantPermissions(TEST_PACKAGE, None), [])):
self.device.Install(
mock_apk_with_fake, fake_modules=fake_modules, retries=0)
def testInstall_packageNotAvailableAfterInstall(self):
with self.patch_call(
self.call.device.product_name,
return_value='notflounder'), (self.patch_call(
self.call.device.build_version_sdk, return_value=23)), (
self.patch_call(self.call.device.IsApplicationInstalled,
return_value=False)):
with self.assertCalls(
(self.call.device._FakeInstall(set(), None, 'test.package')),
(mock.call.os.path.exists(TEST_APK_PATH), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
self.call.adb.Install(TEST_APK_PATH,
reinstall=False,
streaming=None,
allow_downgrade=False)):
with six.assertRaisesRegex(
self, device_errors.CommandFailedError,
'not installed on device after explicit install attempt'):
self.device.Install(
DeviceUtilsInstallTest.mock_apk, retries=0)
class DeviceUtilsInstallSplitApkTest(DeviceUtilsTest):
mock_apk = _MockApkHelper('base.apk', TEST_PACKAGE, ['p1'],
['split1.apk', 'split2.apk'])
def testInstallSplitApk_noPriorInstall(self):
with self.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(mock.call.devil.android.apk_helper.ToSplitHelper(
'base.apk', ['split1.apk', 'split2.apk']),
DeviceUtilsInstallSplitApkTest.mock_apk),
(self.call.device._CheckSdkLevel(21)),
(mock.call.os.path.exists('base.apk'), True),
(mock.call.os.path.exists('split1.apk'), True),
(mock.call.os.path.exists('split2.apk'), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
(self.call.adb.InstallMultiple(
['base.apk', 'split1.apk', 'split2.apk'],
partial=None,
reinstall=False,
streaming=None,
allow_downgrade=False)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.InstallSplitApk(
'base.apk', ['split1.apk', 'split2.apk'], permissions=[], retries=0)
def testInstallSplitApk_noStreaming(self):
with self.patch_call(
self.call.device.product_name, return_value='flounder'):
with self.assertCalls(
(mock.call.devil.android.apk_helper.ToSplitHelper(
'base.apk', ['split1.apk', 'split2.apk']),
DeviceUtilsInstallSplitApkTest.mock_apk),
(self.call.device._CheckSdkLevel(21)),
(mock.call.os.path.exists('base.apk'), True),
(mock.call.os.path.exists('split1.apk'), True),
(mock.call.os.path.exists('split2.apk'), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE), []),
(self.call.adb.InstallMultiple(
['base.apk', 'split1.apk', 'split2.apk'],
partial=None,
reinstall=False,
streaming=False,
allow_downgrade=False)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.InstallSplitApk(
'base.apk', ['split1.apk', 'split2.apk'], permissions=[], retries=0)
def testInstallSplitApk_partialInstall(self):
with self.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(mock.call.devil.android.apk_helper.ToSplitHelper(
DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk']),
DeviceUtilsInstallSplitApkTest.mock_apk),
(self.call.device._CheckSdkLevel(21)),
(mock.call.os.path.exists('base.apk'), True),
(mock.call.os.path.exists('split1.apk'), True),
(mock.call.os.path.exists('split2.apk'), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE),
['base-on-device.apk', 'split2-on-device.apk']),
(self.call.device._ComputeStaleApks(
TEST_PACKAGE, ['base.apk', 'split1.apk', 'split2.apk']),
(['split2.apk'], None)),
(self.call.adb.InstallMultiple(['split2.apk'],
partial=TEST_PACKAGE,
reinstall=True,
streaming=None,
allow_downgrade=False)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.InstallSplitApk(
DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk'],
reinstall=True,
permissions=[],
retries=0)
def testInstallSplitApk_downgrade(self):
with self.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(mock.call.devil.android.apk_helper.ToSplitHelper(
DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk']),
DeviceUtilsInstallSplitApkTest.mock_apk),
(self.call.device._CheckSdkLevel(21)),
(mock.call.os.path.exists('base.apk'), True),
(mock.call.os.path.exists('split1.apk'), True),
(mock.call.os.path.exists('split2.apk'), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE),
['base-on-device.apk', 'split2-on-device.apk']),
(self.call.device._ComputeStaleApks(
TEST_PACKAGE, ['base.apk', 'split1.apk', 'split2.apk']),
(['split2.apk'], None)),
(self.call.adb.InstallMultiple(['split2.apk'],
partial=TEST_PACKAGE,
reinstall=True,
streaming=None,
allow_downgrade=True)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.InstallSplitApk(
DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk'],
reinstall=True,
permissions=[],
retries=0,
allow_downgrade=True)
def testInstallSplitApk_missingSplit(self):
with self.assertCalls(
(mock.call.devil.android.apk_helper.ToSplitHelper(
DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk']),
DeviceUtilsInstallSplitApkTest.mock_apk),
(self.call.device._CheckSdkLevel(21)),
(mock.call.os.path.exists('base.apk'), True),
(mock.call.os.path.exists('split1.apk'), True),
(mock.call.os.path.exists('split2.apk'), False)),\
self.assertRaises(device_errors.CommandFailedError):
self.device.InstallSplitApk(
DeviceUtilsInstallSplitApkTest.mock_apk, ['split1.apk', 'split2.apk'],
permissions=[],
retries=0)
def testInstallSplitApk_previouslyNonSplit(self):
with self.patch_call(
self.call.device.product_name, return_value='notflounder'):
with self.assertCalls(
(mock.call.devil.android.apk_helper.ToSplitHelper(
DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk']),
DeviceUtilsInstallSplitApkTest.mock_apk),
(self.call.device._CheckSdkLevel(21)),
(mock.call.os.path.exists('base.apk'), True),
(mock.call.os.path.exists('split1.apk'), True),
(mock.call.os.path.exists('split2.apk'), True),
(self.call.device._GetApplicationPathsInternal(TEST_PACKAGE),
['/fake/data/app/test.package.apk']),
self.call.device.Uninstall(TEST_PACKAGE),
(self.call.adb.InstallMultiple(
['base.apk', 'split1.apk', 'split2.apk'],
partial=None,
reinstall=False,
streaming=None,
allow_downgrade=False)),
(self.call.device.IsApplicationInstalled(TEST_PACKAGE, None), True)):
self.device.InstallSplitApk(
DeviceUtilsInstallSplitApkTest.mock_apk,
['split1.apk', 'split2.apk'],
permissions=[],
retries=0)
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.assertEqual('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.assertEqual('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'], check_return=True)
def testRunShellCommand_commandAsListQuoted(self):
with self.assertCall(self.call.adb.Shell("echo 'hello world' '$10'"), ''):
self.device.RunShellCommand(['echo', 'hello world', '$10'],
check_return=True)
def testRunShellCommand_commandAsString(self):
with self.assertCall(self.call.adb.Shell('echo "$VAR"'), ''):
self.device.RunShellCommand('echo "$VAR"', shell=True, check_return=True)
def testNewRunShellImpl_withEnv(self):
with self.assertCall(
self.call.adb.Shell('VAR=some_string echo "$VAR"'), ''):
self.device.RunShellCommand(
'echo "$VAR"',
shell=True,
check_return=True,
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'],
check_return=True,
env={'PATH': '$PATH:/other/path'})
def testNewRunShellImpl_withEnv_failure(self):
with self.assertRaises(KeyError):
self.device.RunShellCommand(['some_cmd'],
check_return=True,
env={'INVALID NAME': 'value'})
def testNewRunShellImpl_withCwd(self):
with self.assertCall(self.call.adb.Shell('cd /some/test/path && ls'), ''):
self.device.RunShellCommand(['ls'],
check_return=True,
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'],
check_return=True,
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.assertEqual([payload],
self.device.RunShellCommand(['echo', payload],
check_return=True))
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.assertEqual([payload],
self.device.RunShellCommand(['echo', payload],
check_return=True,
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'],
check_return=True,
as_root=True)
def testRunShellCommand_withRunAs(self):
expected_cmd_without_run_as = "sh -c 'mkdir -p files'"
expected_cmd = (
'run-as org.devil.test_package %s' % expected_cmd_without_run_as)
with self.assertCall(self.call.adb.Shell(expected_cmd), ''):
self.device.RunShellCommand(['mkdir', '-p', 'files'],
check_return=True,
run_as='org.devil.test_package')
def testRunShellCommand_withRunAsAndSu(self):
expected_cmd_with_nothing = "sh -c 'mkdir -p files'"
expected_cmd_with_run_as = (
'run-as org.devil.test_package %s' % expected_cmd_with_nothing)
expected_cmd_without_su = (
'sh -c %s' % cmd_helper.SingleQuote(expected_cmd_with_run_as))
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(['mkdir', '-p', 'files'],
check_return=True,
run_as='org.devil.test_package',
as_root=True)
def testRunShellCommand_manyLines(self):
cmd = 'ls /some/path'
with self.assertCall(self.call.adb.Shell(cmd), 'file1\nfile2\nfile3\n'):
self.assertEqual(['file1', 'file2', 'file3'],
self.device.RunShellCommand(cmd.split(),
check_return=True))
def testRunShellCommand_manyLinesRawOutput(self):
cmd = 'ls /some/path'
with self.assertCall(self.call.adb.Shell(cmd), '\rfile1\nfile2\r\nfile3\n'):
self.assertEqual(
'\rfile1\nfile2\r\nfile3\n',
self.device.RunShellCommand(cmd.split(),
check_return=True,
raw_output=True))
def testRunShellCommand_singleLine_success(self):
cmd = 'echo $VALUE'
with self.assertCall(self.call.adb.Shell(cmd), 'some value\n'):
self.assertEqual(
'some value',
self.device.RunShellCommand(cmd,
shell=True,
check_return=True,
single_line=True))
def testRunShellCommand_singleLine_successEmptyLine(self):
cmd = 'echo $VALUE'
with self.assertCall(self.call.adb.Shell(cmd), '\n'):
self.assertEqual(
'',
self.device.RunShellCommand(cmd,
shell=True,
check_return=True,
single_line=True))
def testRunShellCommand_singleLine_successWithoutEndLine(self):
cmd = 'echo -n $VALUE'
with self.assertCall(self.call.adb.Shell(cmd), 'some value'):
self.assertEqual(
'some value',
self.device.RunShellCommand(cmd,
shell=True,
check_return=True,
single_line=True))
def testRunShellCommand_singleLine_successNoOutput(self):
cmd = 'echo -n $VALUE'
with self.assertCall(self.call.adb.Shell(cmd), ''):
self.assertEqual(
'',
self.device.RunShellCommand(cmd,
shell=True,
check_return=True,
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, shell=True, check_return=True, 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.assertEqual([output.rstrip()],
self.device.RunShellCommand(cmd,
shell=True,
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.split(), 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.assertEqual([output.rstrip()],
self.device.RunShellCommand(cmd.split(),
check_return=False))
def testRunShellCommand_largeOutput_enabled(self):
cmd = 'echo $VALUE'
temp_file = MockTempFile('/sdcard/temp-123')
cmd_redirect = '( %s )>%s 2>&1' % (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, encoding='utf8'), 'something')):
self.assertEqual(['something'],
self.device.RunShellCommand(cmd,
shell=True,
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 2>&1' % (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,
encoding='utf8'), 'something')):
self.assertEqual(['something'],
self.device.RunShellCommand(cmd,
shell=True,
check_return=True))
class DeviceUtilsRunPipedShellCommandTest(DeviceUtilsTest):
def testRunPipedShellCommand_success(self):
with self.assertCall(
self.call.device.RunShellCommand(
'ps | grep foo; echo "PIPESTATUS: ${PIPESTATUS[@]}"',
shell=True,
check_return=True), ['This line contains foo', 'PIPESTATUS: 0 0']):
self.assertEqual(['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[@]}"',
shell=True,
check_return=True), ['PIPESTATUS: 1 0']):
with self.assertRaises(device_errors.AdbShellCommandFailedError) as ec:
self.device._RunPipedShellCommand('ps | grep foo')
self.assertEqual([1, 0], ec.exception.status)
def testRunPipedShellCommand_secondCommandFails(self):
with self.assertCall(
self.call.device.RunShellCommand(
'ps | grep foo; echo "PIPESTATUS: ${PIPESTATUS[@]}"',
shell=True,
check_return=True), ['PIPESTATUS: 0 1']):
with self.assertRaises(device_errors.AdbShellCommandFailedError) as ec:
self.device._RunPipedShellCommand('ps | grep foo')
self.assertEqual([0, 1], ec.exception.status)
def testRunPipedShellCommand_outputCutOff(self):
with self.assertCall(
self.call.device.RunShellCommand(
'ps | grep foo; echo "PIPESTATUS: ${PIPESTATUS[@]}"',
shell=True,
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.ListProcesses('test_process'), []):
with self.assertRaises(device_errors.CommandFailedError):
self.device.KillAll('test_process')
def testKillAll_noMatchingProcessesQuiet(self):
with self.assertCall(self.call.device.ListProcesses('test_process'), []):
self.assertEqual(0, self.device.KillAll('test_process', quiet=True))
def testKillAll_nonblocking(self):
with self.assertCalls((self.call.device.ListProcesses('some.process'),
Processes(('some.process', 1234),
('some.process.thing', 5678))),
(self.call.adb.Shell('kill -9 1234 5678'), '')):
self.assertEqual(2, self.device.KillAll('some.process', blocking=False))
def testKillAll_blocking(self):
with self.assertCalls(
(self.call.device.ListProcesses('some.process'),
Processes(('some.process', 1234), ('some.process.thing', 5678))),
(self.call.adb.Shell('kill -9 1234 5678'), ''),
(self.call.device.ListProcesses('some.process'),
Processes(('some.process.thing', 5678))),
(
self.call.device.ListProcesses('some.process'),
# Other instance with different pid.
Processes(('some.process', 111)))):
self.assertEqual(2, self.device.KillAll('some.process', blocking=True))
def testKillAll_exactNonblocking(self):
with self.assertCalls((self.call.device.ListProcesses('some.process'),
Processes(('some.process', 1234),
('some.process.thing', 5678))),
(self.call.adb.Shell('kill -9 1234'), '')):
self.assertEqual(
1, self.device.KillAll('some.process', exact=True, blocking=False))
def testKillAll_exactBlocking(self):
with self.assertCalls((self.call.device.ListProcesses('some.process'),
Processes(('some.process', 1234),
('some.process.thing', 5678))),
(self.call.adb.Shell('kill -9 1234'), ''),
(self.call.device.ListProcesses('some.process'),
Processes(('some.process', 1234),
('some.process.thing', 5678))),
(self.call.device.ListProcesses('some.process'),
Processes(('some.process.thing', 5678)))):
self.assertEqual(
1, self.device.KillAll('some.process', exact=True, blocking=True))
def testKillAll_root(self):
with self.assertCalls(
(self.call.device.ListProcesses('some.process'),
Processes(('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.assertEqual(1, self.device.KillAll('some.process', as_root=True))
def testKillAll_sigterm(self):
with self.assertCalls((self.call.device.ListProcesses('some.process'),
Processes(('some.process', 1234))),
(self.call.adb.Shell('kill -15 1234'), '')):
self.assertEqual(
1, self.device.KillAll('some.process', signum=device_signal.SIGTERM))
def testKillAll_multipleInstances(self):
with self.assertCalls((self.call.device.ListProcesses('some.process'),
Processes(('some.process', 1234),
('some.process', 4567))),
(self.call.adb.Shell('kill -15 1234 4567'), '')):
self.assertEqual(
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=[
intent.FLAG_ACTIVITY_NEW_TASK,
intent.FLAG_ACTIVITY_RESET_TASK_IF_NEEDED
])
with self.assertCall(
self.call.adb.Shell('am start '
'-a android.intent.action.VIEW '
'-n test.package/.Main '
'-f 0x10200000'),
'Starting: Intent { act=android.intent.action.VIEW }'):
self.device.StartActivity(test_intent)
class DeviceUtilsStartServiceTest(DeviceUtilsTest):
def testStartService_success(self):
test_intent = intent.Intent(
action='android.intent.action.START',
package=TEST_PACKAGE,
activity='.Main')
with self.patch_call(
self.call.device.build_version_sdk, return_value=version_codes.NOUGAT):
with self.assertCall(
self.call.adb.Shell('am startservice '
'-a android.intent.action.START '
'-n test.package/.Main'),
'Starting service: Intent { act=android.intent.action.START }'):
self.device.StartService(test_intent)
def testStartService_failure(self):
test_intent = intent.Intent(
action='android.intent.action.START',
package=TEST_PACKAGE,
activity='.Main')
with self.patch_call(
self.call.device.build_version_sdk, return_value=version_codes.NOUGAT):
with self.assertCall(
self.call.adb.Shell('am startservice '
'-a android.intent.action.START '
'-n test.package/.Main'),
'Error: Failed to start test service'):
with self.assertRaises(device_errors.CommandFailedError):
self.device.StartService(test_intent)
def testStartService_withUser(self):
test_intent = intent.Intent(
action='android.intent.action.START',
package=TEST_PACKAGE,
activity='.Main')
with self.patch_call(
self.call.device.build_version_sdk, return_value=version_codes.NOUGAT):
with self.assertCall(
self.call.adb.Shell('am startservice '
'--user TestUser '
'-a android.intent.action.START '
'-n test.package/.Main'),
'Starting service: Intent { act=android.intent.action.START }'):
self.device.StartService(test_intent, user_id='TestUser')
def testStartService_onOreo(self):
test_intent = intent.Intent(
action='android.intent.action.START',
package=TEST_PACKAGE,
activity='.Main')
with self.patch_call(
self.call.device.build_version_sdk, return_value=version_codes.OREO):
with self.assertCall(
self.call.adb.Shell('am start-service '
'-a android.intent.action.START '
'-n test.package/.Main'),
'Starting service: Intent { act=android.intent.action.START }'):
self.device.StartService(test_intent)
class DeviceUtilsStartInstrumentationTest(DeviceUtilsTest):
def testStartInstrumentation_nothing(self):
with self.assertCalls(
self.call.device.RunShellCommand(
'p=test.package;am instrument "$p"/.TestInstrumentation',
shell=True,
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',
shell=True,
check_return=True,
large_output=True), ['OK (1 test)'])):
output = self.device.StartInstrumentation(
'test.package/.TestInstrumentation',
finish=True,
raw=False,
extras=None)
self.assertEqual(['OK (1 test)'], output)
def testStartInstrumentation_raw(self):
with self.assertCalls(
self.call.device.RunShellCommand(
'p=test.package;am instrument -r "$p"/.TestInstrumentation',
shell=True,
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',
shell=True,
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', 'activity', 'activities'],
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', 'activity', 'activities'],
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', 'activity', 'activities'],
check_return=True,
large_output=True),
['mResumedActivity Launcher'])):
self.device.GoHome()
def testGoHome_willRetry(self):
with self.assertCalls(
(self.call.device.RunShellCommand(['dumpsys', 'activity', 'activities'],
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', 'activity', 'activities'],
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', 'activity', 'activities'],
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', 'activity', 'activities'],
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', 'activity', 'activities'],
check_return=True,
large_output=True),
['mResumedActivity Launcher']):
self.device.GoHome()
def testGoHome_alreadyFocusedAlternateCase(self):
with self.assertCall(
self.call.device.RunShellCommand(['dumpsys', 'activity', 'activities'],
check_return=True,
large_output=True),
[' mResumedActivity .launcher/.']):
self.device.GoHome()
def testGoHome_obtainsFocusAfterGoingHome(self):
with self.assertCalls(
(self.call.device.RunShellCommand(['dumpsys', 'activity', 'activities'],
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', 'activity', 'activities'],
check_return=True,
large_output=True), ['mResumedActivity Launcher'])):
self.device.GoHome()
class DeviceUtilsForceStopTest(DeviceUtilsTest):
def testForceStop(self):
with self.assertCalls(
(self.call.device.GetApplicationPids(TEST_PACKAGE), [1111]),
(self.call.device.RunShellCommand(['am', 'force-stop', TEST_PACKAGE],
check_return=True), ['Success'])):
self.device.ForceStop(TEST_PACKAGE)
def testForceStop_NoProcessFound(self):
with self.assertCall(self.call.device.GetApplicationPids(TEST_PACKAGE), []):
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')]
with self.assertCalls((self.call.device._MaybeInstallCommands(), False)):
self.assertFalse(
self.device._PushChangedFilesZipped(test_files, ['/test/dir']))
def _testPushChangedFilesZipped_spec(self, test_files, test_dirs):
@contextlib.contextmanager
def mock_zip_temp_dir():
yield '/test/temp/dir'
expected_cmd = ''.join([
'\n /data/local/tmp/bin/unzip %s &&',
' (for dir in %s\n do\n chmod -R 777 "$dir" || exit 1\n',
' done)\n'
]) % ('/sdcard/foo123.zip', ' '.join(test_dirs))
with self.assertCalls(
(self.call.device._MaybeInstallCommands(), True),
(mock.call.py_utils.tempfile_ext.NamedTemporaryDirectory(),
mock_zip_temp_dir), (mock.call.devil.utils.zip_utils.WriteZipFile(
'/test/temp/dir/tmp.zip', test_files)),
(mock.call.os.path.getsize('/test/temp/dir/tmp.zip'), 123),
(self.call.device.NeedsSU(), True),
(mock.call.devil.android.device_temp_file.DeviceTempFile(
self.adb, suffix='.zip'), MockTempFile('/sdcard/foo123.zip')),
self.call.adb.Push('/test/temp/dir/tmp.zip', '/sdcard/foo123.zip'),
(mock.call.devil.android.device_temp_file.DeviceTempFile(
self.adb, suffix='.sh'), MockTempFile('/sdcard/temp-123.sh')),