[telemetry_mini] Add AndroidActions class

experimental/telemetry_mini/telemetry_mini.py

 #!/usr/bin/env python
 # Copyright 2017 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 # This is intended to be a very trimmed down, single-file, hackable, and easy
 # to understand version of Telemetry. It's able to run simple user stories on
 # Android, grab traces, and extract metrics from them. May be useful to
 # diagnose issues with Chrome, reproduce regressions or prototype new user
 # stories.
@@ skipping 12 matching lines @@
 import posixpath
 import re
 import socket
 import subprocess
 import tempfile
 import time
 import websocket  # pylint: disable=import-error
 from xml.etree import ElementTree as element_tree
 
 
+KEYCODE_BACK = 4
+
 # Parse rectangle bounds given as: '[left,top][right,bottom]'.
 RE_BOUNDS = re.compile(
     r'\[(?P<left>\d+),(?P<top>\d+)\]\[(?P<right>\d+),(?P<bottom>\d+)\]')
 
 # TODO: Maybe replace with a true on-device temp file.
 UI_DUMP_TEMP = '/data/local/tmp/tm_ui_dump.xml'
 
 
 def RetryOn(exc_type=(), returns_falsy=False, retries=5):
   """Decorator to retry a function in case of errors or falsy values.
@@ skipping 92 matching lines @@
       row = line.split(None, 8)
       try:
         pid = int(row[1])
         process_name = row[-1]
       except StandardError:
         continue
       result[process_name].append(pid)
     return result
 
   @RetryOn(AdbCommandError)
-  def GetUiDump(self):
+  def GetUiScreenDump(self):
     """Return the root XML node with screen captured from the device."""
     self.RunShellCommand('rm', '-f', UI_DUMP_TEMP)
     output = self.RunShellCommand('uiautomator', 'dump', UI_DUMP_TEMP).strip()
 
     if output.startswith('ERROR:'):
       # uiautomator may fail if device is not in idle state, e.g. animations
       # or video playing. Retry if that's the case.
       raise AdbCommandError(output)
 
     with tempfile.NamedTemporaryFile(suffix='.xml') as f:
       f.close()
       self.RunCommand('pull', UI_DUMP_TEMP, f.name)
       return element_tree.parse(f.name)
 
   @RetryOn(LookupError)
-  def FindUiNode(self, attr_values):
-    """Find a UI node on screen capture, retrying if not yet visible."""
-    root = self.GetUiDump()
+  def FindUiElement(self, attr_values):
+    """Find a UI element on screen capture, retrying if not yet visible."""
+    root = self.GetUiScreenDump()
     for node in root.iter():
       if all(node.get(k) == v for k, v in attr_values):
         return node
-    raise LookupError('Specified UI node not found')
+    raise LookupError('Specified UI element not found')
 
-  def TapUiNode(self, *args, **kwargs):
-    node = self.FindUiNode(*args, **kwargs)
+  def TapUiElement(self, *args, **kwargs):
+    node = self.FindUiElement(*args, **kwargs)
     m = RE_BOUNDS.match(node.get('bounds'))
     left, top, right, bottom = (int(v) for v in m.groups())
     x, y = (left + right) / 2, (top + bottom) / 2
     self.RunShellCommand('input', 'tap', str(x), str(y))
 
 
+def _UserAction(f):
+  """Decorator to add repeat, and action_delay options to user action methods.
+
+  Note: The values (or their defaults) supplied for these extra options will
+  also be passed down to the decorated method. It's thus advisable to collect
+  them in a catch-all **kwargs, even if just to discard them immediately.
+
+  This is a workaround for https://github.com/PyCQA/pylint/issues/258 in which
+  decorators confuse pylint and trigger spurious 'unexpected-keyword-arg'
+  warnings on method calls that use the extra options.
+  """
+  @functools.wraps(f)
+  def Wrapper(self, *args, **kwargs):
+    repeat = kwargs.setdefault('repeat', 1)
+    action_delay = kwargs.setdefault('action_delay', None)
+    for _ in xrange(repeat):
+      f(self, *args, **kwargs)
+      self.Idle(action_delay)
+  return Wrapper
+
+
+class AndroidActions(object):
+  def __init__(self, device, user_action_delay=1):
+    self.device = device
+    self.user_action_delay = user_action_delay
+
+  def Idle(self, duration=None):
+    if duration is None:
+      duration = self.user_action_delay
+    if duration:
+      time.sleep(duration)
+
+  @_UserAction
+  def GoBack(self, **kwargs):
+    del kwargs
+    self.device.RunShellCommand('input', 'keyevent', str(KEYCODE_BACK))
+
+  @_UserAction
+  def StartActivity(
+      self, data_uri, action='android.intent.action.VIEW', **kwargs):
+    del kwargs
+    self.device.RunShellCommand('am', 'start', '-a', action, '-d', data_uri)
+
+  @_UserAction
+  def TapUiElement(self, attr_values, **kwargs):
+    del kwargs
+    self.device.TapUiElement(attr_values)
+
+  @_UserAction
+  def SwipeUp(self, **kwargs):
+    del kwargs
+    # Hardcoded values for 480x854 screen size; should work reasonably on
+    # other screen sizes.
+    # Command args: swipe <x1> <y1> <x2> <y2> [duration(ms)]
+    self.device.RunShellCommand(
+        'input', 'swipe', '240', '568', '240', '284', '400')
+
+
 class DevToolsWebSocket(object):
   def __init__(self, url):
     self._url = url
     self._socket = None
     self._cmdid = 0
 
   def __enter__(self):
     self.Open()
     return self
 
@@ skipping 200 matching lines @@
   def Uninstall(self):
     # System Chrome app cannot be (un)installed, so we enable/disable instead.
     logging.warning('Disabling %s', self.PACKAGE_NAME)
     self.device.RunShellCommand('pm', 'disable', self.PACKAGE_NAME)
 
 
 class UserStory(object):
   def __init__(self, browser):
     self.device = browser.device
     self.browser = browser
+    self.actions = AndroidActions(self.device)
 
   def GetExtraStoryApps(self):
     """Sequence of AndroidApp's, other than the browser, used in the story."""
     return ()
 
   def EnsureExtraStoryAppsClosed(self):
     running_processes = self.device.ProcessStatus()
     for app in self.GetExtraStoryApps():
       if app.PACKAGE_NAME in running_processes:
         app.ForceStop()
@@ skipping 32 matching lines @@
     if event['ph'] == 'v':
       # Extract any metrics you may need from the trace.
       value = event['args']['dumps']['allocators'][
           'java_heap/allocated_objects']['attrs']['size']
       assert value['units'] == 'bytes'
       processes[event['pid']]['java_heap'] = int(value['value'], 16)
     elif event['ph'] == 'M' and event['name'] == 'process_name':
       processes[event['pid']]['name'] = event['args']['name']
 
   return processes.values()