接⼝测试python脚本，接⼝并发python脚本
⼀、接⼝测试python脚本
写了很久了，⼀直没发博客，今⼉多发⼏篇，先上代码。
#__author__ = 'chubby_superman'
#_*_coding=utf-8 _*
import requests
from datetime import datetime
class Httpreq():
def__init__(self,url,data,heards):
self.url=url
self.data=data
self.heards=heards
def req_1(self):
requests_statuscode=requests.post(url=self.url,json=self.data).status_code
try:
if requests_statuscode ==200:
print(1)
#return "请求成功，开始判断响应状态"
else:
#return requests_statuscode
excel中的index函数怎么用print("请求失败if内")
except Exception as e:
print("请求失败try内")
php下载txt
x = requests.post(url=self.url,json=self.data)
try:
a = x.text
with w().strftime('%Y%m%d%H%M%S')+'.txt','w',encoding="utf-8")as f:
f.write(a)
except Exception as e:
print("写⼊⽂件有问题")
a = x.json()["code"]
if a ==1000:
print("正常返回")
elif a ==-1:
print("系统繁忙")
elif a ==1001:
print("partnerId⽆效")
elif a ==1002:
print("IP⽩名单错误，禁⽌服务")
elif a ==1003:
print("签名错误")
京东python入门教程elif a ==1004:
print("缺少必要的参数")
else:
print("返回未知的错误")
if __name__ =='__main__':
the_url='test.pipifit/analysisShopCartOverCate'
#接⼝url
the_data={'day':1,'period':1,'goodsInfo':[{"goodsId":1213,"goodsNumber":1}],'sex':0,'symptoms':1001}
#请求body
Httpreq(the_url,the_data,'').req_1()
下⾯ 为接⼝并发测试脚本，使⽤的是多线程，⽽⾮协程（也就是说，⽐较鸡肋）。
⼆、接⼝并发
脚本1，。该脚本功能为：向京东开普勒发送post请求
class Http_request():
def__init__(self,url,**kwargs):
self.url = url
self.kwargs=kwargs
自媒体标签大全def update_port_datas(self):
datas ={
"method":"biz.product.sku.query",
"app_key":r'appkey',#需要从京东获得
'access_token':r'token',#需要从京东获得
'timestamp':'2019-02-1416:13:10',
'v':'1.0',
'format':'json',
'param_json':'{"name":"测试商品池","pageNum":"1"}'
}
try:
for k in self.kwargs:
datas[k]=self.kwargs[k]
except Exception as e:
print(e)
return datas
def post_request(self):
datas =self.update_port_datas()
try:
post_response =requests.post(url=self.url,data=datas)
return
except Exception as e:
print(e)
if __name__=="__main__":
url ="router.jd/api"
a = Http_request(url,method="biz.product.PageNum.query",param_json='{}').post_request() print(a)
脚本2，。该脚本主要功能：建⽴线程。
import time
#定义⼀个⽅法
def tr(a,b):
print([x for x in range(a+b)])
pass
class Mythread(threading.Thread):
#使⽤继承实现多线程
def__init__(self,func):
threading.Thread.__init__(self)
self.func=func  #功能名
#self.args=args  #功能所需参数反函数的定义及求解
#初始化⽗类  threading.Thread
def run(self):
lock = threading.Lock()
if lock.acquire():
self.func  #调⽤run⽅法，实现功能的多线程
微社区print(self.ime())
#给需要实现多线程的功能加锁。注意：run⽅法为重写⽗类threading.Thread的run⽅法，函数名不得改变，且此⽅法必须有。if __name__=="__main__":
thread_list=[]
for i in range(5):#需要开启多少线程，range中的数字填多少，此处需要5个
t = thread_list.append(Mythread(tr(3,4)))
for t in thread_list:
t.start()
for t in thread_list:
t.join()
脚本3，。该脚本功能为：设定线程数，开始并发post请求。
from kepler import Http_request
from test import Mythread
import gevent
import requests
def tr():
url ="router.jd/api"
a = Http_request(url,method="biz.product.PageNum.query",param_json='{}').post_request() print(a)
def calc():
url ='te.pipifit/goods/list'
res = (url,params={'foodId':1},headers={"token":"a93c0fea8150ef04b96ac90ccb8a"}) )
def bing_fa(x):
c =0
while1:
calc()
c+=1
if c>=x:
break
def thread_func(x,tr):
thread_list=[]
for i in range(x):#需要开启多少线程，range中的数字填多少，此处需要5个
t = thread_list.append(Mythread(tr))
for t in thread_list:
t.start()
for t in thread_list:
t.join()
#
gevent.spawn(thread_func(2,bing_fa(500000)))

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。