# Copyright 2018 the V8 project authors. All rights reserved. # Use of this source code is governed by a BSD-style license that can be # found in the LICENSE file. """ Wrapper around the Android device abstraction from src/build/android. """ import logging import os import sys BASE_DIR = os.path.normpath( os.path.join(os.path.dirname(__file__), '..', '..', '..')) ANDROID_DIR = os.path.join(BASE_DIR, 'build', 'android') DEVICE_DIR = '/data/local/tmp/v8/' class TimeoutException(Exception): def __init__(self, timeout, output=None): self.timeout = timeout self.output = output class CommandFailedException(Exception): def __init__(self, status, output): self.status = status self.output = output class _Driver(object): """Helper class to execute shell commands on an Android device.""" def __init__(self, device=None): assert os.path.exists(ANDROID_DIR) sys.path.insert(0, ANDROID_DIR) # We import the dependencies only on demand, so that this file can be # imported unconditionally. import devil_chromium from devil.android import device_errors # pylint: disable=import-error from devil.android import device_utils # pylint: disable=import-error from devil.android.perf import cache_control # pylint: disable=import-error from devil.android.perf import perf_control # pylint: disable=import-error from devil.android.sdk import adb_wrapper # pylint: disable=import-error global cache_control global device_errors global perf_control devil_chromium.Initialize() if not device: # Detect attached device if not specified. devices = adb_wrapper.AdbWrapper.Devices() assert devices, 'No devices detected' assert len(devices) == 1, 'Multiple devices detected.' device = str(devices[0]) self.adb_wrapper = adb_wrapper.AdbWrapper(device) self.device = device_utils.DeviceUtils(self.adb_wrapper) # This remembers what we have already pushed to the device. self.pushed = set() def tear_down(self): """Clean up files after running all tests.""" self.device.RemovePath(DEVICE_DIR, force=True, recursive=True) def push_file(self, host_dir, file_name, target_rel='.', skip_if_missing=False): """Push a single file to the device (cached). Args: host_dir: Absolute parent directory of the file to push. file_name: Name of the file to push. target_rel: Parent directory of the target location on the device (relative to the device's base dir for testing). skip_if_missing: Keeps silent about missing files when set. Otherwise logs error. """ file_on_host = os.path.join(host_dir, file_name) # Only push files not yet pushed in one execution. if file_on_host in self.pushed: return file_on_device_tmp = os.path.join(DEVICE_DIR, '_tmp_', file_name) file_on_device = os.path.join(DEVICE_DIR, target_rel, file_name) folder_on_device = os.path.dirname(file_on_device) # Only attempt to push files that exist. if not os.path.exists(file_on_host): if not skip_if_missing: logging.critical('Missing file on host: %s' % file_on_host) return # Work-around for 'text file busy' errors. Push the files to a temporary # location and then copy them with a shell command. output = self.adb_wrapper.Push(file_on_host, file_on_device_tmp) # Success looks like this: '3035 KB/s (12512056 bytes in 4.025s)'. # Errors look like this: 'failed to copy ... '. if output and not re.search('^[0-9]', output.splitlines()[-1]): logging.critical('PUSH FAILED: ' + output) self.adb_wrapper.Shell('mkdir -p %s' % folder_on_device) self.adb_wrapper.Shell('cp %s %s' % (file_on_device_tmp, file_on_device)) self.pushed.add(file_on_host) def push_executable(self, shell_dir, target_dir, binary): """Push files required to run a V8 executable. Args: shell_dir: Absolute parent directory of the executable on the host. target_dir: Parent directory of the executable on the device (relative to devices' base dir for testing). binary: Name of the binary to push. """ self.push_file(shell_dir, binary, target_dir) # Push external startup data. Backwards compatible for revisions where # these files didn't exist. Or for bots that don't produce these files. self.push_file( shell_dir, 'natives_blob.bin', target_dir, skip_if_missing=True, ) self.push_file( shell_dir, 'snapshot_blob.bin', target_dir, skip_if_missing=True, ) self.push_file( shell_dir, 'snapshot_blob_trusted.bin', target_dir, skip_if_missing=True, ) self.push_file( shell_dir, 'icudtl.dat', target_dir, skip_if_missing=True, ) def run(self, target_dir, binary, args, rel_path, timeout, env=None, logcat_file=False): """Execute a command on the device's shell. Args: target_dir: Parent directory of the executable on the device (relative to devices' base dir for testing). binary: Name of the binary. args: List of arguments to pass to the binary. rel_path: Relative path on device to use as CWD. timeout: Timeout in seconds. env: The environment variables with which the command should be run. logcat_file: File into which to stream adb logcat log. """ binary_on_device = os.path.join(DEVICE_DIR, target_dir, binary) cmd = [binary_on_device] + args def run_inner(): try: output = self.device.RunShellCommand( cmd, cwd=os.path.join(DEVICE_DIR, rel_path), check_return=True, env=env, timeout=timeout, retries=0, ) return '\n'.join(output) except device_errors.AdbCommandFailedError as e: raise CommandFailedException(e.status, e.output) except device_errors.CommandTimeoutError as e: raise TimeoutException(timeout, e.output) if logcat_file: with self.device.GetLogcatMonitor(output_file=logcat_file) as logmon: result = run_inner() logmon.Close() return result else: return run_inner() def drop_ram_caches(self): """Drop ran caches on device.""" cache = cache_control.CacheControl(self.device) cache.DropRamCaches() def set_high_perf_mode(self): """Set device into high performance mode.""" perf = perf_control.PerfControl(self.device) perf.SetHighPerfMode() def set_default_perf_mode(self): """Set device into default performance mode.""" perf = perf_control.PerfControl(self.device) perf.SetDefaultPerfMode() _ANDROID_DRIVER = None def android_driver(device=None): """Singleton access method to the driver class.""" global _ANDROID_DRIVER if not _ANDROID_DRIVER: _ANDROID_DRIVER = _Driver(device) return _ANDROID_DRIVER