如何在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
相关产品推荐
相关产品推荐

