如何Mock测试boto3的describe_subnets函数避免AuthFailure报错
报错原因
- monkeypatch方案错误点:你打桩的是
boto3.client,但原函数使用的是boto3.resource创建资源,打桩目标不匹配,且自定义的MockClientBoto3未实现对应接口逻辑,导致boto3实际走了真实请求链路触发认证报错。 - mock_ec2方案错误点:
@mock_ec2未生效大概率是导入/版本问题,另外即使mock生效,你没有在虚拟aws环境中提前创建对应ID的子网,直接查询不存在的子网也会抛资源不存在错误,不会正常返回。
正确实现方案
方案1:monkeypatch打桩测试
直接mock完整的调用链路,不需要依赖第三方库,完全隔离真实请求:
def test_check_belongsto_az_monkeypatch(monkeypatch): # 构造mock的describe_subnets方法 def mock_describe_subnets(SubnetIds): # 这里可以自定义返回你需要的测试数据 return { "Subnets": [ {"SubnetId": "testid1", "AvailabilityZone": "us-east-1a"}, {"SubnetId": "testid2", "AvailabilityZone": "us-east-1b"} ] } # 构造mock的ec2 client mock_ec2_client = type('', (), {})() mock_ec2_client.describe_subnets = mock_describe_subnets # 构造mock的ec2 resource mock_ec2_resource = type('', (), {})() mock_ec2_resource.meta = type('', (), {})() mock_ec2_resource.meta.client = mock_ec2_client # 打桩boto3.resource,匹配原函数的调用 def mock_resource(service, region_name): assert service == 'ec2' assert region_name == 'us-east-1' return mock_ec2_resource monkeypatch.setattr(boto3, "resource", mock_resource) list_subnet = ['testid1','testid2'] response = fw_init.check_belongsto_describe(list_subnet) assert response == None
该方案完全不会触发真实的AWS请求,不会出现认证错误。
方案2:moto @mock_ec2 测试
首先确保你已经正确安装moto:pip install moto,然后按以下逻辑写测试:
from moto import mock_ec2 import boto3 @mock_ec2 def test_check_belongsto_az_mock(): # 先在mock的AWS环境中创建测试需要的子网资源 ec2 = boto3.resource('ec2', region_name='us-east-1') # 先创建VPC vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16') # 创建两个测试子网,拿到真实的子网ID subnet1 = ec2.create_subnet(VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone='us-east-1a') subnet2 = ec2.create_subnet(VpcId=vpc.id, CidrBlock='10.0.1.0/24', AvailabilityZone='us-east-1b') test_subnet_ids = [subnet1.id, subnet2.id] # 调用被测函数 response = fw_init.check_belongsto_describe(test_subnet_ids) assert response == None
@mock_ec2会自动patch当前环境所有boto3的调用,所有请求都只会在本地mock环境中执行,不会访问真实AWS服务,不会出现认证错误。
异常管控优化
可以在原函数中增加异常捕获逻辑,处理各类AWS接口异常:
import boto3 from botocore.exceptions import ClientError def check_belongsto_describe(list_subnet): try: ec2 = boto3.resource('ec2', region_name='us-east-1') ec2_client = ec2.meta.client response = ec2_client.describe_subnets(SubnetIds=list_subnet) # 这里可以加你的业务逻辑 return None except ClientError as e: error_code = e.response['Error']['Code'] if error_code == 'AuthFailure': # 处理认证失败场景 raise RuntimeError("AWS认证失败,请检查密钥配置") from e elif error_code == 'InvalidSubnetID.NotFound': # 处理子网不存在场景 raise ValueError("传入的子网ID不存在") from e else: # 其他异常统一抛出 raise
内容的提问来源于stack exchange,提问作者Guillermo
相关产品推荐
相关产品推荐

