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
|
#!/usr/bin/env python
"""
Test that aidl generates functional code by running it on an Android device.
"""
import argparse
import pipes
import subprocess
import shlex
JAVA_OUTPUT_READER = 'aidl_test_sentinel_searcher'
NATIVE_TEST_CLIENT = 'aidl_test_client'
NATIVE_TEST_SERVICE = 'aidl_test_service'
TEST_FILTER_ALL = 'all'
TEST_FILTER_JAVA = 'java'
TEST_FILTER_NATIVE = 'native'
JAVA_CLIENT_TIMEOUT_SECONDS = 30
JAVA_LOG_FILE = '/data/data/android.aidl.tests/files/test-client.log'
JAVA_SUCCESS_SENTINEL = '>>> Java Client Success <<<'
JAVA_FAILURE_SENTINEL = '>>> Java Client Failure <<<'
class TestFail(Exception):
"""Raised on test failures."""
pass
class ShellResult(object):
"""Represents the result of running a shell command."""
def __init__(self, exit_status, stdout, stderr):
"""Construct an instance.
Args:
exit_status: integer exit code of shell command
stdout: string stdout of shell command
stderr: string stderr of shell command
"""
self.stdout = stdout
self.stderr = stderr
self.exit_status = exit_status
def printable_string(self):
"""Get a string we could print to the logs and understand."""
output = []
output.append('stdout:')
for line in self.stdout.splitlines():
output.append(' > %s' % line)
output.append('stderr:')
for line in self.stderr.splitlines():
output.append(' > %s' % line)
return '\n'.join(output)
class AdbHost(object):
"""Represents a device connected via ADB."""
def __init__(self, device_serial=None, verbose=None):
"""Construct an instance.
Args:
device_serial: options string serial number of attached device.
verbose: True iff we should print out ADB commands we run.
"""
self._device_serial = device_serial
self._verbose = verbose
def run(self, command, background=False, ignore_status=False):
"""Run a command on the device via adb shell.
Args:
command: string containing a shell command to run.
background: True iff we should run this command in the background.
ignore_status: True iff we should ignore the command's exit code.
Returns:
instance of ShellResult.
Raises:
subprocess.CalledProcessError on command exit != 0.
"""
if background:
command = '( %s ) </dev/null >/dev/null 2>&1 &' % command
return self.adb('shell %s' % pipes.quote(command),
ignore_status=ignore_status)
def mktemp(self):
"""Make a temp file on the device.
Returns:
path to created file as a string
Raises:
subprocess.CalledProcessError on failure.
"""
# Work around b/19635681
result = self.run('source /system/etc/mkshrc && mktemp')
return result.stdout.strip()
def adb(self, command, ignore_status=False):
"""Run an ADB command (e.g. `adb sync`).
Args:
command: string containing command to run
ignore_status: True iff we should ignore the command's exit code.
Returns:
instance of ShellResult.
Raises:
subprocess.CalledProcessError on command exit != 0.
"""
command = 'adb %s' % command
if self._verbose:
print(command)
p = subprocess.Popen(command, shell=True, close_fds=True,
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=True)
stdout, stderr = p.communicate()
if not ignore_status and p.returncode:
raise subprocess.CalledProcessError(p.returncode, command)
return ShellResult(p.returncode, stdout, stderr)
def run_test(test_native, test_java, apk_path=None, refresh_binaries=False,
device_serial=None, verbose=False):
"""Body of the test.
Args:
test_native: True iff we should test native Binder clients.
test_java: True iff we shoudl test Java Binder clients.
apk_path: Optional path to an APK to install via `adb install`
refresh_binaries: True iff we should `adb sync` new binaries to the
device.
device_serial: Optional string containing the serial number of the
device under test.
verbose: True iff we should enable verbose output during the test.
"""
print('Starting aidl integration testing...')
host = AdbHost(device_serial=device_serial, verbose=verbose)
if apk_path is not None:
host.adb('install -r %s' % apk_path)
if refresh_binaries:
host.adb('remount')
host.adb('sync')
host.run('setenforce 0')
# Kill any previous test context
host.run('rm -f %s' % JAVA_LOG_FILE, ignore_status=True)
host.run('pkill %s' % NATIVE_TEST_SERVICE, ignore_status=True)
# Start up a native server
host.run(NATIVE_TEST_SERVICE, background=True)
# Start up clients
if test_native:
host.run('pkill %s' % NATIVE_TEST_CLIENT, ignore_status=True)
result = host.run(NATIVE_TEST_CLIENT, ignore_status=True)
if result.exit_status:
print(result.printable_string())
raise TestFail('%s returned status code %d' %
(NATIVE_TEST_CLIENT, result.exit_status))
if test_java:
host.run('am start -S -a android.intent.action.MAIN '
'-n android.aidl.tests/.TestServiceClient '
'--es sentinel.success "%s" '
'--es sentinel.failure "%s"' %
(JAVA_SUCCESS_SENTINEL, JAVA_FAILURE_SENTINEL))
result = host.run('%s %d %s "%s" "%s"' %
(JAVA_OUTPUT_READER, JAVA_CLIENT_TIMEOUT_SECONDS,
JAVA_LOG_FILE, JAVA_SUCCESS_SENTINEL,
JAVA_FAILURE_SENTINEL),
ignore_status=True)
if result.exit_status:
print(result.printable_string())
raise TestFail('Java client did not complete successfully.')
print('Success!')
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--apk', dest='apk_path', type=str, default=None,
help='Path to an APK to install on the device.')
parser.add_argument('--refresh-bins', action='store_true', default=False,
help='Pass this flag to have the test run adb sync')
parser.add_argument('--serial', '-s', type=str, default=None,
help='Serial number of device to test against')
parser.add_argument(
'--test-filter', default=TEST_FILTER_ALL,
choices=[TEST_FILTER_ALL, TEST_FILTER_JAVA, TEST_FILTER_NATIVE])
parser.add_argument('--verbose', '-v', action='store_true', default=False)
args = parser.parse_args()
run_test(args.test_filter in (TEST_FILTER_ALL, TEST_FILTER_NATIVE),
args.test_filter in (TEST_FILTER_ALL, TEST_FILTER_JAVA),
apk_path=args.apk_path, refresh_binaries=args.refresh_bins,
device_serial=args.serial, verbose=args.verbose)
if __name__ == '__main__':
main()
|
