TA的每日心情 | 无聊 2024-11-5 10:03 |
---|
签到天数: 77 天 连续签到: 1 天 [LV.6]测试旅长
|
目录
一、安装
二、常见使用方法
1、直接HttpUser中(常见,一般是代码量少)
2、写在模块中(少见)
3、单独写在taskset中(常见)
三、运行
四、TaskSet与SequentialTaskSet区别
场景一
场景二
五、代码执行顺顺序
六、参数化
场景一
场景二
场景三
一、安装官方:https://docs.locust.io/en/stable/installation.html - pip3 install locust
-
- # 查看版本
- locust -V
复制代码 二、常见使用方法1、直接HttpUser中(常见,一般是代码量少)- from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
- class Login(HttpUser):
- host = 'http://1.72.24.40:23'
- # wait_time = constant(3) # 每次请求停顿时间 (思考时间)
- wait_time = between(1, 3)
- @task
- def login(self):
- url = '/login'
- json = {
- "password": "45",
- "username": "34",
- }
- header = {"Content-Type": "application/json;charset=UTF-8"}
- self.client.request(method='POST', url=url, json=json, headers=header, name='登录1')
复制代码 2、写在模块中(少见)
- from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
- @task
- def login(param):
- url = '/login'
- json = {
- "password": "45",
- "username": "34",
- }
- header = {"Content-Type": "application/json;charset=UTF-8"}
- param.client.request(method='POST', url=url, json=json, headers=header, name='登录1')
-
- @task
- def home(param):
- url = '/#/login'
- param.client.request(method='GET', url=url, name='首页')
-
- class Login(HttpUser):
- host = 'http://1.72.24.40:23'
- # wait_time = constant(3) # 每次请求停顿时间 (思考时间)
- wait_time = between(1, 3)
- tasks = [login, home]
复制代码 3、单独写在taskset中(常见)
- from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
-
-
- class TaskTest(TaskSet):
- @task
- def login(self):
- url = '/login'
- json = {
- "password": "rewr",
- "username": "rewr",
- }
- header = {"Content-Type": "application/json;charset=UTF-8"}
- self.client.request(method='POST', url=url, json=json, headers=header, name='登录')
-
-
-
-
- class Login(HttpUser):
- host = 'http://s2.x.x.x:x'
- # wait_time = constant(3) # 每次请求停顿时间 (思考时间)
- wait_time = between(1, 3)
- tasks = [TaskTest]
-
复制代码 三、运行
浏览器输入:http://localhost:8089/
四、TaskSet与SequentialTaskSet区别 继承TaskSet类,那么对应里面的方法是并行执行的,继承SequentialTaskSet类,那么对应里面的方法是串行执行的。 场景一
- 例如:登录一次,添加50次,那么可以考虑使用继承SequentialTaskSet类,然后在装饰器task当中进行设置
复制代码
场景二
- 配合on_start、on_stop(这里例子没有使用到)使用askSet然后通过配置task。on_start、on_stop是每个用户开始和结束的时候都会执行一次。
- 模拟100个用户,先登录系统,然后90%执行A操作,10%执行B操作
复制代码- # -*- coding: utf-8 -*-
- # @Time : 2021/6/14 8:17
- # @Project : virlocustdemo
- # @Author : testing
- # @File : locus2ss.py
- # @Software: PyCharm
- from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
-
-
- class TaskTest(TaskSet):
- """
- 这里继承的是串行执行的类
- """
- token = None
-
- # 用户执行task前调用
- def on_start(self):
- url = '/XX/XX/login'
- json = {
- "password": "888X23338",
- "username": "XXXX"
- }
- header = {"Content-Type": "application/json}
- try:
- with self.client.post(url=url, json=json, headers=header, name='登录') as response:
- self.token = 'Bearer' + ' ' + res['data']['access_token']
- except Exception as e:
- print("错误信息为", e)
- # 用户执行task后调用
- def on_stop(self):
- print("用户结束")
- @task(90)
- def query_users(self):
- url = '/DSAD/userMDSADe/DS'
- json = {
- "pageIndex": "1",
- "pageSize": "200"
- }
- header = {
- "Content-Type": "application/json,
- "Authorization": self.token}
- try:
- self.client.get(url=url, params=json, headers=header, name='A')
- except Exception as e:
- print("错误信息为", e)
-
-
- @task(10)
- def query_oErg_tEWee(self):
- url = '/uWEW/usEWEWe/orEWEree'
- json = {"enRdLeERvTel": "1"}
- header = {
- "Content-Type": "application/json,
- "Authorization": self.token}
- try:
- self.client.get(url=url, params=json, headers=header, name='B')
- except Exception as e:
- print("错误信息为", e)
- class Login(HttpUser):
- host = 'http://X2.122.123.20:138'
- # wait_time = constant(3) # 每次请求停顿时间 (思考时间)
- wait_time = between(1, 3)
- tasks = [TaskTest]
复制代码 五、代码执行顺顺序测试项目执行前会执行一次SETUP,测试项目结束后会执行一次TEARDOWN。 - import time
- from locust import HttpUser,task,between,events
- import urllib3
- urllib3.disable_warnings()
-
- @events.test_start.add_listener
- def on_test_start(**kwargs):
- print('===测试最开始提示===')
-
- @events.test_stop.add_listener
- def on_test_stop(**kwargs):
- print('===测试结束了提示===')
-
-
- class TestTask(HttpUser):
-
- wait_time = between(1, 5)
- host = 'https://www.baidu.com'
- def on_start(self):
- print('这是SETUP,每次实例化User前都会执行!')
-
- @task(1)
- def getBaidu(self):
- self.client.get(url="/",verify=False)
-
- def on_stop(self):
- print('这是TEARDOWN,每次销毁User实例时都会执行!')
-
-
- if __name__ == "__main__":
- import os
- os.system("locust -f testpy.py --headless -u 10 -r 10 -t 2")
复制代码 六、参数化场景一 模拟3个用户并发请求网页,共有100个URL地址,每个虚拟用户都会依次循环加载100个URL地址
- <font size="3" color="#000000">
- # -*- coding: utf-8 -*-
- """
- @ auth : carl_DJ 来自博主
- @ time : 2020-9-23
- """
-
-
- from locust import TaskSet, task, HttpUser
-
- class UserBehav(TaskSet):
- def on_start(self):
- self.index = 0
-
- @task
- def test_visit(self):
- url = self.user.share_data[self.index]
- # url = self.locust.share_data[self.index]
- print('visit url: %s' % url)
- self.index = (self.index + 1) % len(self.locust.share_data)
- self.client.get(url)
-
- class WebsiteUser( HttpUser):
- host = 'http://www.xxx.com'
- task_set = task(UserBehav)
- share_data = ['url1', 'url2', 'url3', 'url4', 'url5']
- min_wait = 1000
- max_wait = 15000
- </font>
复制代码
场景二模拟3用户并发注册账号,共有9个账号,要求注册账号不重复,注册完毕后结束测试 概括:
保证并发测试数据唯一性,不循环取数据
>>>所有并发虚拟用户共享同一份测试数据,并且保证虚拟用户使用的数据不重复; 代码
采用队列 - <font size="3" color="#000000">
- # -*- coding: utf-8 -*-
- """
- @ auth : carl_DJ 来自博主 这里使用了队列
- @ time : 2020-9-23
- """
-
-
- from locust import TaskSet, task, HttpUser
- import queue
- class UserBehav(TaskSet):
- @task
- def test_register(self):
- try:
- data = self.user.user_data_queue.get()
- # data = self.locust.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://www.xxx.com'
- task_set = task(UserBehav)
- 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 = 15000</font>
复制代码 场景三模拟3个用户并发登录账号,总共有9个账号,要求并发登录账号不相同,但数据可循环使用; 概括:
保证并发测试数据唯一性,循环取数据;
>>>所有并发虚拟用户共享同一份测试数据,保证并发虚拟用户使用的数据不重复,并且数据可循环重复使用 - <font size="3" color="#000000"># -*- coding: utf-8 -*-
- """
- @ auth : carl_DJ 来自博主 收藏学习
- @ time : 2020-9-23
- """
-
-
- from locust import TaskSet, task, HttpUser
- import queue
- class UserBehav(TaskSet):
- @task
- def test_register(self):
- try:
- data = self.user.user_data_queue.get()
- # data = self.locust.user_data_queue.get()
- except queue.Empty:
- print('account data run out, test ended')
- exit(0)
- print('register with user: {0}, pwd: {1}' .format(data['username'], data['password']))
- payload = {
- 'username': data['username'],
- 'password': data['password']
- }
- self.client.post('/register', data=payload)
- self.user.user_data_queue.put_nowait(data)
- # self.locust.user_data_queue.put_nowait(data)
- class WebsiteUser(HttpUser):
- host = 'http://www.xxx.com'
- task_set = task(UserBehav)
- 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 = 15000</font>
复制代码 转自:http://t.csdn.cn/IWo2a
|
|