概要

iOSアプリのバックエンドAPIに対して、負荷テストをしたい
- アプリの使用開始時の挙動(APIコール内容)をシナリオ的に検証したい
  - API:Aにコールが集中している状況でAPI:Bをコールすると発生する問題がある

アプリの通信内容をキャプチャして、テストシナリオを自動で生成できないか？
- 経験則からテスト対象APIを絞ることはできるが、意図しない組合せでAPIシステムが落ちることを防げない
- アプリのテストケースはない前提
プロトタイプ的にかんたんに動くものを作ってみた

注意事項

負荷テストは自身が管理するシステム以外に実行しないでください

登場人物

アプリ
- APIコールをするやつ
charles for ios
- アプリの通信内容をキャプチャするアプリ
APIサーバ
- アプリのバックエンド: REST APIサーバ
locust
- 負荷試験ツール
僕: システムエンジニア
- 許されることなら何もしたくない

やりたいこと

概要

負荷テストシナリオを生成する

負荷テストシナリオを実行する

詳細

負荷テストシナリオを生成する

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」の内容にリプレースされる

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でアプリの表示テストを動かしたい
そのついでにスクリプトを自動生成できない？