DZone
Thanks for visiting DZone today,
Edit Profile
  • Manage Email Subscriptions
  • How to Post to DZone
  • Article Submission Guidelines
Sign Out View Profile
  • Post an Article
  • Manage My Drafts
Over 2 million developers have joined DZone.
Log In / Join
Refcards Trend Reports Events Over 2 million developers have joined DZone. Join Today! Thanks for visiting DZone today,
Edit Profile Manage Email Subscriptions Moderation Admin Console How to Post to DZone Article Submission Guidelines
View Profile
Sign Out
Refcards
Trend Reports
Events
Zones
Culture and Methodologies Agile Career Development Methodologies Team Management
Data Engineering AI/ML Big Data Data Databases IoT
Software Design and Architecture Cloud Architecture Containers Integration Microservices Performance Security
Coding Frameworks Java JavaScript Languages Tools
Testing, Deployment, and Maintenance Deployment DevOps and CI/CD Maintenance Monitoring and Observability Testing, Tools, and Frameworks
Culture and Methodologies
Agile Career Development Methodologies Team Management
Data Engineering
AI/ML Big Data Data Databases IoT
Software Design and Architecture
Cloud Architecture Containers Integration Microservices Performance Security
Coding
Frameworks Java JavaScript Languages Tools
Testing, Deployment, and Maintenance
Deployment DevOps and CI/CD Maintenance Monitoring and Observability Testing, Tools, and Frameworks
  1. DZone
  2. Software Design and Architecture
  3. Cloud Architecture
  4. Getting Started With AWS Device Farm and Python Appium

Getting Started With AWS Device Farm and Python Appium

In this quick tutorial, you'll learn how to run mobile integration tests with AWS Device Farm and the Python Appium framework.

Chase Seibert user avatar by
Chase Seibert
·
Mar. 21, 17 · Tutorial
Like (2)
Save
Tweet
Share
4.83K Views

Join the DZone community and get the full member experience.

Join For Free

AWS Device Farm is a service for running mobile app integration tests against a suite of physical devices. The Amazon documentation is exhaustive, and they support many different API clients. However, the documentation does not have a quick start example for the thing most people are going to want to do first: run a test. Here is some working code using the Python boto3 client.

Setup

First, you will want to create a Python virtual environment for this and install some dependencies. Feel free to omit the version numbers, but I've tested the script against these versions.

virtualenv venv
. venv/bin/activate
pip install boto3==1.4.4 requests==2.13.0

Pre-Requisites

This script assumes you have already built an APK file for your mobile app, and that you have bundled your Appium Python tests in a zip file as per the Amazon Appium Python documentation.

Need to write your first tests? I have an Appium Python Quickstart, too.

Note: for the bundled Appium tests, make sure to do the wheel step on a Linux x86_64 vm.

Also, you need to be authenticated to AWS. The easiest method is to generate an API key and secret for your user in the AWS web console and run aws configure via the AWS Command line interface.

Running the Test Suite

The script bellow will do the following:

  1. Find the ID your AWS device farm project by name.
  2. Find the ID of the pool of devices you want to use to test by name.
  3. Upload a local Android APK file to AWS.
  4. Upload a local zip file of your tests, requirements.txt, and wheelhouse.
  5. Start a test run.
  6. Poll until the test run is completed.
#!/usr/bin/env python
import logging
import pprint
import time

import boto3
import requests


REGION = 'us-west-2'
PROJECT_NAME = 'My Mobile App'
DEVICE_POOL_NAME = 'Top Devices'  # this is a default pool
RUN_TIMEOUT_SECONDS = 60 * 20
WEB_URL_TEMPLATE = 'https://us-west-2.console.aws.amazon.com/devicefarm/home#/projects/%s/runs/%s'


device_farm = boto3.client('devicefarm', region_name=REGION)
s3 = boto3.client('s3', region_name=REGION)
logger = logging.getLogger(__name__)


def get_project_arn(name):
    for project in device_farm.list_projects()['projects']:
        if project['name'] == name:
            return project['arn']
    raise KeyError('Could not find project %r' % name)


def get_device_pool(project_arn, name):
    for device_pool in device_farm.list_device_pools(arn=project_arn)['devicePools']:
        if device_pool['name'] == name:
            return device_pool['arn']
    raise KeyError('Could not find device pool %r' % name)


def _upload_presigned_url(url, file_path):
    with open(file_path) as fp:
        data = fp.read()
        result = requests.put(url, data=data, headers={'content-type': 'application/octet-stream'})
        assert result.status_code == 200


def create_upload(project_arn, upload_type, name, file_path):
    # name needs to be a file name like app-releaseProduction.apk, not "Android App"
    logger.info('Uploading %s %r' % (upload_type, file_path))
    result = device_farm.create_upload(
        projectArn=project_arn,
        name=name,
        type=upload_type,
        contentType='application/octet-stream',
    )
    upload = result['upload']
    _upload_presigned_url(upload['url'], file_path)
    return upload['arn']


def schedule_run(project_arn, name, device_pool_arn, app_arn, test_package_arn):
    logger.info('Scheduling test run %r' % name)
    result = device_farm.schedule_run(
        projectArn=project_arn,
        appArn=app_arn,
        devicePoolArn=device_pool_arn,
        name=name,
        test={
            'type': 'APPIUM_PYTHON',
            'testPackageArn': test_package_arn,
        }
    )
    run = result['run']
    return run['arn']


def _poll_until(method, arn, get_status_callable, success_statuses, timeout_seconds=10):
    check_every_seconds = 10 if timeout_seconds == RUN_TIMEOUT_SECONDS else 1
    start = time.time()
    while True:
        result = method(arn=arn)
        current_status = get_status_callable(result)
        if current_status in success_statuses:
            return result
        logger.info('Waiting for %r status %r to be in %r' % (arn, current_status, success_statuses))
        now = time.time()
        if now - start > timeout_seconds:
            raise StopIteration('Time out waiting for %r to be done' % arn)
        time.sleep(check_every_seconds)


def wait_for_upload(arn):
    return _poll_until(
        device_farm.get_upload,
        arn,
        get_status_callable=lambda x: x['upload']['status'],
        success_statuses=('SUCCEEDED', ),
    )


def wait_for_run(test_package_arn):
    result = _poll_until(
        device_farm.get_run,
        test_package_arn,
        get_status_callable=lambda x: x['run']['status'],
        success_statuses=('COMPLETED', ),
        timeout_seconds=RUN_TIMEOUT_SECONDS,
    )
    final_run = result['run']
    logger.info('Final run counts: %(counters)s' % final_run)
    return final_run['result'] == 'PASSED'


def get_run_web_url(project_arn, test_run_arn):
    # project_arn = arn:aws:devicefarm:us-west-2:foo:project:NEW-ARN-HERE
    # test_run_arn = arn:aws:devicefarm:us-west-2:foo:run:project-arn/NEW-ARN-HERE
    project_arn_id = project_arn.split(':')[6]
    test_run_arid = test_run_arn.split('/')[1]
    return WEB_URL_TEMPLATE % (
        project_arn_id,
        test_run_arid,
    )


if __name__ == '__main__':
    logging.basicConfig(format='%(message)s')
    logger.setLevel(logging.INFO)
    project_arn = get_project_arn(PROJECT_NAME)
    logger.info('Project: %r' % project_arn)
    device_pool_arn = get_device_pool(project_arn, DEVICE_POOL_NAME)
    logger.info('Device pool: %r' % device_pool_arn)
    app_arn = create_upload(
        project_arn,
        'ANDROID_APP',
        'my-app.apk',
        '/tmp/my-app.apk',
    )
    wait_for_upload(app_arn)
    logger.info('App: %s' % app_arn)
    test_package_arn = create_upload(
        project_arn,
        'APPIUM_PYTHON_TEST_PACKAGE',
        'test_bundle.zip',
        '/tmp/test_bundle.zip',
    )
    wait_for_upload(test_package_arn)
    logger.info('Test package: %s' % test_package_arn)
    test_run_arn = schedule_run(
        project_arn,
        name='Test Run 1',
        device_pool_arn=device_pool_arn,
        app_arn=app_arn,
        test_package_arn=test_package_arn,
    )
    logger.info('Scheduled test run %r' % test_run_arn)
    logger.info('View scheduled run at %s' % get_run_web_url(project_arn, test_run_arn))
    success = wait_for_run(test_run_arn)
    logger.info('Success')

The only tricky parts here are using the pre-signed URLs from the create_upload step to make a subsequent PUT call with the file contents and setting the ContentType correctly for that to work. Also, we need to poll for both uploads to be processed before we kick off the test run.


I'm currently working at NerdWallet, a startup in San Francisco trying to bring clarity to all of life's financial decisions. We're hiring like crazy. Hit me up on Twitter, I would love to talk.

AWS Python (language)

Published at DZone with permission of Chase Seibert, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.

Popular on DZone

  • Type Variance in Java and Kotlin
  • How to Develop a Portrait Retouching Function
  • Unlocking the Power of Polymorphism in JavaScript: A Deep Dive
  • What Should You Know About Graph Database’s Scalability?

Comments

Partner Resources

X

ABOUT US

  • About DZone
  • Send feedback
  • Careers
  • Sitemap

ADVERTISE

  • Advertise with DZone

CONTRIBUTE ON DZONE

  • Article Submission Guidelines
  • Become a Contributor
  • Visit the Writers' Zone

LEGAL

  • Terms of Service
  • Privacy Policy

CONTACT US

  • 600 Park Offices Drive
  • Suite 300
  • Durham, NC 27709
  • support@dzone.com
  • +1 (919) 678-0300

Let's be friends: