前言登陆接口:

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天)：

on_start(),登陆一次之后，更新token

3、登录、查报告接口共用一个相同的headers

示例代码

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-----

****************************************************************

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 = "/api/v1/login"

login_data = [{"account": "***", "client": "android", "credential": "e10adc3949ba59abbe56e057f20f883e","identityType": "passwd", "isFirstPart": "1"}]

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 = "/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

执行性能测试

Hatching and swarming 1 clients at the rate 1 clients/s...

登录只执行了一次

总结

1、HttpLocust类从继承 Locust的类，并把它添加一个客户端属性，它是的一个实例 HttpSession，可用于使HTTP请求。

2、http协议是无状态的，所以登录请求和登录后的请求它是独立的，但是登录后的请求需要依赖先登录拿到cookies,才能保持登录状态，python接口自动化里面可以用session来解决：s = requests.session()

HttpLocust类从继承 Locust的类，并把它添加一个客户端属性，它是的一个实例 HttpSession，可用于使HTTP请求,这就相当于它自动使用了session机制，类似于client = requests.session()

所以后面的请求，直接拿client.get()、client.post()请求就可以了