1、循環取可以重復使用的數據
csvData.csv
1,1
2,2
3,3
4,4
5,5
6,6
7,7
8,8
9,9
locust2.py
from locust import TaskSet, task, HttpUser, HttpLocust, between, tag, constant
import csv
class UserBehavior(TaskSet):
def on_start(self):
self.index = 0
# @tag('getJoke分頁查詢')
@task
def test_visit(self):
self.index = (self.index + 1 ) % len(self.parent.share_data)
url = "/getJoke?page=" + self.parent.share_data[self.index] + "&count=10&type=video" + self.parent.share_data[self.index]
print('visit url: %s' % url, self.parent.share_data)
self.client.get(url, name='getJoke分頁查詢')
class WebsiteUser(HttpUser):
host = 'https://api.apiopen.top'
# task_set = UserBehavior
tasks = [UserBehavior]
# my_data = [1, 3, 5]
share_data = []
with open('csvData.csv', newline='') as csvfile:
# 此方法:當文件不用時會自動關閉文件
csvReader = csv.reader(csvfile)
for content in csvReader:
# print(content)
share_data.append(content[1])
# min_wait = 1000
# max_wait = 3000
wait_time = between(1, 3)
2、不循環取數據，測試數據唯一
使用隊列存儲數據
from locust import TaskSet, task, HttpUser
import queue
class UserBehavior(TaskSet):
@task
def test_register(self):
try:
data = self.parent.user_data_queue.get()
except queue.Empty:
print('account data run out, test ended.')
exit(0)
print('register with user: {}, pwd: {}'.format(data['username'], data['password']))
payload = { 'username': data['username'], 'password': data['password'] }
self.client.post('/register', data=payload)
class WebsiteUser(HttpUser):
host = 'http://debugtalk.com'
# task_set = UserBehavior
tasks = [UserBehavior]
user_data_queue = queue.Queue()
for index in range(100):
data = { "username": "test%04d" % index,
"password": "pwd%04d" % index,
"email": "test%04d@debugtalk.test" % index,
"phone": "186%08d" % index,
}
user_data_queue.put_nowait(data)
min_wait = 1000
max_wait = 3000
3、循環取數據，測試數據唯一
from locust import TaskSet, task, HttpLocust
import queue
class UserBehavior(TaskSet):
@task
def test_register(self):
try:
data = self.parent.user_data_queue.get()
except queue.Empty:
print('account data run out, test ended.')
exit(0)
print('register with user: {}, pwd: {}'.format(data['username'], data['password']))
payload = { 'username': data['username'], 'password': data['password'] }
self.client.post('/register', data=payload)
self.parent.user_data_queue.put_nowait(data) # 將取出的數據重新寫入隊列
class WebsiteUser(HttpLocust):
host = 'http://debugtalk.com'
# task_set = UserBehavior
tasks = [UserBehavior]
user_data_queue = queue.Queue()
for index in range(100):
data = { "username": "test%04d" % index,
"password": "pwd%04d" % index,
"email": "test%04d@debugtalk.test" % index,
"phone": "186%08d" % index, }
user_data_queue.put_nowait(data)
min_wait = 1000
max_wait = 3000
個人博客 蝸牛