1.参数化

方式1：循环取数据，数据可以重复使用。

所有并发虚拟用户共享同一份测试数据，各虚拟用户在数据列表中循环取值。
　例如，模拟10个用户并发请求网页，总共有4个URL地址，每个虚拟用户都会依次循环加载这4个URL地址。

class HttpTask(TaskSet):

    index = 0

    @task

    def test001(self):

        url = self.user.data[self.index]

        self.index = (self.index + 1) % len(self.user.data)

        print(url)

        self.client.get(url)

class MyUser(HttpUser):

    host = "https://www.baidu.com"

    tasks = [HttpTask]

    between(0, 3)

    data = ['https://www.baidu.com/',

            'https://www.baidu.com/s?wd=1',

            'https://www.baidu.com/s?wd=2',

            'https://www.baidu.com/s?wd=3',

            'https://www.baidu.com/s?wd=4']

方式2：保证并发测试数据唯一性，不循环取数据，使用队列的方法。

所有并发虚拟用户共享同一份测试数据，并且保证虚拟用户使用的数据不重复。

列如，模拟10用户并发注册账号，总共有100个账号，邀请注册账号不重复。注册完毕后自动结束测试。

class HttpTask(TaskSet):

    @task

    def test001(self):

        try:

            url = self.user.queue.get()

        except queue.Empty:

            exit(0)

        self.client.get(url)

class MyUser(HttpUser):

    host = "https://www.baidu.com"

    tasks = [HttpTask]

    between(0, 3)

    queue = queue.Queue()

    data = 'https://www.baidu.com/'

    for i in range(5):

        queue.put_nowait(data)

方式3：保证并发数据的唯一性，循环取数据，使用队列方法，，欸出使用完数据后在放入到队列中。

所有并发虚拟用户共享同一份测试数据，保证并发虚拟用户使用的数据不重复，并且数据可循环重复使用。

列如：模拟10用户并发登录账号。总共有100个账号，要求并发登录账号不相同，但数据可循环使用。

class HttpTask(TaskSet):

    @task

    def test001(self):

        try:

            url = self.user.queue.get()

        except queue.Empty:

            exit(0)

        self.client.get(url)

        self.user.queue.put_nowait(url)

class MyUser(HttpUser):

    host = "https://www.baidu.com"

    tasks = [HttpTask]

    between(0, 3)

    queue = queue.Queue()

    data1 = ["https://www.baidu.com/",

             "https://www.baidu.com/s?wd=1",

             "https://www.baidu.com/s?wd=2",

             "https://www.baidu.com/s?wd=3",

             "https://www.baidu.com/s?wd=4"]

    for i in range(0,5):

        data = data1[i]

        queue.put_nowait(data)

2集合

from locust import SequentialTaskSet, task, HttpUser,events

from gevent._semaphore import Semaphore

all_locusts_spawned = Semaphore()

all_locusts_spawned.acquire()

def on_hatch_complete(**kwargs):

    all_locusts_spawned.release() # 创建钩子方法

# events.on_hatch_complete += on_hatch_complete

# 挂在到locust钩子函数（所有的Locust示例产生完成时触发）

events.spawning_complete.add_listener(on_hatch_complete)

class UserBehavior(SequentialTaskSet):

    index = 0

    # 限制在所有用户准备完成前处于等待状态

    def on_start(self):

        all_locusts_spawned.wait()

    @task

    def test_visit(self):

        all_locusts_spawned.wait() # 限制在所有用户准备完成前处于等待状态

        url = self.user.share_data[self.index]

        self.index = (self.index + 1) % len(self.user.share_data)

        with self.client.get(url,catch_response=True) as response:

            if response.status_code == 200:

                response.success()

            else:

                response.failure("查询失败")

        print('visit url: %s' % url)

class WebsiteUser(HttpUser):

    host = "https://www.baidu.com"

    tasks = [UserBehavior]

    share_data = ['https://www.baidu.com/',

                  'https://www.baidu.com/s?wd=1',

                  'https://www.baidu.com/s?wd=2',

                  'https://www.baidu.com/s?wd=3',

                  'https://www.baidu.com/s?wd=4']

    min_wait = 100

    max_wait = 300

if __name__ == '__main__':

    import os

    os.system("locust -f parameterized.py")

巴特西

locust 检查点、集合点、参数化设置。

1.参数化

2集合

最新文章

热门文章