locust使用post请求的两个实例---下单

上一篇 / 下一篇  2018-11-06 14:37:30 / 个人分类:Python

1、下单,请求字段参数化
# coding=utf-8
fromlocustimportHttpLocust, TaskSet, task
fromCrypto.PublicKeyimportRSA
fromCrypto.CipherimportPKCS1_v1_5asCipher_pkcs1_v1_5
fromCrypto.SignatureimportPKCS1_v1_5asSignature_pkcs1_v1_5
fromCrypto.HashimportMD5
fromCrypto.HashimportSHA256
importbase64
importrequests
importjson
importsys
importbinascii
importQueue

reload(sys)
sys.setdefaultencoding('utf-8')
pv ="""-----xxxx-----"""

pb ='''-----xxxx-----'''

pv1 ="""-----xxxx-----"""

pb1 ='''-----xxxx-----'''


# 私钥签名
defsign_pv(data):
key = RSA.importKey(pv1)
h = SHA256.new(data)
signer = Signature_pkcs1_v1_5.new(key)
signature = signer.sign(h)
returnbase64.b64encode(signature)


# 公钥加密,转为16进制
defcipher(data):
key = RSA.importKey(pb1)
cipher = Cipher_pkcs1_v1_5.new(key)
# cipher_text=base64.b64encode(cipher.encrypt(data))
cipher_text = binascii.hexlify(cipher.encrypt(data))
returncipher_text

classtest_126(TaskSet):
# task装饰该方法为一个事务方法的参数用于指定该行为的执行权重。参数越大,每次被虚拟用户执行概率越高,不设置默认是1,
@task()
deftest(self):
try:
data2 =self.locust.queueData.get()#获取队列里的数据
printdata2
except:
print('no data exist')
exit(0)
data = {
"amount":"1",
"cert_type":"01",
"cert_no":"xxxx",
"acct_type":"0",
"api_user_no":"001390",
"txn_type":"12",
"pc_no":"10001",
"settle_period":"0",
"ext_order_no": data2['ext_order_no'], #该字段需要唯一,所以参数化
"bank_name":"招商银行",
"acct_no":"6214xxxx77187",
"acct_name":"陈x",
"phone_no":"15xx3",
"remark":"hxxx啦",
"call_back_url":"http://fenxxxxxxx/pab"
}

url=''
url1 ="http://xxxx/trade/pay"

# 对cert——no,acct-no,acct_name,phone_no加密

ifdata.get("cert_no") !=None:
cipher_text1 = cipher(data.get("cert_no"))
data["cert_no"] = cipher_text1

ifdata.get("acct_no") !=None:
cipher_text2 = cipher(data.get("acct_no"))
data["acct_no"] = cipher_text2

ifdata.get("acct_name") !=None:
cipher_text3 = cipher(data.get("acct_name"))
data["acct_name"] = cipher_text3

ifdata.get("phone_no") !=None:
cipher_text4 = cipher(data.get("phone_no"))
data["phone_no"] = cipher_text4

# 按照键值进行排序,降序

da1 =sorted(data.items(),key=lambdax: x[0],reverse=False)

# 拼接字符串
str1 = []
forkey, valueinda1:
str1.extend([key +"="+ value])

end_date ="&".join(str1)

# print 'end_date>>>',end_date

# 签名
raw_data = end_date
sign_data = sign_pv(raw_data)

####把加密结果组成字典
dict2 = {}
d1= dict2.setdefault('sign', sign_data)

##添加含有sign的字符串到原始参数d中

new_data= data.update(dict2)
# print 'data1>>>>>>',data
data1 = json.dumps(data)

# 定义requests的请求头
header_dict = {"Content-Type":"application/json"}
# r是包含所有响应内容的一个对象
r =self.client.post(url=url1,data=data1,headers=header_dict)
# 这里可以使用assert断言请求是否正确,也可以使用if判断
assertr.status_code ==200



# 这个类类似设置性能测试,继承HttpLocust
classwebsitUser(HttpLocust):
# 指向一个上面定义的用户行为类
task_set = test_126
# 执行事物之间用户等待时间的下界,单位毫秒,相当于lr中的think time
min_wait =3000
max_wait =6000
queueData = Queue.Queue()
foriinrange(10):
data2 = {
"ext_order_no":"20181105%d"% i,
}
queueData.put_nowait(data2)#值存入队列

TAG:

 

评分:0

我来说两句

Open Toolbar