Jmeter使用基礎筆記-寫一個http請求

 

前言

本篇文章主要講述2個部分：

1. 搭建一個簡單的測試環境
2. 用Jmeter發送一個簡單的http請求

搭建測試環境

1. 編寫flask代碼(我參考了開源項目HttpRunner的測試服務器)，將以下的代碼保存爲api_server.py：

   import hashlib
   import hmac
   import json
   from functools import wraps
   
   from flask import Flask, make_response, request
   from httprunner.built_in import gen_random_string
   
   try:
       from httpbin import app as httpbin_app
       HTTPBIN_HOST = "127.0.0.1"
       HTTPBIN_PORT = 3458
       HTTPBIN_SERVER = "http://{}:{}".format(HTTPBIN_HOST, HTTPBIN_PORT)
   except ImportError:
       httpbin_app = None
       HTTPBIN_HOST = "httpbin.org"
       HTTPBIN_PORT = 443
       HTTPBIN_SERVER = "https://{}:{}".format(HTTPBIN_HOST, HTTPBIN_PORT)
   
   FLASK_APP_PORT = 5000
   SECRET_KEY = "DebugTalk"
   
   app = Flask(__name__)
   
   """ storage all users' data
   data structure:
       users_dict = {
           'uid1': {
               'name': 'name1',
               'password': 'pwd1'
           },
           'uid2': {
               'name': 'name2',
               'password': 'pwd2'
           }
       }
   """
   users_dict = {}
   
   """ storage all token data
   data structure:
       token_dict = {
           'device_sn1': 'token1',
           'device_sn2': 'token1'
       }
   """
   token_dict = {}
   
   
   def get_sign(*args):
       content = ''.join(args).encode('ascii')
       sign_key = SECRET_KEY.encode('ascii')
       sign = hmac.new(sign_key, content, hashlib.sha1).hexdigest()
       return sign
   
   def gen_md5(*args):
       return hashlib.md5("".join(args).encode('utf-8')).hexdigest()
   
   
   def validate_request(func):
   
       @wraps(func)
       def wrapper(*args, **kwargs):
           device_sn = request.headers.get('device_sn', "")
           token = request.headers.get('token', "")
   
           if not device_sn or not token:
               result = {
                   'success': False,
                   'msg': "device_sn or token is null."
               }
               response = make_response(json.dumps(result), 401)
               response.headers["Content-Type"] = "application/json"
               return response
   
           if token_dict[device_sn] != token:
               result = {
                   'success': False,
                   'msg': "Authorization failed!"
               }
               response = make_response(json.dumps(result), 403)
               response.headers["Content-Type"] = "application/json"
               return response
   
           return func(*args, **kwargs)
   
       return wrapper
   
   
   @app.route('/')
   def index():
       return "Hello World!"
   
   @app.route('/api/get-token', methods=['POST'])
   def get_token():
       user_agent = request.headers.get('User-Agent', "")
       device_sn = request.headers.get('device_sn', "")
       os_platform = request.headers.get('os_platform', "")
       app_version = request.headers.get('app_version', "")
       data = request.get_json()
       sign = data.get('sign', "")
   
       expected_sign = get_sign(user_agent, device_sn, os_platform, app_version)
   
       if expected_sign != sign:
           result = {
               'success': False,
               'msg': "Authorization failed!"
           }
           response = make_response(json.dumps(result), 403)
       else:
           token = gen_random_string(16)
           token_dict[device_sn] = token
   
           result = {
               'success': True,
               'token': token
           }
           response = make_response(json.dumps(result))
   
       response.headers["Content-Type"] = "application/json"
       return response
   
   @app.route('/api/users')
   @validate_request
   def get_users():
       users_list = [user for uid, user in users_dict.items()]
       users = {
           'success': True,
           'count': len(users_list),
           'items': users_list
       }
       response = make_response(json.dumps(users))
       response.headers["Content-Type"] = "application/json"
       return response
   
   @app.route('/api/reset-all')
   @validate_request
   def clear_users():
       users_dict.clear()
       result = {
           'success': True
       }
       response = make_response(json.dumps(result))
       response.headers["Content-Type"] = "application/json"
       return response
   
   @app.route('/api/users/<int:uid>', methods=['POST'])
   @validate_request
   def create_user(uid):
       user = request.get_json()
       if uid not in users_dict:
           result = {
               'success': True,
               'msg': "user created successfully."
           }
           status_code = 201
           users_dict[uid] = user
       else:
           result = {
               'success': False,
               'msg': "user already existed."
           }
           status_code = 500
   
       response = make_response(json.dumps(result), status_code)
       response.headers["Content-Type"] = "application/json"
       return response
   
   @app.route('/api/users/<int:uid>')
   @validate_request
   def get_user(uid):
       user = users_dict.get(uid, {})
       if user:
           result = {
               'success': True,
               'data': user
           }
           status_code = 200
       else:
           result = {
               'success': False,
               'data': user
           }
           status_code = 404
   
       response = make_response(json.dumps(result), status_code)
       response.headers["Content-Type"] = "application/json"
       return response
   
   @app.route('/api/users/<int:uid>', methods=['PUT'])
   @validate_request
   def update_user(uid):
       user = users_dict.get(uid, {})
       if user:
           user = request.get_json()
           success = True
           status_code = 200
           users_dict[uid] = user
       else:
           success = False
           status_code = 404
   
       result = {
           'success': success,
           'data': user
       }
       response = make_response(json.dumps(result), status_code)
       response.headers["Content-Type"] = "application/json"
       return response
   
   @app.route('/api/users/<int:uid>', methods=['DELETE'])
   @validate_request
   def delete_user(uid):
       user = users_dict.pop(uid, {})
       if user:
           success = True
           status_code = 200
       else:
           success = False
           status_code = 404
   
       result = {
           'success': success,
           'data': user
       }
       response = make_response(json.dumps(result), status_code)
       response.headers["Content-Type"] = "application/json"
       return response

2. 啓動測試服務器：

   $ export FLASK_APP=tests/api_server.py    # 路徑填寫本身保存的api_server.py絕對路徑
   $ flask run
    * Serving Flask app "tests.api_server"
    * Running on http://127.0.0.1:5000/ (Press CTRL+C to quit)

3. 測試服務器啓動起來以後下面就能夠開始寫Jmeter的http請求了

使用Jmeter編寫Http請求

1. 先看下接口實例：

   url:
       /api/get-token/
   
   method: 
       POST
   
   Headers:
       app_version:2.8.6
       Content-Type:application/json
       os_platform:ios
       user_agent:iOS/10.3
       device_sn:FwgRiO7CNA50DSU
   
   body:
       {"sign": "958a05393efef0ac7c0fb80a7eac45e24fd40c27"}
   
   response:
       {
           "success": true,
           "token": "qba756V9z6rOYOVH"
       }

2. 使用postman請求實例：
   headers:
   

Body(注意：這裏選擇JSON(application/json)):

4. 知道了如何去發接口請求，如今就能夠用Jmeter寫腳本啦
   3.1 新建線程組，全部參數都是默認值便可
   3.2 添加配置文件：HTTP信息請求頭管理，添加以下參數

Headers:
    app_version:2.8.6
    Content-Type:application/json
    os_platform:ios
    user_agent:iOS/10.3
    device_sn:FwgRiO7CNA50DSU

3.3 添加Sampler=>HTTP請求，參數如圖所示

3.4 添加監聽器=>查看結果樹（注意：是在HTTP請求這個Sampler的節點下），運行腳本結果以下，則腳本接口請求成功

下一篇文章中將介紹如何作斷言。

Jmeter使用基礎系列文章大綱：
Jmeter使用基礎筆記-認識Jmeter
Jmeter使用基礎筆記-寫一個http請求
Jmeter使用基礎筆記-API接口返回數據斷言