You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Locust.io中配置AWS认证凭证以进行API负载测试?

在Locust.io中提供AWS认证凭证的方案

针对你的问题,以下几种方法可以在Locust测试中注入AWS Access Key、Secret Key和Session Token,解决API调用因缺少凭证失败的问题:

方法1:通过环境变量传递凭证

AWS SDK默认会读取环境变量,你可以在启动Locust前设置这些变量,也可以在代码中显式加载:

# 启动Locust前设置环境变量
export AWS_ACCESS_KEY_ID="your-access-key"
export AWS_SECRET_ACCESS_KEY="your-secret-key"
export AWS_SESSION_TOKEN="your-session-token"

在Locust代码中,若API需要直接携带凭证头,可按如下方式修改:

import os
import time
from locust import HttpUser, task, between

class QuickstartUser(HttpUser):
    wait_time = between(1, 5)
    
    def on_start(self):
        # 从环境变量加载AWS凭证
        self.aws_access_key = os.getenv("AWS_ACCESS_KEY_ID")
        self.aws_secret_key = os.getenv("AWS_SECRET_ACCESS_KEY")
        self.aws_session_token = os.getenv("AWS_SESSION_TOKEN")

    @task(3)
    def view_items(self):
        # 根据API要求的格式构造认证头
        headers = {
            "X-AWS-ACCESS-KEY": self.aws_access_key,
            "X-AWS-SECRET-KEY": self.aws_secret_key,
            "X-AWS-SESSION-TOKEN": self.aws_session_token
        }
        for item_id in range(10):
            self.client.get(f"/item?id={item_id}", name="/item", headers=headers)
            time.sleep(1)

方法2:使用AWS签名库生成认证头

如果你的API采用AWS Signature Version 4认证,推荐用requests-aws4auth库自动生成签名头:
先安装依赖:

pip install requests-aws4auth

再修改Locust代码:

import time
from locust import HttpUser, task, between
from requests_aws4auth import AWS4Auth
import os

class QuickstartUser(HttpUser):
    wait_time = between(1, 5)
    
    def on_start(self):
        self.auth = AWS4Auth(
            os.getenv("AWS_ACCESS_KEY_ID"),
            os.getenv("AWS_SECRET_ACCESS_KEY"),
            # 替换为你的AWS区域,例如us-east-1
            "us-east-1",
            # 替换为对应服务,例如API Gateway填execute-api
            "execute-api",
            session_token=os.getenv("AWS_SESSION_TOKEN")
        )

    @task(3)
    def view_items(self):
        for item_id in range(10):
            # 用auth参数传递AWS认证
            self.client.get(f"/item?id={item_id}", name="/item", auth=self.auth)
            time.sleep(1)

方法3:直接在代码中配置(不推荐)

虽然可以硬编码凭证,但强烈不建议这么做,会带来安全风险,仅适合测试环境临时验证:

import time
from locust import HttpUser, task, between
from requests_aws4auth import AWS4Auth

class QuickstartUser(HttpUser):
    wait_time = between(1, 5)
    
    def on_start(self):
        self.auth = AWS4Auth(
            "your-access-key",
            "your-secret-key",
            "us-east-1",
            "execute-api",
            session_token="your-session-token"
        )

    @task(3)
    def view_items(self):
        for item_id in range(10):
            self.client.get(f"/item?id={item_id}", name="/item", auth=self.auth)
            time.sleep(1)

内容的提问来源于stack exchange,提问作者Rock

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.17 01:55:21