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
|
# Copyright 2015 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
import re
# http://developer.android.com/reference/android/test/InstrumentationTestRunner.html
STATUS_CODE_START = 1
STATUS_CODE_OK = 0
STATUS_CODE_ERROR = -1
STATUS_CODE_FAILURE = -2
# AndroidJUnitRunner would status output -3 to indicate a test is skipped
STATUS_CODE_SKIP = -3
# AndroidJUnitRunner outputs -4 to indicate a failed assumption
# "A test for which an assumption fails should not generate a test
# case failure"
# http://junit.org/junit4/javadoc/4.12/org/junit/AssumptionViolatedException.html
STATUS_CODE_ASSUMPTION_FAILURE = -4
STATUS_CODE_TEST_DURATION = 1337
# When a test batch fails due to post-test Assertion failures (eg.
# LifetimeAssert).
STATUS_CODE_BATCH_FAILURE = 1338
# http://developer.android.com/reference/android/app/Activity.html
RESULT_CODE_OK = -1
RESULT_CODE_CANCELED = 0
_INSTR_LINE_RE = re.compile(r'^\s*INSTRUMENTATION_([A-Z_]+): (.*)$')
class InstrumentationParser:
def __init__(self, stream):
"""An incremental parser for the output of Android instrumentation tests.
Example:
stream = adb.IterShell('am instrument -r ...')
parser = InstrumentationParser(stream)
for code, bundle in parser.IterStatus():
# do something with each instrumentation status
print('status:', code, bundle)
# do something with the final instrumentation result
code, bundle = parser.GetResult()
print('result:', code, bundle)
Args:
stream: a sequence of lines as produced by the raw output of an
instrumentation test (e.g. by |am instrument -r|).
"""
self._stream = stream
self._code = None
self._bundle = None
def IterStatus(self):
"""Iterate over statuses as they are produced by the instrumentation test.
Yields:
A tuple (code, bundle) for each instrumentation status found in the
output.
"""
def join_bundle_values(bundle):
for key in bundle:
bundle[key] = '\n'.join(bundle[key])
return bundle
bundle = {'STATUS': {}, 'RESULT': {}}
header = None
key = None
for line in self._stream:
m = _INSTR_LINE_RE.match(line)
if m:
header, value = m.groups()
key = None
if header in ['STATUS', 'RESULT'] and '=' in value:
key, value = value.split('=', 1)
bundle[header][key] = [value]
elif header == 'STATUS_CODE':
yield int(value), join_bundle_values(bundle['STATUS'])
bundle['STATUS'] = {}
elif header == 'CODE':
self._code = int(value)
else:
logging.warning('Unknown INSTRUMENTATION_%s line: %s', header, value)
elif key is not None:
bundle[header][key].append(line)
self._bundle = join_bundle_values(bundle['RESULT'])
def GetResult(self):
"""Return the final instrumentation result.
Returns:
A pair (code, bundle) with the final instrumentation result. The |code|
may be None if no instrumentation result was found in the output.
Raises:
AssertionError if attempting to get the instrumentation result before
exhausting |IterStatus| first.
"""
assert self._bundle is not None, (
'The IterStatus generator must be exhausted before reading the final'
' instrumentation result.')
return self._code, self._bundle
|
