build/android/pylib/instrumentation/instrumentation_parser.py

   1 # Copyright 2015 The Chromium Authors. All rights reserved.
   2 # Use of this source code is governed by a BSD-style license that can be
   3 # found in the LICENSE file.
   4
   5 import logging
   6 import re
   7
   8 # http://developer.android.com/reference/android/test/InstrumentationTestRunner.html
   9 STATUS_CODE_START = 1
  10 STATUS_CODE_OK = 0
  11 STATUS_CODE_ERROR = -1
  12 STATUS_CODE_FAILURE = -2
  13
  14 # http://developer.android.com/reference/android/app/Activity.html
  15 RESULT_CODE_OK = -1
  16 RESULT_CODE_CANCELED = 0
  17
  18 _INSTR_LINE_RE = re.compile(r'^\s*INSTRUMENTATION_([A-Z_]+): (.*)$')
  19
  20
  21 class InstrumentationParser(object):
  22
  23   def __init__(self, stream):
  24     """An incremental parser for the output of Android instrumentation tests.
  25
  26     Example:
  27
  28       stream = adb.IterShell('am instrument -r ...')
  29       parser = InstrumentationParser(stream)
  30
  31       for code, bundle in parser.IterStatus():
  32         # do something with each instrumentation status
  33         print 'status:', code, bundle
  34
  35       # do something with the final instrumentation result
  36       code, bundle = parser.GetResult()
  37       print 'result:', code, bundle
  38
  39     Args:
  40       stream: a sequence of lines as produced by the raw output of an
  41         instrumentation test (e.g. by |am instrument -r| or |uiautomator|).
  42     """
  43     self._stream = stream
  44     self._code = None
  45     self._bundle = None
  46
  47   def IterStatus(self):
  48     """Iterate over statuses as they are produced by the instrumentation test.
  49
  50     Yields:
  51       A tuple (code, bundle) for each instrumentation status found in the
  52       output.
  53     """
  54     def join_bundle_values(bundle):
  55       for key in bundle:
  56         bundle[key] = '\n'.join(bundle[key])
  57       return bundle
  58
  59     bundle = {'STATUS': {}, 'RESULT': {}}
  60     header = None
  61     key = None
  62     for line in self._stream:
  63       m = _INSTR_LINE_RE.match(line)
  64       if m:
  65         header, value = m.groups()
  66         key = None
  67         if header in ['STATUS', 'RESULT'] and '=' in value:
  68           key, value = value.split('=', 1)
  69           bundle[header][key] = [value]
  70         elif header == 'STATUS_CODE':
  71           yield int(value), join_bundle_values(bundle['STATUS'])
  72           bundle['STATUS'] = {}
  73         elif header == 'CODE':
  74           self._code = int(value)
  75         else:
  76           logging.warning('Unknown INSTRUMENTATION_%s line: %s', header, value)
  77       elif key is not None:
  78         bundle[header][key].append(line)
  79
  80     self._bundle = join_bundle_values(bundle['RESULT'])
  81
  82   def GetResult(self):
  83     """Return the final instrumentation result.
  84
  85     Returns:
  86       A pair (code, bundle) with the final instrumentation result. The |code|
  87       may be None if no instrumentation result was found in the output.
  88
  89     Raises:
  90       AssertionError if attempting to get the instrumentation result before
  91       exhausting |IterStatus| first.
  92     """
  93     assert self._bundle is not None, (
  94         'The IterStatus generator must be exhausted before reading the final'
  95         ' instrumentation result.')
  96     return self._code, self._bundle