我有一個定義如下的 S3 函式：
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"]
return sorted(all_files, key=lambda k: k[1])
現在，我希望為此撰寫一個單元測驗，并嘗試使用 botocore 中的 stubber 物件來測驗它。我試過類似的東西：
def test_unprocessed_s3_objects():
client = boto3.client("s3")
stubber = Stubber(client)
list_objects_v2_response = {
"Contents": [
{
"Key": "sensor_1",
"LastModified": "2021-11-30T12:58:14 00:00",
"ETag": '"3a3c5ca43d2f01dba42314c1ca7e2237"',
"Size": 1417,
"StorageClass": "STANDARD",
},
{
"Key": "sensor_2",
"LastModified": "2021-11-30T12:58:05 00:00",
"ETag": '"02bc99343bbdcbdefc9fe691b6f7deaa"',
"Size": 1332,
"StorageClass": "STANDARD",
},
]
}
stubber.add_response("list_objects_v2", list_objects_v2_response)
items = get_unprocessed_s3_object_list(client, "test")
assert len(items) == 2
但是，這又回來了Access Denied with ListObjectsV2 ooperation。也許這與存盤桶 ID 有關。我認為這會繞過實際呼叫。
uj5u.com熱心網友回復：
呼叫后stubber.add_response("list_objects_v2", list_objects_v2_response)，您需要呼叫stubber.activate()才能使用存根。
轉載請註明出處，本文鏈接：https://www.uj5u.com/gongcheng/374053.html