Locust性能测试--接口加密参数关联

前言
登陆接口:
headers：时间戳--timestamp实时获取，clientId--android/ios/web/微信小程序固定值，sign--privatekey加密后得到

查询报告接口:

headers：时间戳--timestamp实时获取，clientId--android/ios/web/微信小程序固定值，sign--privatekey加密后得到，token--登录返回的参数

思路
1、解决sign加密问题：_dispose_params()、_rsa()；
2、只登陆一次，拿到token后，后续查报告接口都使用同一个token（短期内不会失效，开发设置的5天）
3、查报告及登

示例代码

import time
import rsa
import base64
from locust import HttpLocust, TaskSet, task

# 将参数按照规则排序
def _dispose_params(dict, timestamp=None):
    if not timestamp:
        timestamp = int(time.time() * 1000)
    string = 'timestamp={}'.format(timestamp)
    if not dict:
        return string
    sorted_list = sorted(dict.items(), key=lambda item: item[0], reverse=False)
    l = []
    for i in sorted_list:
        if i[1] or i[1] == 0:
            l.append(i)
    if not l:
        return string
    sorted_str = [str(x[0]) + "=" + str(x[1]) for x in l]
    string = string + "&" + '&'.join(sorted_str)
    return string

# 使用rsa非对称进行签名
def _rsa(message, privite_key):
    privkey = rsa.PrivateKey.load_pkcs1(privite_key)
    crypto_text = rsa.sign(message.encode(), privkey, 'MD5')
    crypto_text = base64.b64encode(crypto_text)
    return crypto_text

#获得请求头
privite_key = """-----BEGIN RSA PRIVATE KEY-----
               MIICXwIBAAKBgQDxYdFgdsPG2vG8V2RyFjWErp8vRlF3hjAJgXSju786gvujmi5R
               kS9aE1vWqval9oaEufeJ44Bq/FtlXCI/WpuiMdhvre2TEgUwQiRvBTHitt/nn93Q
               /mgm1P8Rh1xTl/Fqbz+QPzv5m8Q+/cbEYJHU5DSRUcYj7cGWWMbKaPWnEQIDAQAB
               AoGBAJuzfz18HnkNDdVxgeddkBQQF7qE31P8CMx45uYe2xFBIFS9xZbz0dmGlzz3
               drZDOS4wTw5w44sfUU0miyjWp4BNa/QFbefVf3urQUiChPi+dbg/g4484Gymjk30
               cPDQeNGzkGNeejmU06YUCUe3YBsC3pglIMHagaH3gUhpuhyBAkEA/OnAnfLkFm4P
               O/Vk9PWhaGoQNMBS2/WA5OPPHqkQt8Hkspa+pBcmZJpgPmed/5+hd7XhReM3xcwc
               t/3nThLT6QJBAPRUCTMdcqjAh/Fh8fmT9WtkTL8um5MtZdBbRwsMihIJ5BtuXPrB
               Ex8N5jXKEkgklQxkjp7XaSBqo3deONLniOkCQQCCMLH+nLHc+frNmpzFnEviO/br
               sLT2lYIsUTilR9JbRWw8omywe1pAH1e7yAsFM18jCHVWTIjBQTjnXcRb5EVBAkEA
               nYuunvInJB95yTFCCY/ZmZUEU+bm1IFafH7PEZXVVgpgtztz4xEqlkStJBQXs8M7
               BNG2kUAPfqKko2zzMTOxOQJBAN8tCqM0ezh1ahRI21RHgg3QdgSrlaQsDXtk4AWC
               CBM8ep9GRp1vzT6EyMPSNyeySXSb7LQa0Pl3dpSNRter3Fs=
               -----END RSA PRIVATE KEY-----
               """
timestamp = str(int(time.time() * 1000))
message = _dispose_params({}, timestamp)
sign = _rsa(message, privite_key)
headers = {"clientId": "8b4701753a65837e3c48d3af9b3a91bc",
           "timestamp": timestamp,
           "sign": sign}
print("请求头是：", headers)

class testReport(TaskSet):
    # 登录
    def doLogin(self):
        login_url = "/cpcs-server/api/v1/login"
        login_data = [{"account": "15611111111", "client": "android", "credential": "e10adc3949ba59abbe56e057f20f883e","identityType": "passwd", "isFirstPart": "1"}]
        print("登录前请求头是", headers)
        response = self.client.post(login_url, json=login_data[0], headers=headers, verify=False).json()
        print("登录后接口的请求头是：", headers)
        print("LOGIN RESULT:", response)
        return response

    # on_start() 在测试前的初始化，先登录且只登陆一次
    def on_start(self):
        print("--------------性能测试来咯！--------------")
        response = self.doLogin()
        token = response["respData"]
        headers["token"] = token
        print("登录后更新的请求头是：", headers)

    # 查询报告列表
    @task
    def test_get_report_list(self):
        report_url = "/cpcs-server/api/v1/report/info"
        report_data = None
        print("执行查询报告的请求头是：",headers)
        response = self.client.get(report_url, headers=headers, data=report_data, verify=False).json()
        print('查询结果 :', response)

#locust类中的子类HttpLocust类中封装好了http方法，继承HttpLocust可以直接使用http相关方法
class WebSite(HttpLocust):
    task_set = testReport
    host = "https://xxx.xxx.in"
    min_wait = 1000
    max_wait = 3000

　　

执行性能测试