Device Farm for Python (Boto3) - AWSSDK コードサンプル

AWSDocAWS SDKGitHub サンプルリポジトリには、さらに多くの SDK サンプルがあります


Device Farm for Python (Boto3)

次のコード例は、AWS SDK for Python (Boto3) with Device Farm を使用してアクションを実行する方法を示しています。






次のコード例は、Device Farm でブラウザテストを実行し、スクリーンショットを撮る方法を示しています。

SDK for Python (Boto3)

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

PyTestと Selenium を使用して、指定した Web サイトを閲覧したり、スクリーンショットを撮ったり、実際の Web サイトのコンテンツと予想されるコンテンツと比較したりします。

import datetime import os import subprocess import boto3 import pytest from selenium import webdriver from selenium.webdriver import DesiredCapabilities from import By from selenium.webdriver.common.keys import Keys from import expected_conditions from import WebDriverWait def get_git_hash(): """ Get the short Git hash of the current commit of the repository """ try: return subprocess.check_output( ['git', 'rev-parse', '--short', 'HEAD']).decode('utf-8').strip() except: return "norepo" class TestHelloSuite: """ Our test suite. This style of test suite allows us to use setup_method and teardown_method. """ def save_screenshot(self, name): self.driver.save_screenshot(os.path.join(self.screenshot_path, name)) def setup_method(self, method): """ Set up a test. This makes sure that the session for an individual test is ready. The AWS credentials are read from the default ~/.aws/credentials or from the command line by setting the AWS_ACCESS_KEY_ID and AWS_SECRET_KEY environment variables. The project Amazon Resource Name (ARN) is determined by the PROJECT_ARN environment variable. """ devicefarm_client = boto3.client("devicefarm") project_arn = os.environ.get("PROJECT_ARN", None) if project_arn is None: raise ValueError("Must set PROJECT_ARN") # Request a driver hub URL for the Selenium client testgrid_url_response = devicefarm_client.create_test_grid_url( projectArn=project_arn, expiresInSeconds=300) # We want a directory to save our files into. We're going to make a directory # in the current directory that holds our results. self.screenshot_path = os.path.join( '.', 'results', get_git_hash() + '-' + ( if not os.path.exists(self.screenshot_path): os.makedirs(self.screenshot_path, exist_ok=True) # We want a Firefox instance on Windows desired_cap = DesiredCapabilities.FIREFOX desired_cap['platform'] = 'windows' desired_cap['BrowserVersion'] = 'latest' # Configure the webdriver with the appropriate remote endpoint. self.driver = webdriver.Remote(testgrid_url_response['url'], desired_cap) # # Auto-Tagging # # In order to get the Session ARN, we need to look up the session by the # Project ARN and session ID (from the driver). testgrid_session_arn_response = devicefarm_client.get_test_grid_session( projectArn=project_arn, sessionId=self.driver.session_id ) # Save the session's ARN so we can tag the session. self.session_arn = testgrid_session_arn_response['testGridSession']['arn'] # In order to tag it, we're going to use the resourcegroupstaggingapi client to # add a tag to the session ARN that we just got. tag_client = boto3.client('resourcegroupstaggingapi') tag_client.tag_resources( ResourceARNList=[self.session_arn], Tags={ "TestSuite": f"testsuite {method.__name__}", "GitId": get_git_hash()}) def teardown_method(self, method): """ Clean up resources used by each method. """ # End the Selenium session so we're off the clock. self.driver.quit() @pytest.mark.parametrize( 'query,leading', [ pytest.param('Seattle', 'Seattle (/siˈætəl/ (listen) see-AT-əl) is a seaport city on the West Coast of the United States.'), pytest.param('Selenium', "Selenium is a chemical element with the symbol Se and atomic number 34."), pytest.param('Amazon Locker', 'Amazon Locker is a self-service package delivery service offered by online retailer Amazon.'), pytest.param('Kootenai Falls','Kootenai Falls is a waterfall on the Kootenay River located in Lincoln County, Montana, just off U.S. Route 2.'), pytest.param('Dorayaki', 'Dorayaki (どら焼き, どらやき, 銅鑼焼き, ドラ焼き) is a type of Japanese confection.'), pytest.param('Robot Face', '<|°_°|> (also known as Robot Face or Robot)') ]) def test_first_paragraph_text(self, query, leading): """ This test looks at the first paragraph of a page on Wikipedia, comparing it to a known leading sentence. If the leading sentence matches, the test passes. A screenshot is taken before the final assertion is made, letting us debug if something isn't right. """ # Open the main page of Wikipedia self.driver.get("") # Find the search box, enter a query, and press enter search_input = self.driver.find_element(By.ID, "searchInput") search_input.send_keys(query) search_input.send_keys(Keys.ENTER) # Wait for the search box to go stale -- This means we've navigated fully. WebDriverWait(self.driver, 5).until(expected_conditions.staleness_of(search_input)) # Get the leading paragraph of the article. lead = leading.lower() # Find the element... lead_para = self.driver.find_element( By.XPATH, "//div[@class='mw-parser-output']//p[not(@class)]") # ... and copy out its text. our_text = lead_para.text.lower() our_text = our_text[:len(lead)] # Take a screenshot and compare the strings. self.save_screenshot(f"leadingpara_{query}.png") assert our_text.startswith(lead) @pytest.mark.parametrize( "query,expected", [ pytest.param("Automation Testing", "Test Automation"), pytest.param("DevOps", "DevOps"), pytest.param("Jackdaws Love My Big Sphinx Of Quartz", "Pangram"), pytest.param("EarthBound", "EarthBound"), pytest.param("Covered Bridges Today", "Covered Bridges Today"), pytest.param("Kurt Godel", "Kurt Gödel"), pytest.param("N//ng language", "Nǁng language"), pytest.param("Who the Fuck Is Jackson Pollock?", "Who the $&% Is Jackson Pollock?") ]) def test_redirect_titles(self, query, expected): """ A test comparing pages we expect to (or not to) redirect on Wikipedia. This test checks to see that the page ("query") redirects (or doesn't) to the "expected" page title. Several of these are common synonyms ("Jackdaws...") while others are because of characters untypable by most keyboards ("Nǁng language") A screenshot is taken just before the final assertion is made to aid in debugging and verification. """ # Open the main page of Wikipedia self.driver.get("") # Find the search box, enter some text into it, and send an enter key. search_input = self.driver.find_element(By.ID, "searchInput") search_input.send_keys(query) search_input.send_keys(Keys.ENTER) # wait until the page has rolled over -- once the search input handle is stale, # the browser has navigated. WebDriverWait(self.driver, 5).until(expected_conditions.staleness_of(search_input)) # Get the first heading & take a screenshot our_text = self.driver.find_element(By.ID, "firstHeading").text.lower() self.save_screenshot(f"redirect_{query}.png") # did it match? assert our_text == expected.lower()
  • API の詳細については、「AWS SDK for Python (Boto3) API リファレンス」の以下のトピックを参照してください。

次のコード例は、Device Farm でモバイルデバイスパッケージをアップロードしてテストする方法を示しています。

SDK for Python (Boto3)

他にもありますGitHub。用例一覧を検索し、AWS コード例リポジトリでの設定と実行の方法を確認してください。

コンパイル済みの Android アプリケーションとテストパッケージを Device Farm にアップロードしてテストを開始し、テストが完了するのを待って結果を報告します。

import boto3 import os import requests import string import random import datetime import time # Update this dict with your own values before you run the example: config = { # This is our app under test. "appFilePath": "app-debug.apk", "projectArn": "arn:aws:devicefarm:us-west-2:111222333444:project:581f5703-e040-4ac9-b7ae-0ba007bfb8e6", # Since we care about the most popular devices, we'll use a curated pool. "testSpecArn": "arn:aws:devicefarm:us-west-2::upload:20fcf771-eae3-4137-aa76-92e17fb3131b", "poolArn": "arn:aws:devicefarm:us-west-2::devicepool:4a869d91-6f17-491f-9a95-0a601aee2406", "namePrefix": "MyAppTest", # This is our test package. This tutorial won't go into how to make these. "testPackage": "" } client = boto3.client('devicefarm') unique = config['namePrefix']+"-"+(''.join(random.sample(string.ascii_letters, 8))) print(f"The unique identifier for this run is '{unique}'. All uploads will be prefixed " f"with this.") def upload_df_file(filename, type_, mime='application/octet-stream'): upload_response = client.create_upload( projectArn=config['projectArn'], name=unique+"_"+os.path.basename(filename), type=type_, contentType=mime) upload_arn = upload_response['upload']['arn'] # Extract the URL of the upload and use Requests to upload it. upload_url = upload_response['upload']['url'] with open(filename, 'rb') as file_stream: print(f"Uploading {filename} to Device Farm as " f"{upload_response['upload']['name']}... ", end='') put_req = requests.put( upload_url, data=file_stream, headers={"content-type": mime}) print(' done') if not put_req.ok: raise Exception(f"Couldn't upload. Requests says: {put_req.reason}") started = while True: print(f"Upload of {filename} in state {upload_response['upload']['status']} " f"after "+str( - started)) if upload_response['upload']['status'] == 'FAILED': raise Exception( f"The upload failed processing. Device Farm says the reason is: \n" f"{+upload_response['upload']['message']}") if upload_response['upload']['status'] == 'SUCCEEDED': break time.sleep(5) upload_response = client.get_upload(arn=upload_arn) print("") return upload_arn our_upload_arn = upload_df_file(config['appFilePath'], "ANDROID_APP") our_test_package_arn = upload_df_file(config['testPackage'], 'APPIUM_PYTHON_TEST_PACKAGE') print(our_upload_arn, our_test_package_arn) response = client.schedule_run( projectArn=config["projectArn"], appArn=our_upload_arn, devicePoolArn=config["poolArn"], name=unique, test={ "type" :"APPIUM_PYTHON", "testSpecArn": config["testSpecArn"], "testPackageArn": our_test_package_arn}) run_arn = response['run']['arn'] start_time = print(f"Run {unique} is scheduled as arn {run_arn} ") state = 'UNKNOWN' try: while True: response = client.get_run(arn=run_arn) state = response['run']['status'] if state == 'COMPLETED' or state == 'ERRORED': break else: print(f" Run {unique} in state {state}, total " f"time { - start_time}") time.sleep(10) except: client.stop_run(arn=run_arn) exit(1) print(f"Tests finished in state {state} after { - start_time}") # Pull all the logs. jobs_response = client.list_jobs(arn=run_arn) # Save the output somewhere, using the unique value. save_path = os.path.join(os.getcwd(), 'results', unique) os.mkdir(save_path) # Save the last run information. for job in jobs_response['jobs']: job_name = job['name'] os.makedirs(os.path.join(save_path, job_name), exist_ok=True) # Get each suite within the job. suites = client.list_suites(arn=job['arn'])['suites'] for suite in suites: for test in client.list_tests(arn=suite['arn'])['tests']: # Get the artifacts. for artifact_type in ['FILE', 'SCREENSHOT', 'LOG']: artifacts = client.list_artifacts( type=artifact_type, arn=test['arn'])['artifacts'] for artifact in artifacts: # Replace `:` because it has a special meaning in Windows & macOS. path_to = os.path.join( save_path, job_name, suite['name'], test['name'].replace(':', '_')) os.makedirs(path_to, exist_ok=True) filename = artifact['type']+"_"+artifact['name']+"."+artifact['extension'] artifact_save_path = os.path.join(path_to, filename) print(f"Downloading {artifact_save_path}") with open(artifact_save_path, 'wb') as fn: with requests.get(artifact['url'], allow_redirects=True) as request: fn.write(request.content) print("Finished")