Locust脚本报错：'Events'对象无'request_success'属性排查

Locust 2.x版本（包括你使用的2.15.1）中，request_success和request_failure这两个请求事件不再挂载在全局events对象上，而是属于environment.events的属性。你脚本中直接调用ev.request_success.fire()，会因为全局events没有该属性而抛出AttributeError。

另外，你用FastHttpUser作为测试类的父类也不合适——FastHttpUser是Locust针对HTTP请求优化的用户类，你的场景是gRPC测试，应该使用基础的User类。

import sys
import grpc
import time
import gevent

# Libs
from locust import task, events, constant, User
from locust.runners import STATE_STOPPING, STATE_STOPPED, STATE_CLEANUP, WorkerRunner

from **** import PublicAPIStub
import ****.v1.public_api_pb2 as dtos
import domain.v1.dto_pb2 as domain
import domain.v1.geo_pb2 as geo

def stopwatch(func):
    """统计gRPC请求耗时并上报Locust"""
    def wrapper(*args, **kwargs):
        # 获取测试用户实例，通过实例拿到environment
        self = args[0]
        # 直接用函数名作为任务名，比inspect更简洁可靠
        task_name = func.__name__
        
        start = time.time()
        result = None
        try:
            result = func(*args, **kwargs)
        except Exception as e:
            total = int((time.time() - start) * 1000)
            # 从environment中获取request_failure事件并触发
            self.environment.events.request_failure.fire(
                request_type="grpc",
                name=task_name,
                response_time=total,
                response_length=0,
                exception=e
            )
        else:
            total = int((time.time() - start) * 1000)
            # 从environment中获取request_success事件并触发
            self.environment.events.request_success.fire(
                request_type="grpc",
                name=task_name,
                response_time=total,
                response_length=0
            )
        return result
    return wrapper

class GRPCMyLocust(User):
    wait_time = constant(0)

    def on_start(self):
        """Locust实例启动时执行"""
        pass

    def on_stop(self):
        """Locust实例停止时执行"""
        pass

    @task
    @stopwatch
    def grpc_client_task(self):
        """gRPC测试任务"""
        try:
            ssl_metadata = grpc.ssl_channel_credentials()
            with grpc.secure_channel('laas-zz-lt.usehurrier.com:443', ssl_metadata) as channel:
                input_data = dtos.Request(type='test')
                stub = PublicAPIStub(channel)
                stub.ListAvailableVendors(input_data)
        except (KeyboardInterrupt, SystemExit):
            sys.exit(0)

# 当失败率超过阈值时停止测试
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.4:
            print(f"失败率达到{environment.runner.stats.total.fail_ratio}，停止测试")
            environment.runner.quit()
            return

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

• 移除了全局events的重命名导入（from locust import events as ev），改为通过测试用户实例的self.environment.events访问请求事件。
• 将测试类的父类从FastHttpUser改为基础的User类，符合gRPC测试场景。
• 简化了任务名的获取方式，直接使用被装饰函数的__name__属性，替代复杂的inspect调用。

内容的提问来源于stack exchange，提问作者I.A