如何在超时时使 Locust 中的轮询步骤失败并测量总时间

Question

我有一个执行步骤流的 Locust 测试：

@task
def execute_tasks():
   test = TestFlow(self.client)
   test.step1()
   test.step2()
   ...

其中一个步骤涉及轮询状态：

def test4(self, timeout_seconds=420, interval_seconds=5):
    start = time.time()
    is_done = False
    while not is_done and time.time() - start < timeout_seconds:
        time.sleep(interval_seconds)
        response = self.client.get("url", name="wait_for_test")
        test_status = response.json()
        is_done = test_status["status"] == "COMPLETED"

    if not is_done:
        raise Exception("Polling timed out!")

问题：

1. 如果超时，我如何在 Locust 中将此步骤标记为失败（而不是 只是提出例外）？

2. 如何测量总轮询持续时间（从开始 轮询直到收到“COMPLETED”状态）而不仅仅是 跟踪单个 GET 请求？

Answer 1

您可以调用响应的 failure() 方法：

response.failure("Polling timed out!")

这需要向

get

函数传递参数

catch_response

，该参数必须设置为

True

:

response = self.client.get("url", name="wait_for_test", catch_response=True)

可以在这里找到文档：

https://docs.locust.io/en/stable/writing-a-locustfile.html#manually-controlling-if-a-request-should-be-considered-successful-or-a-failure