Add AMP support to test runner.

build/android/pylib/remote/device/remote_device_environment.py

+# Copyright 2014 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.
+
+"""Environment setup and teardown for remote devices."""
+
+import os
+import sys
+
+from pylib import constants
+from pylib.base import environment
+from pylib.remote.device import remote_device_helper
+
+sys.path.append(os.path.join(
+    constants.DIR_SOURCE_ROOT, 'third_party', 'requests', 'src'))
+sys.path.append(os.path.join(
+    constants.DIR_SOURCE_ROOT, 'third_party', 'appurify-python', 'src'))
+import appurify.api
+
+
+class RemoteDeviceEnvironment(environment.Environment):
+  """An environment for running on remote devices."""
+
+  def __init__(self, args, error_func):
+    """Constructor.
+
+    Args:
+      args: Command line arguments.
+      error_func: error to show when using bad command line arguments.
+    """
+    super(RemoteDeviceEnvironment, self).__init__()
+
+    if args.api_key_file:
+      with open(args.api_key_file) as api_key_file:
+        self._api_key = api_key_file.read().strip()
+    elif args.api_key:
+      self._api_key = args.api_key
+    else:
+      error_func('Must set api key with --api-key or --api-key-file')
+
+    if args.api_secret_file:
+      with open(args.api_secret_file) as api_secret_file:
+        self._api_secret = api_secret_file.read().strip()
+    elif args.api_secret:
+      self._api_secret = args.api_secret
+    else:
+      error_func('Must set api secret with --api-secret or --api-secret-file')
+
+    self._api_protocol = args.api_protocol
+    self._api_address = args.api_address
+    self._api_port = args.api_port
+    self._access_token = ''
+    self._results_path = args.results_path
+    self._remote_device = args.remote_device
+    self._remote_device_os = args.remote_device_os
+    self._runner_package = args.runner_package
+    self._runner_type = args.runner_type
+    self._device = ''
+    if not args.trigger and not args.collect:
+      self._trigger = True
+      self._collect = True
+    else:
+      self._trigger = args.trigger
+      self._collect = args.collect
+
+  def SetUp(self):
+    """Set up the test environment."""
+    os.environ['APPURIFY_API_PROTO'] = self._api_protocol
+    os.environ['APPURIFY_API_HOST'] = self._api_address
+    os.environ['APPURIFY_API_PORT'] = self._api_port
+    self._GetAccessToken()
+    if self._trigger:
+      self._device = self._SelectDevice()
+
+  def TearDown(self):
+    """Teardown the test environment."""
+    self._RevokeAccessToken()
+
+  def __enter__(self):
+    """Set up the test run when used as a context manager."""
+    self.SetUp()
+    return self
+
+  def __exit__(self, exc_type, exc_val, exc_tb):
+    """Tears down the test run when used as a context manager."""
+    self.TearDown()
+
+  def _GetAccessToken(self):
+    """Generates access token for remote device service."""
+    access_token_results = appurify.api.access_token_generate(
+        self._api_key, self._api_secret)
+    remote_device_helper.TestHttpResponse(access_token_results,
+                                          'Unable to generate access token.')
+    self._access_token = access_token_results.json()['response']['access_token']
+
+  def _RevokeAccessToken(self):
+    """Destroys access token for remote device service."""
+    revoke_token_results = appurify.api.access_token_revoke(self._access_token)
+    remote_device_helper.TestHttpResponse(revoke_token_results,
+                                          'Unable to revoke access token.')
+
+  def _SelectDevice(self):
+    """Select which device to use."""
+    dev_list_res = appurify.api.devices_list(self._access_token)
+    remote_device_helper.TestHttpResponse(dev_list_res,
+                                          'Unable to generate access token.')
+    device_list = dev_list_res.json()['response']
+    for device in device_list:
+      if (device['name'] == self._remote_device
+          and device['os_version'] == self._remote_device_os):
+        return device['device_type_id']
+    raise remote_device_helper.RemoteDeviceError(
+        'No device found: %s %s' % (self._remote_device,
+        self._remote_device_os))
+
+  @property
+  def device(self):
+    return self._device
+
+  @property
+  def token(self):
+    return self._access_token
+
+  @property
+  def results_path(self):
+    return self._results_path
+
+  @property
+  def runner_type(self):
+    return self._runner_type
+
+  @property
+  def runner_package(self):
+    return self._runner_package
+
+  @property
+  def trigger(self):
+    return self._trigger
+
+  @property
+  def collect(self):
+    return self._collect