为S3函数编写单元测试时botocore Stubber返回ListObjectsV2访问被拒如何解决
问题原因
该错误和桶ID无关。你创建Stubber实例后没有激活,S3客户端仍然会发起真实请求到AWS公有云服务,触发实际的权限校验,因此返回Access Denied报错。
修复步骤
- 激活Stubber拦截请求:Stubber需要通过上下文管理器或者手动调用
activate()/deactivate()方法才能生效,拦截客户端的请求返回模拟响应,不会真的发送请求到AWS。 - 补全响应的必填字段:你的模拟响应缺少
KeyCount字段,原函数逻辑中会先判断KeyCount > 0才会读取Contents内容,缺少该字段会导致即使Stub生效,函数也会返回空列表。 - 修正原函数的排序逻辑错误:你当前的排序规则
key=lambda k: k[1]是取Key字符串的第二个字符排序,不符合按LastModified排序的预期,需要调整为取字典中对应的LastModified值。
修复后代码示例
修正后的原函数
from typing import List import boto3 from datetime import datetime def get_unprocessed_s3_object_list(s3_client, bucket: str) -> List[str]: paginator = s3_client.get_paginator("list_objects_v2") page_iterator = paginator.paginate(Bucket=bucket) all_files = {} for page in page_iterator: if page["KeyCount"] > 0: for item in page["Contents"]: all_files[item["Key"]] = item["LastModified"] # 修正排序逻辑,按LastModified排序 return sorted(all_files.keys(), key=lambda k: all_files[k])
修正后的测试函数
from botocore.stub import Stubber def test_unprocessed_s3_objects(): client = boto3.client("s3", region_name="us-east-1") stubber = Stubber(client) list_objects_v2_response = { "KeyCount": 2, # 补全必填字段 "Contents": [ { "Key": "sensor_1", "LastModified": datetime(2021, 11, 30, 12, 58, 14), "ETag": '"3a3c5ca43d2f01dba42314c1ca7e2237"', "Size": 1417, "StorageClass": "STANDARD", }, { "Key": "sensor_2", "LastModified": datetime(2021, 11, 30, 12, 58, 5), "ETag": '"02bc99343bbdcbdefc9fe691b6f7deaa"', "Size": 1332, "StorageClass": "STANDARD", }, ] } # 用上下文管理器自动激活/停用Stubber with stubber: stubber.add_response("list_objects_v2", list_objects_v2_response, expected_params={"Bucket": "test"}) items = get_unprocessed_s3_object_list(client, "test") assert len(items) == 2 # 可选校验排序结果,sensor_2的修改时间更早,应该排在第一位 assert items == ["sensor_2", "sensor_1"]
内容的提问来源于stack exchange,提问作者Luca
相关产品推荐
相关产品推荐

