locust 實戰：用戶登錄並下單
涉及內容：
1、session信息關聯
2、用戶數據參數化
3、對響應數據進行檢查點設置
具體代碼如下：
#-*-encoding:utf-8-*-
from locust import HttpLocust,TaskSet,task
from locust.clients import HttpSession
import re,random,json
import MyLog
"""
MyLog 自定義日志
使用HttpSession對象提交接口請求進行session信息的關聯，
然后通過json或re 對頁面內容進行提取，並把返回的結果作為檢查點進行比較
思路：
1.將下單的每個步驟對應的請求分別封裝到不同的方法中，並在方法中設置檢查點進行判斷響應結果是否正確
2.將用戶信息在on_start()方法中初始化，如果數據量少的話，可以直接在腳步中直接配置，如果用戶數據量大的話，可以放到txt文檔中，再on_start()中讀取即可
3.將接口中的用戶信息參數進行參數化
4.按下單流程的各個步驟對應的方法，按順序進行調用
5.調用壓測腳本
"""
class loginHttpSession(TaskSet):
#登錄並購買商品壓力測試
def on_start(self):
"""
測試數據初始化
1.創建session會話，提供給后面的請求
2.初始化請求頭的信息
3.初始化用戶信息（用戶信息參數化）
4.引用自定義日志
"""
self.session=HttpSession(base_url="http://localhost:8081/tshop/")
self.header={
"User-Agent": "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36",
"Accept-Language": "zh-CN,zh;q=0.9",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
"Accept-Encoding": "gzip, deflate, br"
}
self.userinfo={
0:["123@qq.com","123456"],
1:["223@qq.com","123456"],
2:["333@qq.com","123456"]
}
self.log=MyLog.MyLog().get_logger()
def get_user_info(self):
#獲取隨機用戶ID
user_id=random.randint(0,len(self.userinfo)-1)
return user_id
def index(self):
#打開商城首頁
return self.session.get("/")
def input_username(self,loginToken,uname):
#進入輸入用戶名頁面
#需要傳送session和登錄token值
#返回響應信息response，和userId
input_url="index.php"
param={
"con":"ajax",
"act":"account",
# "account":"123@qq.com"
"account":uname, #參數化用戶名信息
}
# self.log.debug("[input_username]:userId=%s" %userId) #打印日志
# response=self.session.post(input_url,params=param,headers=self.header)
with self.session.post(input_url,params=param,headers=self.header,catch_response=True) as response:
#獲取請求詳情數據，並設置檢查點
print response.text
response_value=response.text
print type(response_value)
#由於返回的數據是unicode類型，也就是字符串，所以只能截取有效數據進行后面的檢查點的判斷
status=response_value[response_value.find('"status":')+len('"status":'):response_value.find(",")]
msg=response_value[response_value.find('"msg":"')+len('"msg":"'):response_value.find('"}')]
if status!="false":
response.failure("input_username: status error!")
elif msg!="\u6b64\u7528\u6237\u5df2\u7ecf\u6ce8\u518c":
response.failure("input_username: msg error!")
return response,uname
def submit_login(self,loginToken,uname,upwd):
#輸入用戶名和密碼提交登錄
#需要傳遞loginToken，和userId用於用戶登錄
#返回響應信息response，和userId
login_url="index.php?con=simple&act=login_act"
bodys={
"redirectURL":"/tshop/",
# "account":"123@qq.com",
# "password":"123456",
"account":uname, #參數化用戶名
"password":upwd, #參數化密碼
"tiny_token_login":loginToken,
}
# self.log.debug("[submit_login]:userId=%s" %userId)
with self.session.post(login_url,data=bodys,headers=self.header,catch_response=True) as response:
text=response.text
page_text=re.compile(r'<div class="sub-1">(.+)?-')
page_userinfo=page_text.search(text)
userinfo=page_userinfo.group(1)
print userinfo
print uname
if userinfo.find(uname)<0: #str的find方法如果找不到元素，返回值為-1，所以判斷小於0表示，在userinfo中無登錄的用戶名
response.failure("submit_login: not found username in page!")
return response,uname
def go_goods_details(self):
#進入商品詳情頁面
#返回響應數據
goodsDetails_url="index.php"
params={
"con":"index",
"act":"product",
"id":"16"
}
# response=self.session.get(goodsDetails_url,data=params,headers=self.header)
with self.session.get(goodsDetails_url,data=params,headers=self.header,catch_response=True) as response:
text=response.text
page_text=re.compile(r'</span><b class="cart-total red" >(.+)?</b></span></span></div>')
page_price=page_text.search(text)
price=page_price.group(1)
if price <100:
response.failure("go_goods_details: not found price!")
return response
def add_goods_tips(self,loginToken=None):
#在商品詳情頁面，點擊進入到確認訂單頁面
#返回用戶購物車清單的商品信息
goods_tips_url="index.php?con=index&act=goods_add&id=61&num=3"
with self.session.get(goods_tips_url,headers=self.header,catch_response=True) as response:
text=response.text
page_text=re.compile(r"<li >(.+)?<em></em><i></i></li>")
page_info=page_text.search(text)
value=page_info.group(1)
if value!="4、訂購完成":
response.failure("add_goods_tips: not found 4、訂購完成")
return response
def fill_order_info(self):
#進入到填寫訂單信息頁面
#使用正則表達式，抓取頁面返回的order_Token
#返回響應數據與order_Token
fill_order_url="index.php?con=simple&act=order&cart_type=cart"
with self.session.get(fill_order_url,headers=self.header,catch_response=True) as response:
order_re=re.compile(r"'tiny_token_order' value='(.+)?'/>")
order_token=None #order_token賦默認值為None，如果取不到order_token以便下面進行判斷，為None則拋出錯誤
order_token=order_re.search(response.text).group(1)
if order_token==None:
response.failure("fill_order_info: not found order_token")
return response,order_token
def submit_order(self,orderToken,uname):
#提交訂單
#需傳遞填寫訂單信息頁面的orderToken，對提交訂單進行參數化
#根據userId，登錄的userId來區分不同用戶的收貨地址id
#返回響應信息，與orderToken
sub_order_url="index.php?con=simple&act=order_act"
bodys={
"address_id":"30",
"payment_id":"1",
"user_remark":"",
"invoice_type":"0",
"invoice_title":"",
"cart_type":"",
"tiny_token_order":orderToken
}
#根據不同的userId更新參數中的地址id，確保每個用戶的收貨地址正確
if uname=="123@qq.com":
bodys["address_id"]="30"
elif uname=="223@qq.com":
bodys["address_id"]="31"
elif uname=="333@qq.com":
bodys["address_id"]="32"
else :
print "userId not found !!!"
# self.log.debug("[submit_order]:userId=%s,address_id=%s" %(userId,bodys["address_id"]))
with self.session.post(sub_order_url,data=bodys,headers=self.header,catch_response=True) as response:
text=response.text
textview=re.compile(r'<span><i class="icon-success-48"></i>(.+)?</span>')
re_result=textview.search(text)
page_info=re_result.group(1)
# if page_info!="訂單已成功提交！":
# response.failure("submit_order:not found 訂單已成功提交！")
return response,orderToken
def logout(self):
#下單成功后，退出登錄
logout_url="index.php?con=simple&act=logout"
with self.session.get(logout_url,headers=self.header) as response:
return response
@task
def go_shopping(self):
#購買商品流程：登錄->進入商品詳情頁面->將商品添加到購物車->進入填寫訂單信息頁面->提交訂單
user_id=self.get_user_info() #獲取用戶id
username=self.userinfo[user_id][0]
upassword=self.userinfo[user_id][1]
self.index() #進入商城首頁
laster_url="index.php?con=simple&act=login"
first_reponse=self.session.get(laster_url,headers=self.header)
lg_page=first_reponse.text
re_token=re.compile(r"name='tiny_token_login' value='(.+)?'/>")
#通過正則表達式獲取登錄頁面的token信息
re_result=re_token.search(lg_page)
lg_token=None
if re_result:
lg_token=re_result.group(1)
# #輸入用戶名頁面,並獲取login_token 信息
inp_username_response=self.input_username(loginToken=lg_token,uname=username)
inp_username_userId=inp_username_response[1] #返回輸入用戶名時，參數化的userId
# # #輸入賬號、密碼，並提交登錄
sub_login_response=self.submit_login(loginToken=lg_token,uname=username,upwd=upassword)
sub_login_userId=sub_login_response[1] #返回用戶登錄時，參數化的userId
# #進入到商品詳情頁面
goods_details_response=self.go_goods_details()
# # print goods_details_response.text
# #進入購物車清單頁面
goods_tips_response=self.add_goods_tips()
# # print goods_tips_response.text
# #點擊購物車清單頁面的-【立即支付】，跳轉到填寫訂單信息頁面
fill_order_response=self.fill_order_info()
orderToken_value=fill_order_response[1]
# #點擊填寫訂單信息頁面【確認支付】按鈕
sub_order_response=self.submit_order(orderToken=orderToken_value,uname=username)
#退出用戶登錄狀態
logout=self.logout()
class WebUserLocust(HttpLocust):
task_set=loginHttpSession
創建執行腳本：
#-*-encoding:utf-8-*-
import os
os.system("locust.exe -f F:\\testAndStudy\\loginHttpSession.py --host=http://localhost:8081/tshop/")
運行結果：
看看異常的記錄日志：
再看看，我自己設置的locust運行中打印的登錄用戶信息：
最后，登錄商城查看是否有新的訂單：
（特此說明一下，為什么訂單時間有時候一致，有時候不一致。這是因為我在參數化時，取的是隨機用戶id，也就是說，同一秒有時候登錄相同的賬號進行下單，或者同一秒登錄兩個不同的賬號並下單。
這個其實是不影響，但覺得有必要說明一下）