在计时器上调度蝗虫任务

Question

我希望能够有一个以指定间隔运行的任务。例如，假设我想测试3个端点，但每30秒我想ping一个/status/端点

class SomeTasks(TaskSet):
    @task
    self.get("/home/")

    @task
    self.get("/info/")
    
    @task
    self.get("/products/")

    @task
    self.get("/status/")

class SomeUser(FastHttpUser):
    tasks = [SomeTasks]
    wait_time = between(0, 0)

不确定该如何运行，以便像往常一样随机选择/home/、/info/、/products/，但每30秒检查一次状态。也许这可以放在SomeUser类中

Answer 1

听起来您希望/status/的ping与其他用户工作分开。作为测试的一部分，您希望蝗虫生成多个用户来反复点击/home/、/info/、/products/，但您希望单个“用户”每30秒ping一次/status/？如果是这样，这是可能的，但不是很简单。我将在正常的@task之外创建一个单独的函数，它在自己的计时器上重复运行，但仍然向蝗虫报告

有多种方法可以做到这一点，但我倾向于使用Locust events，比如@events.test_start.add_listener，在测试开始时使蝗虫调用函数。在函数中，创建一个greenlet，它每隔30秒在循环中ping /status/端点，并向蝗虫报告。下面的基本文件和关于根据测试状态设置循环条件的位是可选的，取自Locust examples

from locust import HttpUser, TaskSet, task, constant, events
from locust.runners import STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP, WorkerRunner
from locust.clients import HttpSession

import gevent
import time
import os


@events.test_start.add_listener
def on_test_start(environment, **_kwargs):
    gevent.spawn(ping_status, environment)

def ping_status(environment):
    session = HttpSession(base_url=environment.host, request_success=environment.events.request_success, request_failure=environment.events.request_failure)
    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:
        time.sleep(30)
        session.get("/status/")

class UserTasks(TaskSet):

    @task
    def page404(self):
        self.client.get("/does_not_exist")

    @task
    def index(self):
        self.client.get("/")

    @task
    def stats(self):
        self.client.get("/stats/requests")

class WebsiteUser(HttpUser):
    host = "http://127.0.0.1:8089"
    wait_time = constant(2)
    tasks = [UserTasks]