xref: /aosp_15_r20/external/angle/build/android/pylib/instrumentation/instrumentation_parser.py (revision 8975f5c5ed3d1c378011245431ada316dfb6f244) 
1# Copyright 2015 The Chromium Authors
2# Use of this source code is governed by a BSD-style license that can be
3# found in the LICENSE file.
4
5
6import logging
7import re
8
9# http://developer.android.com/reference/android/test/InstrumentationTestRunner.html
10STATUS_CODE_START = 1
11STATUS_CODE_OK = 0
12STATUS_CODE_ERROR = -1
13STATUS_CODE_FAILURE = -2
14
15# AndroidJUnitRunner would status output -3 to indicate a test is skipped
16STATUS_CODE_SKIP = -3
17
18# AndroidJUnitRunner outputs -4 to indicate a failed assumption
19# "A test for which an assumption fails should not generate a test
20# case failure"
21# http://junit.org/junit4/javadoc/4.12/org/junit/AssumptionViolatedException.html
22STATUS_CODE_ASSUMPTION_FAILURE = -4
23
24STATUS_CODE_TEST_DURATION = 1337
25
26# When a test batch fails due to post-test Assertion failures (eg.
27# LifetimeAssert).
28STATUS_CODE_BATCH_FAILURE = 1338
29
30# http://developer.android.com/reference/android/app/Activity.html
31RESULT_CODE_OK = -1
32RESULT_CODE_CANCELED = 0
33
34_INSTR_LINE_RE = re.compile(r'^\s*INSTRUMENTATION_([A-Z_]+): (.*)$')
35
36
37class InstrumentationParser:
38
39  def __init__(self, stream):
40    """An incremental parser for the output of Android instrumentation tests.
41
42    Example:
43
44      stream = adb.IterShell('am instrument -r ...')
45      parser = InstrumentationParser(stream)
46
47      for code, bundle in parser.IterStatus():
48        # do something with each instrumentation status
49        print('status:', code, bundle)
50
51      # do something with the final instrumentation result
52      code, bundle = parser.GetResult()
53      print('result:', code, bundle)
54
55    Args:
56      stream: a sequence of lines as produced by the raw output of an
57        instrumentation test (e.g. by |am instrument -r|).
58    """
59    self._stream = stream
60    self._code = None
61    self._bundle = None
62
63  def IterStatus(self):
64    """Iterate over statuses as they are produced by the instrumentation test.
65
66    Yields:
67      A tuple (code, bundle) for each instrumentation status found in the
68      output.
69    """
70    def join_bundle_values(bundle):
71      for key in bundle:
72        bundle[key] = '\n'.join(bundle[key])
73      return bundle
74
75    bundle = {'STATUS': {}, 'RESULT': {}}
76    header = None
77    key = None
78    for line in self._stream:
79      m = _INSTR_LINE_RE.match(line)
80      if m:
81        header, value = m.groups()
82        key = None
83        if header in ['STATUS', 'RESULT'] and '=' in value:
84          key, value = value.split('=', 1)
85          bundle[header][key] = [value]
86        elif header == 'STATUS_CODE':
87          yield int(value), join_bundle_values(bundle['STATUS'])
88          bundle['STATUS'] = {}
89        elif header == 'CODE':
90          self._code = int(value)
91        else:
92          logging.warning('Unknown INSTRUMENTATION_%s line: %s', header, value)
93      elif key is not None:
94        bundle[header][key].append(line)
95
96    self._bundle = join_bundle_values(bundle['RESULT'])
97
98  def GetResult(self):
99    """Return the final instrumentation result.
100
101    Returns:
102      A pair (code, bundle) with the final instrumentation result. The |code|
103      may be None if no instrumentation result was found in the output.
104
105    Raises:
106      AssertionError if attempting to get the instrumentation result before
107      exhausting |IterStatus| first.
108    """
109    assert self._bundle is not None, (
110        'The IterStatus generator must be exhausted before reading the final'
111        ' instrumentation result.')
112    return self._code, self._bundle
113