1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
# 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):
self.timeout = timeout
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:
raise TimeoutException(timeout)
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
|
