无法停止测试，@events.test_stop.add_listener locust答案

【问题标题】：Unable to stop the test, @events.test_stop.add_listener locust无法停止测试，@events.test_stop.add_listener locust
【发布时间】：2021-03-01 08:38:02
【问题描述】：

如果失败百分比或平均响应时间超过 test_stop 中定义的值，我无法停止测试。下面是示例代码：-

@events.test_stop.add_listener
def results(environment, **kw):
    if environment.stats.total.fail_ratio > 0.01:
        logging.error("Test failed due to failure ratio > 1%")
        environment.process_exit_code = 1
    elif environment.stats.total.avg_response_time > 2:
        logging.error("Test failed due to average response time ratio > 200 ms")
        environment.process_exit_code = 1
    elif environment.stats.total.get_response_time_percentile(0.95) > 8:
        logging.error("Test failed due to 95th percentil response time > 800 ms")
        environment.process_exit_code = 1
    else:
        environment.process_exit_code = 0
     


    @task
    def osp_pick(self):
        return super().request(
            name="pick",
            json_file="data/pick.json",
            ssm=self.ssm,
            my_task=self,
           
        )


class StockTransfer(HttpUser):
    host = "https://zz-zz.dev.zz.zz.zz.com"  # Change this
    tasks = [Handler]

我正在使用以下命令从终端运行测试：-

locust -f --headless -u 1000 -r 100 --run-time 2m

命令按预期工作，测试运行 2 分钟，但如果错误百分比 > 0.01 或平均响应时间 > 2 秒，则测试不会失败

【问题讨论】：

您是否验证了更改退出代码的函数正在运行？我将从那里开始，也许将您的 if 检查替换为只是更改退出代码。然后从那里打印出你关心的支票和/或统计数据，确保你的单位和一切都匹配。

标签： python load-testing locust

【解决方案1】：

查看 locust-plugins 的 --check-fail-ratio 以了解实现这一目标的实战方法：

https://github.com/SvenskaSpel/locust-plugins#command-line-options

或者，如果您想在实际测试期间运行检查并在失败时中断它，您可以使用后台greenlet：

def checker(environment):
    while not environment.runner.state in [STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP]:
        time.sleep(1)
        if environment.runner.stats.total.fail_ratio > 0.2:
            logging.info(f"fail ratio was {environment.runner.stats.total.fail_ratio}, quitting")
            environment.process_exit_code = 2
            environment.runner.quit()
            return


@events.init.add_listener
def on_locust_init(environment, **_kwargs):
    if not isinstance(environment.runner, WorkerRunner):
        gevent.spawn(checker, environment)

【讨论】：