aboutsummaryrefslogtreecommitdiff
path: root/lava_android_test/test_definitions/cts/cts_wrapper.py
blob: fe6b4a18b4668858c4132f26eb30edc5134d709c (plain)
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
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
#!/usr/bin/python

# Copyright (c) 2012 Linaro

# Author: Linaro Validation Team <linaro-dev@lists.linaro.org>
#
# This file is part of LAVA Android Test.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os
import re
import sys
import pexpect
import time
import xml.dom.minidom
from zipfile import ZipFile

from lava_android_test.adb import ADB
from lava_android_test.utils import stop_at_pattern
from lava_android_test.utils import find_files

adb = ADB(sys.argv[1])
curdir = os.path.realpath(os.path.dirname(__file__))


def stop_at_cts_pattern(command=None, pattern=None, timeout=-1):
    if not command:
        return

    if not pattern:
        response = [pexpect.EOF]
    else:
        response = [pattern, pexpect.EOF]

    result = True
    proc_cts = pexpect.spawn(command, logfile=sys.stdout)
    time.sleep(200)
    try:
        match_id = proc_cts.expect(response, timeout=timeout)
        if match_id == 0:
            time.sleep(5)
    except pexpect.TIMEOUT:
        result = False
    finally:
        proc_cts.sendcontrol('C')
        proc_cts.sendline('')

        target_dir = os.path.join(os.getcwd(),
                                  './android-cts/repository/results/')
        for zip_f in find_files(target_dir, '.zip'):
            ret_code = adb.push(zip_f, '/data/local/tmp/cts-results.zip')[0]
            if ret_code != 0:
                print "Failed to push file %s to device(%s)" % (zip_f,
                                                           adb.get_serial())
        log_target_dir = os.path.join(os.getcwd(),
                                  './android-cts/repository/logs/')
        for zip_f in find_files(log_target_dir, '.zip'):
            base_name = os.path.basename(zip_f)
            if base_name.startswith('device_logcat_'):
                base_name = 'device_logcat.zip'
            if base_name.startswith('host_log_'):
                base_name = 'host_log.zip'

            ret_code = adb.push(zip_f, '/data/local/tmp/%s' % base_name)[0]
            if ret_code != 0:
                print "Failed to push file %s to device(%s)" % (zip_f,
                                                           adb.get_serial())
#                with ZipFile(zip_f) as log_fd:
#                    print '=========Log file [%s] starts=========>>>>>' % (
#                                                                       base_name)
#                    f_name = base_name.replace('.zip', '.txt')
#                    for line in log_fd.open(f_name).readlines():
#                        print line.rstrip()
#                    print '<<<<<=========Log file [%s] ends=========' % base_name
    return result


def get_not_executed():
    list_result_path = os.path.join(curdir, 'cts_list_result_wrapper.sh')
    list_result_cmd = "bash %s" % list_result_path

    pattern = 'All done'
    if not stop_at_pattern(command=list_result_cmd,
                            pattern=pattern, timeout=60):
        print "Failed to list the cts result for device(%s)" % adb.get_serial()

    with open('cts_list_results.log') as fd:
        #0        17237  126   0     2012.06.23_03.31.49  CTS        unknown
        pattern = ("^\s*\d+\s+\d+\s+\d+\s+(?P<no_executed>\d+)"
                   "\s+\S+\s+\S+\s+unknown\s*$")
        pat = re.compile(pattern)
        for line in fd.readlines():
            match = pat.search(line)
            if not match:
                print line.rstrip()
                continue
            return match.groupdict()['no_executed']
        return 0


def prepare_cts():
    cts_prepare_path = os.path.join(curdir, 'cts_prepare.sh')
    cts_helper_jar_path = os.path.join(curdir, 'ctshelper.jar')
    cts_prepare_cmd = "bash %s" % cts_prepare_path
    if not stop_at_pattern(command="%s %s %s" % (cts_prepare_cmd,
                                adb.get_serial(), cts_helper_jar_path),
                           timeout=18000):
        print "Preapration for CTS test times out"
        return False
    return True


def run_cts_with_plan(cts_cmd=None, plan='CTS', timeout=36000):
    pattern = "Time:"
    plan_command = '--plan %s' % plan
    if cts_cmd:
        plan_command = "%s %s --disable-reboot" % (cts_cmd, plan_command)
    if not stop_at_cts_pattern(command=plan_command, pattern=pattern,
                            timeout=timeout):
        print "CTS test times out"
        return False

    return True


def run_cts_with_package(cts_cmd=None, package=None, timeout=36000):
    if not package:
        return True
    pattern = "Time:"
    plan_command = '--package %s' % package
    if cts_cmd:
        plan_command = "%s %s --disable-reboot" % (cts_cmd, plan_command)
    if not stop_at_cts_pattern(command=plan_command, pattern=pattern,
                            timeout=timeout):
        print "CTS test times out"
        return False

    return True


def run_cts_with_class(cts_cmd=None, cls=None, method=None, timeout=36000):
    if not cls:
        return True
    pattern = "Time:"
    cmd = '--class %s' % cls
    if method:
        cmd = '%s --method %s' % (cmd, method)

    if cts_cmd:
        cmd = "%s %s --disable-reboot" % (cts_cmd, cmd)
    if not stop_at_cts_pattern(command=cmd, pattern=pattern,
                            timeout=timeout):
        print "CTS test times out"
        return False

    return True


def run_cts_continue(cts_cmd=None, timeout=36000):
    pattern = "Time:"
    continue_command = '--continue-session 0'
    if cts_cmd:
        continue_command = "%s %s" % (cts_cmd, continue_command)

    while True:
        number_of_not_executed = get_not_executed()
        if number_of_not_executed and int(number_of_not_executed) > 0:
            print ('Reconnect the adb connection before continuing '
                   'the CTS on device(%s)') % adb.get_serial()
            if not adb.reconnect():
                print "Faile to reconnect the adb connection of device(%s)" % (
                                                             adb.get_serial())
                break

            print "Continue the uncompleted CTS test on device(%s)" % (
                                                           adb.get_serial())

            if not stop_at_cts_pattern(command=continue_command,
                                   pattern=pattern,
                                   timeout=timeout):
                print "CTS test times out"
        else:
            break


def collect_log(command=None, output_file=None):
    if command and output_file:
        print 'Redirect the output of command[%s] to file[%s]' % (command,
                                                                  output_file)
        cmd = 'bash %s %s "%s"' % (os.path.join(curdir, 'cts_redirect.sh'),
                                 output_file, command)
        stdout = adb.run_cmd_host(cmd)[1]
        if stdout:
            return stdout[0].strip()

    return None


def collect_logs():

    kmsg = {'command':
                    'adb -s %s shell cat /proc/kmsg' % (adb.get_serial()),
            'output_file': 'kmsg.log'}

    logcat = {'command':
                'adb -s %s logcat -c; adb -s %s logcat -v time' % (
                                    adb.get_serial(), adb.get_serial()),
              'output_file': 'logcat.log'}

    ## define all the logs need to be collected
    logs = [kmsg, logcat]
    for log in logs:
        pid = collect_log(command=log.get('command'),
                         output_file=log.get('output_file'))
        if pid:
            log['pid'] = pid
    return logs


def push_log(logs=[]):
    for log in logs:
        log_file = log.get('output_file')
        base_name = os.path.basename(log_file)
        if log_file:
            ret_code = adb.push(log_file, '/data/local/tmp/%s' % base_name)[0]
            if ret_code != 0:
                print "Failed to push file %s to device(%s)" % (log_file,
                                                           adb.get_serial())
                with open(log_file) as log_fd:
                    print '=========Log file [%s] starts=========>>>>>' % (
                                                                       log_file)
                    for line in log_fd.readlines():
                        print line.rstrip()
                    print '<<<<<=========Log file [%s] ends=========' % log_file


def get_all_packages(plan_file=None):
    if not plan_file:
        return []
    if not os.path.exists(plan_file):
        print "file(%s) does not exist" % plan_file
        return []

    package_list = []
    try:
        dom = xml.dom.minidom.parse(plan_file)
        test_plan = dom.getElementsByTagName("TestPlan")[0]
        for entry in test_plan.getElementsByTagName("Entry"):
            package_list.append(entry.attributes.get('uri').value)
    except Exception as e:
        print "Has exception to parse the xml file"
        print "Exception: %s" % e
    finally:
        return package_list


def get_value_from_paras(paras=[], option=None, default=None):
    if not option:
        return default

    if not option in paras:
        return default

    index = paras.index(option)
    if len(paras) > index + 1:
        return paras[index + 1]

    return default


def main():

    package_name = None
    plan_name = 'CTS'
    class_name = None
    method_name = None
    timeout = 36000
    force_abi = None
    #--cts_pkg cts_package_file --package package_name --timeout 36000
    #--cts_pkg cts_package_file --plan plan_name --timeout 36000
    if len(sys.argv) > 2:
        paras = sys.argv[2:]
        cts_pkg = get_value_from_paras(paras=paras, option='--cts-pkg')
        if cts_pkg:
            os.environ["cts_pkg"] = cts_pkg

        java_home = get_value_from_paras(paras=paras, option='--java-home')
        if java_home:
            os.environ["PATH"] = java_home + "/bin" + os.pathsep + java_home \
                                      + "/jre/bin" + os.pathsep + os.environ["PATH"]
            os.environ["JAVA_HOME"] = java_home

        package_name = get_value_from_paras(paras=paras, option='--package')
        plan_name = get_value_from_paras(paras=paras,
                                         option='--plan',
                                         default='CTS')
        timeout = get_value_from_paras(paras=paras, option='--timeout',
                                       default=36000)
        if timeout:
            timeout = int(timeout)

        class_name = get_value_from_paras(paras=paras, option='--class')
        method_name = get_value_from_paras(paras=paras, option='--method')

        force_abi = get_value_from_paras(paras=paras, option='--force-abi')

    run_wrapper_path = os.path.join('./android-cts/tools/cts-tradefed ')
    run_wrapper_cmd = "%s" % run_wrapper_path
    run_wrapper_cmd = '%s run cts --serial %s' % (run_wrapper_cmd,
                                                      adb.get_serial())

    if force_abi:
        run_wrapper_cmd = '%s --force-abi %s' % (run_wrapper_cmd,
                                                      force_abi)
    print "run_wrapper_cmd=%s" % run_wrapper_cmd
    logs = collect_logs()
    if not prepare_cts():
        sys.exit(1)

    try:
        if package_name:
            run_cts_with_package(cts_cmd=run_wrapper_cmd, package=package_name,
                                 timeout=timeout)
        elif class_name:
            run_cts_with_class(cts_cmd=run_wrapper_cmd, cls=class_name,
                               method=method_name, timeout=timeout)
        else:
            run_cts_with_plan(cts_cmd=run_wrapper_cmd, plan=plan_name,
                              timeout=timeout)
        run_cts_continue(cts_cmd=run_wrapper_cmd, timeout=timeout)

    finally:
        for log in logs:
            pid = log.get('pid')
            if pid:
                adb.run_cmd_host('kill -9 %s' % pid)

        push_log(logs)

    sys.exit(0)


if __name__ == '__main__':
    main()