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
|
'''
Uploads an APK and a test APK to AWS Device Farm and runs the Espresso tests.
Exit code is 0 if all tests pass, 1 otherwise. See README.md for details.
'''
from time import sleep
import argparse
import boto3
import requests
'''
Parser
'''
parser = argparse.ArgumentParser(
prog='Device Farm Runner',
description='Runs the Espresso tests on AWS Device Farm.',
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
add_help=True)
parser.add_argument('--project-arn',
type=str, help='The project ARN (Amazon Resource Name)')
parser.add_argument('--device-pool-arn',
type=str, help='The device pool ARN (Amazon Resource Name)')
parser.add_argument('--app-apk-path',
type=str, help='Path to the app APK')
parser.add_argument('--test-apk-path',
type=str, help='Path to the tests APK')
args = vars(parser.parse_args())
project_arn = args.get('project_arn')
device_pool_arn = args.get('device_pool_arn')
app_apk_path = args.get('app_apk_path')
test_apk_path = args.get('test_apk_path')
# Check
if not project_arn or not device_pool_arn:
raise Exception('You need to set both the project and the device pool ARN.')
elif not app_apk_path or not test_apk_path:
raise Exception('You need to set both the app and test APK path.')
'''
The AWS Device Farm client
'''
client = boto3.client('devicefarm', region_name='us-west-2')
'''
Methods
'''
def upload_apk(apk_name, apk_path, apk_type):
print 'Starting upload: %s' % apk_type
result = client.create_upload(
projectArn=project_arn, name=apk_name, type=apk_type)
presigned_url = result.get('upload').get('url')
upload_arn = result.get('upload').get('arn')
# PUT the file content and wait
put_file(apk_path=apk_path, presigned_url=presigned_url)
wait_for_upload(upload_arn=upload_arn)
return upload_arn
def put_file(apk_path, presigned_url):
print 'Uploading: %s' % apk_path
with open(apk_path, 'rb') as f:
data = f.read()
result = requests.put(presigned_url, data=data)
if result.status_code != 200:
raise Exception(
'PUT failed with status code: %s' % result.status_code)
def wait_for_upload(upload_arn):
print 'Checking if the upload succeeded.'
succeeded = False
while not succeeded:
result = client.get_upload(arn=upload_arn)
status = result.get('upload').get('status')
succeeded = (status == 'SUCCEEDED')
if status == 'FAILED':
raise Exception('Upload failed.')
elif not succeeded:
print 'Upload is not ready (status is %s), waiting for 5 seconds.' % status
sleep(5)
def schedule_run(app_arn, test_arn):
print 'Scheduling a run.'
result = client.schedule_run(
projectArn=project_arn,
appArn=app_arn,
devicePoolArn=device_pool_arn,
name='automated_run',
test={ 'type': 'INSTRUMENTATION', 'testPackageArn': test_arn})
run_arn = result.get('run').get('arn')
return_code = wait_for_run(run_arn=run_arn)
return return_code
def wait_for_run(run_arn):
print 'Checking if the run succeeded.'
return_code = ''
succeeded = False
while not succeeded:
result = client.get_run(arn=run_arn)
status = result.get('run').get('status')
return_code = result.get('run').get('result')
succeeded = (status == 'COMPLETED')
if not succeeded:
print 'Run not completed (status is %s), waiting for 60 seconds.' % status
sleep(60)
return return_code
'''
Main flow
'''
# 1. Upload the app APK
app_arn = upload_apk(
apk_name='app-debug-unaligned.apk',
apk_path=app_apk_path,
apk_type='ANDROID_APP')
# 2. Upload the test APK
test_arn = upload_apk(
apk_name='app-debug-androidTest-unaligned.apk',
apk_path=test_apk_path,
apk_type='INSTRUMENTATION_TEST_PACKAGE')
# 3. Schedule the run
return_code = schedule_run(app_arn=app_arn, test_arn=test_arn)
exit_code = 0 if return_code == 'PASSED' else 1
print 'Run completed: %s' % return_code
exit(exit_code)
|
