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

如何对连接AWS Secrets Manager和RDS MySQL的Python Lambda做单元测试

本地测试方案指导

方案1:对接AWS测试环境RDS直接测试(新手友好,门槛最低)

  • 本地创建Python虚拟环境,安装所需依赖:pip install boto3 pymysql aws-lambda-context
  • 配置本地AWS凭证:将AWS账号下有权限访问RDS、密钥服务的AK/SK写入本地~/.aws/credentials文件,或临时设置环境变量,保证get_secret()可以正常拉取到RDS连接凭证
  • 调整全局连接逻辑:当前代码的数据库连接定义在全局作用域,仅在Lambda冷启动时初始化,本地多次测试容易出现连接断开问题,建议把连接初始化逻辑移入lambda_handler内部,或新增连接存活校验逻辑
  • 编写本地测试入口脚本,构造和线上结构一致的模拟event:
from 你的Lambda代码文件名 import lambda_handler
# 此处event结构和AWS上实际触发Lambda的event结构保持一致即可
test_event = {
    "request": {
        "userAttributes": {
            "email": "test@email.com"
        }
    },
    "userName": "test@email.com"
}
if __name__ == "__main__":
    resp = lambda_handler(test_event, None)
    print("运行结果:", resp)
  • 直接运行测试脚本即可看到运行结果,异常信息会直接抛出方便排查。

方案2:全本地模拟(无需连接AWS服务,适配moto测试场景)

  • 本地启动MySQL服务,创建和线上结构一致的数据库,提前建好group1、group2_account、onboarding_status三张表并插入测试数据
  • 安装模拟所需依赖:pip install moto[all] pymysql
  • 测试代码中用moto模拟Secrets Manager服务,让get_secret()返回本地MySQL的连接参数;你现有代码中初始化的rds_client没有实际使用,可以直接删除避免多余的mock操作
  • 示例测试代码参考:
import moto
import boto3
import json
from 你的Lambda代码文件名 import lambda_handler

@moto.mock_secretsmanager
def test_lambda_logic():
    # 模拟密钥服务返回本地MySQL配置
    sm_client = boto3.client('secretsmanager', region_name='us-east-1')
    sm_client.create_secret(
        Name='你线上RDS密钥的名称',
        SecretString=json.dumps({
            "host": "127.0.0.1",
            "username": "本地MySQL用户名",
            "password": "本地MySQL密码",
            "dbname": "本地测试库名",
            "port": 3306
        })
    )
    # 构造测试event
    test_event = {
        "request": {
            "userAttributes": {
                "email": "test@normal.com"
            }
        },
        "userName": "test@normal.com"
    }
    resp = lambda_handler(test_event, None)
    print("测试通过,返回结果:", resp)

if __name__ == "__main__":
    test_lambda_logic()

现有代码优化建议

  • 数据库查询时直接对fetchone()的返回结果取[0]会在查询无结果时抛出索引错误,建议先判断返回值是否为空再取值,示例逻辑:
res = cursor.fetchone()
return res[0] if res else None
  • 全局的数据库连接在Lambda多次触发时可能会失效,建议每次执行查询前校验连接是否存活,或引入连接池管理连接生命周期。

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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.10.07 11:27:04