基于Locust库从代码动态分配Testplan并输出至Timescale

• 通过事件钩子注入自定义Testplan
  Locust的请求成功/失败事件会传递统计上下文，你可以在测试初始化时定义全局Testplan标识，再通过事件监听器修改统计数据的testplan字段：

  from locust import events, HttpUser, task
  import sys
  
  # 从启动参数或环境变量获取Testplan名称
  TEST_PLAN = sys.argv[1] if len(sys.argv) > 1 else "default_test"
  
  @events.request_success.add_listener
  def inject_testplan_on_success(request_type, name, response_time, response_length, **kwargs):
      kwargs["context"]["testplan"] = TEST_PLAN
  
  @events.request_failure.add_listener
  def inject_testplan_on_failure(request_type, name, response_time, exception, **kwargs):
      kwargs["context"]["testplan"] = TEST_PLAN
  
  class TestUser(HttpUser):
      @task
      def sample_request(self):
          self.client.get("/api/test")
  

  启动测试时直接传入Testplan参数，比如python test_script.py checkout_perf_test，就能让本次所有请求的Testplan字段统一为指定值。

• 直接修改Locust统计对象属性
  如果事件钩子方式不生效，可以直接修改Locust的核心统计实例，强制设置testplan：

  from locust import stats
  
  def set_testplan(testplan_name):
      # 给所有请求统计条目设置Testplan
      for entry in stats.stats.entries.values():
          entry.testplan = testplan_name
      # 同时设置全局汇总统计的Testplan
      stats.stats.total.testplan = testplan_name
  
  # 在测试开始前调用该函数
  set_testplan("login_flow_test")
  

  这种方式直接修改统计核心对象，能确保所有请求的统计数据都带上自定义的Testplan标识。

• 在TimescaleDB写入阶段添加标识
  若你通过自定义脚本将Locust数据写入TimescaleDB，可在写入前直接给数据追加Testplan字段，无需修改Locust内部逻辑：

  def write_to_timescale(request_record):
      # 注入当前测试的Testplan标识
      request_record["testplan"] = "payment_api_test"
      # 执行数据库插入操作
      # 示例SQL：INSERT INTO requests (run_id, testplan, ...) VALUES (...)
  

内容的提问来源于stack exchange，提问作者aemreg