如何使用 PyTest 对 Locust 执行负载测试？

Question

您认为使用 PyTest 执行负载测试有什么可能？ 例如：

import locust

class UsersTest(locust.TaskSet):

    @locust.seq_task(1)
    def api_get_task(self):
        self.client.get("/api", name="GET /api") # Самое действие

    @locust.seq_task(2)
    def api_post_task(self):
        payload = {"username": "user1", "password": "123456"}
        self.client.post("/api", data=payload, name="POST /api")

class SituationTest(locust.HttpLocust):

    task_set = UsersTest 
    min_wait = 1000 
    max_wait = 2000
    host = "http://127.0.0.1:3000"

这里是 2 个 url 的 2 个简单任务的示例。进入类 UsersTest 我有我的测试用例本身。进入课堂 SituationTest 我有我的参数。

那么问题是如何将这 2 个类集成到 pytest 固定装置装饰器中并将其拆分为 test_file.py 和 conftest.py？

Answer 1

此外，您可以将 locust 用作库而不是 CLI。

import gevent
import locust


class UsersTest(locust.SequentialTaskSet):

    @locust.task
    def api_get_task(self):
        self.client.get("/api", name="GET /api")  # Самое действие

    @locust.task
    def api_post_task(self):
        payload = {"username": "user1", "password": "123456"}
        self.client.post("/api", data=payload, name="POST /api")


class SituationTest(locust.HttpUser):

    task_set = UsersTest
    min_wait = 1000
    max_wait = 2000
    host = "http://127.0.0.1:3000"


def test__your_pytest_example():
    env = locust.env.Environment(user_classes=[SituationTest])
    env.create_local_runner()
    gevent.spawn(locust.stats.stats_history, env.runner)
    env.runner.start(1, spawn_rate=1)
    gevent.spawn_later(10, lambda: env.runner.quit())
    env.runner.greenlet.join()

    assert env.stats.total.avg_response_time < 60
    assert env.stats.total.num_failures == 0
    assert env.stats.total.get_response_time_percentile(0.95) < 100

查看官方文档：https://docs.locust.io/en/stable/use-as-lib.html

Answer 2

一种可能的方法是将单独的 pytest 测试添加为 locust 任务。

如果您的测试很简单并且不使用 pytest 固定装置，您可以导入所有测试函数（或类）并将它们添加为任务。

Here 是您以编程方式添加任务的方式。

生成的代码将具有这种效果。

from test_module import test_a
from locust import TaskSet, HttpLocust


class TestTaskSet(TaskSet):
     @task
     def test_task(self):
         self.schedule_task(test_a)


class WebsiteUser(HttpLocust):
    task_set = TestTaskSet

如果您的测试代码使用了 pytest 功能，如fixture、参数化等，您可以使用pytest.main() 收集所有测试并将单个测试添加为任务并将它们作为 pytest 测试执行。

例如

import pytest
from locust import TaskSet, HttpLocust, between


class TestCollector:

    def __init__(self):
        self.collected = []

    def pytest_collection_modifyitems(self, items):
        for item in items:
            self.collected.append(item.nodeid)


test_collector = TestCollector()

pytest.main(['tests_dir', '--collect-only'], plugins=[test_collector])


class TestTaskSet(TaskSet):

    @task
    def task_gen(self):
        for test in test_collector.collected:
            self.schedule_task(pytest.main, args=[test])


class WebsiteUser(HttpLocust):
    task_set = TestTaskSet
    wait_time = between(1, 5)

此代码将使单个测试成为蝗虫任务。