如何使用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"

这里有两个URL的两个简单任务例子。在UsersTest类中包含测试用例，在SituationTest类中包含参数。

问题是如何将这两个类整合到pytest fixtures装饰器中，并在test_file.py和conftest.py之间拆分？

- Anton Lisetsky

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

- Darío López Padial

Answer 2

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

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

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

生成的代码应该类似于这个效果。

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

这段代码将会把单独的测试转换为一个Locust任务。

- SilentGuy

---

此代码可能已过时，因为Locust进行了更改。当Locust升级到1.0版本时，它进行了同样的重大更改。 - SilentGuy