1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
# 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 logging
import os
import random
import sys
from pylib import constants
from pylib.base import environment
from pylib.remote.device import appurify_sanitized
from pylib.remote.device import remote_device_helper
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 = ''
self._verbose_count = args.verbose_count
self._timeouts = {
'queueing': 60 * 10,
'installing': 60 * 10,
'in-progress': 60 * 30,
'unknown': 60 * 5
}
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."""
try:
self.SetUp()
return self
except:
self.__exit__(*sys.exc_info())
raise
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."""
logging.info('Generating remote service access token')
with appurify_sanitized.SanitizeLogging(self._verbose_count,
logging.WARNING):
access_token_results = appurify_sanitized.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."""
logging.info('Revoking remote service access token')
with appurify_sanitized.SanitizeLogging(self._verbose_count,
logging.WARNING):
revoke_token_results = appurify_sanitized.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."""
logging.info('Finding device to run tests on.')
with appurify_sanitized.SanitizeLogging(self._verbose_count,
logging.WARNING):
dev_list_res = appurify_sanitized.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']
random.shuffle(device_list)
for device in device_list:
if device['os_name'] != 'Android':
continue
if self._remote_device and device['name'] != self._remote_device:
continue
if (self._remote_device_os
and device['os_version'] != self._remote_device_os):
continue
if ((self._remote_device and self._remote_device_os)
or device['available_devices_count']):
logging.info('Found device: %s %s',
device['name'], device['os_version'])
return device['device_type_id']
self._NoDeviceFound(device_list)
def _PrintAvailableDevices(self, device_list):
def compare_devices(a,b):
for key in ('os_version', 'name'):
c = cmp(a[key], b[key])
if c:
return c
return 0
logging.critical('Available Android Devices:')
android_devices = (d for d in device_list if d['os_name'] == 'Android')
for d in sorted(android_devices, compare_devices):
logging.critical(' %s %s', d['os_version'].ljust(7), d['name'])
def _NoDeviceFound(self, device_list):
self._PrintAvailableDevices(device_list)
raise remote_device_helper.RemoteDeviceError('No device found.')
@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
@property
def verbose_count(self):
return self._verbose_count
@property
def timeouts(self):
return self._timeouts
|
