blob: 871c0a83a43699e4bd21924978b0c0e2250dd39d [file] [log] [blame]
# 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')
if not args.api_protocol:
error_func('Must set api protocol with --api-protocol. Example: http')
self._api_protocol = args.api_protocol
if not args.api_address:
error_func('Must set api address with --api-address')
self._api_address = args.api_address
if not args.api_port:
error_func('Must set api port with --api-port.')
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