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

Question

您认为使用 PyTest 执行负载测试是否可行？\n例如：

\n\n

import locust\n\nclass UsersTest(locust.TaskSet):\n\n    @locust.seq_task(1)\n    def api_get_task(self):\n        self.client.get("/api", name="GET /api") # \xd0\xa1\xd0\xb0\xd0\xbc\xd0\xbe\xd0\xb5 \xd0\xb4\xd0\xb5\xd0\xb9\xd1\x81\xd1\x82\xd0\xb2\xd0\xb8\xd0\xb5\n\n    @locust.seq_task(2)\n    def api_post_task(self):\n        payload = {"username": "user1", "password": "123456"}\n        self.client.post("/api", data=payload, name="POST /api")\n\nclass SituationTest(locust.HttpLocust):\n\n    task_set = UsersTest \n    min_wait = 1000 \n    max_wait = 2000\n    host = "http://127.0.0.1:3000"\n

\n\n

这是针对 2 个 url 的 2 个简单任务的示例。在 UsersTest 类中，我有我的测试用例本身。进入 SituationTest 类，我有我的参数。

\n\n

所以问题是如何将这两个类集成到 pytest 装置装饰器中并将其拆分到 test_file.py 和 conftest.py 之间？

\n

Answer 1

另外，您可以使用locust其用作库而不是 CLI。

\n

import gevent\nimport locust\n\n\nclass UsersTest(locust.SequentialTaskSet):\n\n    @locust.task\n    def api_get_task(self):\n        self.client.get("/api", name="GET /api")  # \xd0\xa1\xd0\xb0\xd0\xbc\xd0\xbe\xd0\xb5 \xd0\xb4\xd0\xb5\xd0\xb9\xd1\x81\xd1\x82\xd0\xb2\xd0\xb8\xd0\xb5\n\n    @locust.task\n    def api_post_task(self):\n        payload = {"username": "user1", "password": "123456"}\n        self.client.post("/api", data=payload, name="POST /api")\n\n\nclass SituationTest(locust.HttpUser):\n\n    task_set = UsersTest\n    min_wait = 1000\n    max_wait = 2000\n    host = "http://127.0.0.1:3000"\n\n\ndef test__your_pytest_example():\n    env = locust.env.Environment(user_classes=[SituationTest])\n    env.create_local_runner()\n    gevent.spawn(locust.stats.stats_history, env.runner)\n    env.runner.start(1, spawn_rate=1)\n    gevent.spawn_later(10, lambda: env.runner.quit())\n    env.runner.greenlet.join()\n\n    assert env.stats.total.avg_response_time < 60\n    assert env.stats.total.num_failures == 0\n    assert env.stats.total.get_response_time_percentile(0.95) < 100\n

\n

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

\n

Answer 2

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

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

这里是如何以编程方式添加任务。

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

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 功能，例如装置、参数化等，您可以使用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)

这段代码将使单独的测试成为蝗虫任务。