概要
- iOSアプリのバックエンドAPIに対して、負荷テストをしたい
- アプリの使用開始時の挙動(APIコール内容)をシナリオ的に検証したい
- API:Aにコールが集中している状況でAPI:Bをコールすると発生する問題がある
- アプリの使用開始時の挙動(APIコール内容)をシナリオ的に検証したい
-
アプリの通信内容をキャプチャして、テストシナリオを自動で生成できないか?
- 経験則からテスト対象APIを絞ることはできるが、意図しない組合せでAPIシステムが落ちることを防げない
- アプリのテストケースはない前提
-
プロトタイプ的にかんたんに動くものを作ってみた
注意事項
- 負荷テストは自身が管理するシステム以外に実行しないでください
登場人物
-
- APIコールをするやつ
-
- アプリの通信内容をキャプチャするアプリ
-
- アプリのバックエンド: REST APIサーバ
-
- 負荷試験ツール
-
- 許されることなら何もしたくない
やりたいこと
概要
- 負荷テストシナリオを生成する
- 負荷テストシナリオを実行する
詳細
負荷テストシナリオを生成する
-
charles for iosでアプリの通信内容(chlsj)をキャプチャし、エクスポートする
-> charles session json(chlsj)- 参考: [iOS] Charles for iOSがリリースされたので試してみた
- アプリからcharles session をエクスポートしてpcに送る
- 例: シナリオ名: initial_session.chlsj
-
charles session json(chlsj)からシナリオテンプレート(CSV)を生成する
-
依存
- python3
-
pip install python-dateutil
変換スクリプトを起動する
$python chlsj_2_scenario_csv.py {シナリオ名}.chlsj
-> {シナリオ名}_template.csvが生成される
chlsj_2_scenario_csv.py
# coding: UTF-8
import json
import sys
import csv
import dateutil.parser
import os
initial_time = None
def session_json_2_scenario_csv(data):
request = data['request']
request_body = ''
if ('body' in request) and ('text' in request['body']):
request_body = request['body']['text']
# 呼び出し開始msecを計算
run_at = dateutil.parser.parse(data['times']['start'])
global initial_time
initial_time = initial_time or run_at
run_msec_at = int((run_at-initial_time).total_seconds()*1000)
# headerを{"name":"Authorizatoin","value":"~~~"}から{"Authorizatoin": "~~~"}方式に変換
request_header={}
if 'header' in data['request']:
for kv in data['request']['header']['headers']:
request_header[kv['name']] = kv['value']
# 不要なヘッダーを削除
del request_header['Host']
return {
"start_msec_at": run_msec_at,
"method": data['method'],
"request": '?'.join( filter(None,[data['path'],data['query']])),
"request_header":json.dumps(request_header),
"request_body":request_body,
"response_to_variable": None }
def chlsj_2_scenario(chlsj_path):
scenario = []
with open(chlsj_path) as in_file:
for request in json.load(in_file):
scenario.append( session_json_2_scenario_csv(request) )
out_file_path = os.path.basename(chlsj_path).split('.', 1)[0] + "_template.csv"
with open(out_file_path, 'w') as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=scenario[0].keys(),quotechar="'",quoting=csv.QUOTE_NONNUMERIC)
writer.writeheader()
writer.writerows(scenario)
if __name__ == '__main__':
chlsj_2_scenario(sys.argv[1])
- シナリオテンプレ(CSV)からシナリオ(CSV)を作る
- 必要に応じてシナリオ_テンプレの項目を変更し、「シナリオ」を作成する
start_msec_at | method | request | request_header | request_body | response_to_replace |
---|---|---|---|---|---|
テスト開始から何秒でコールするか ※プロトタイプでは未実装 |
HTTP メソッド GET/POST/PATCH |
REST path+クエリ | HTTP header | HTTP request body | レスポンス内容を以降のリクエストで使用したい場合、 {変数名: '値の場所'}のjsonで変数をセットする |
- 例
- 「POST /api/user_devices」をたたき、response bodyの「oauth_access_token.access_token」を「auth_token」として以降のリクエストで使用する
- header/bodyに"auth_token"の文字列が含まれる場合、「oauth_access_token.access_token」の内容にリプレースされる
- 「POST /api/user_devices」をたたき、response bodyの「oauth_access_token.access_token」を「auth_token」として以降のリクエストで使用する
start_msec_at | method | request | request_header | request_body | response_to_replace |
---|---|---|---|---|---|
0 | POST' | /api/user_devices' | {"X-LANGUAGE": "ja_JP", "X-TIME-ZONE": "Asia/Tokyo", "Authorization": "Bearer ~~~", "Accept": "/"} | {"auth_token": "oauth_access_token.access_token"}' |
負荷テストシナリオを実行する
- シナリオからlocustテストを実行する
locust起動
## 初回のみlocustをインストールする
$pip install locust
$script_csv={シナリオ}.csv locust -f scenario.py -H https://{サーバアドレス}
scenario.py
from locust import HttpUser,task, between
import os
import csv
import json
def is_json(string):
try:
json_object = json.loads(string)
except ValueError as e:
return False
return True
def xpath_get(mydict, path):
elem = mydict
try:
for x in path.strip(".").split("."):
elem = elem.get(x)
except:
pass
return elem
class ScenarioLoadTest(HttpUser):
wait_time = between(0.500, 1)
@task
def test(self):
api_list_csv_path = os.environ['script_csv']
replaces = {}
with open(api_list_csv_path) as f:
replace_map={}
for row in csv.DictReader(f,quotechar="'",quoting=csv.QUOTE_NONNUMERIC):
_request = row['request']
_header = row['request_header']
_request_body = row['request_body'].encode('utf8')
# 変数のレンダー(文字リプレース)
for key in replace_map:
_request = _request.replace(key,replace_map[key])
_header = _header.replace(key,replace_map[key])
request = _request
header = json.loads(_header)
request_body = _request_body
response = None
if row['method']=='GET':
self.client.get(request,headers=header)
elif row['method']=='POST':
raw_response = self.client.post(request,headers=header,data=request_body)
response = raw_response.json()
elif row['method']=='PATCH':
self.client.patch(request,headers=header,data=request_body)
else:
print('called else ' + str(row))
# 変数の保存
if is_json(row['response_to_replace']):
replace_map = json.loads(row['response_to_replace'])
for key in replace_map:
replace_map[key] = xpath_get(response,replace_map[key])
今後やりたいこと
- aws device farmでアプリの表示テストを動かしたい
- そのついでにスクリプトを自動生成できない?