单元测试中如何stub boto3 S3.Object.wait_until_exists

你遇到的问题由两个容易忽略的boto3 stub + waiter机制细节导致：

• 第一，wait_until_exists waiter完全依赖响应的HTTP状态码匹配内置规则：返回200判定等待成功终止流程，返回404判定需要重试。你配置的head_object响应桩没有显式携带ResponseMetadata.HTTPStatusCode字段，waiter匹配不到任何终止/重试规则，会按照默认配置持续发起head_object重试，直到达到最大重试次数。一旦重试次数超过你提前配置的响应桩数量，就会抛出你看到的UnStubbedResponseError，你观察到的5秒多运行时长刚好和waiter默认重试间隔的累计值吻合。
• 第二，wait_until_exists 内部调用head_object拿到的返回值不会自动写入S3.Object实例的属性缓存，后续访问o.content_length时，boto3会自动触发一次额外的head_object请求做懒加载，这一步如果没配置响应桩同样会报错。

修复后的可运行代码如下，关键改动点已加注释标注：

from datetime import datetime
from io import BytesIO

import boto3
import botocore
import botocore.stub

testing_bucket = "bucket"
testing_key = "key/of/object"
testing_data = b"data"

s3 = boto3.resource("s3")


def put():
    try:
        o = s3.Object(testing_bucket, testing_key)
        o.load()  # 第一次head_object调用
    except botocore.exceptions.ClientError as e:
        if e.response["Error"]["Code"] == "NoSuchKey":
            etag = ""
        else:
            raise e
    else:
        etag = o.e_tag
    try:
        o.upload_fileobj(BytesIO(testing_data))  # put_object调用
    except botocore.exceptions.ClientError as e:
        raise e
    else:
        o.wait_until_exists(IfNoneMatch=etag)  # waiter轮询head_object
        return o.content_length  # 访问属性会触发懒加载，额外调用一次head_object


with botocore.stub.Stubber(s3.meta.client) as s3_stub:
    # 初始o.load()调用的head_object桩
    s3_stub.add_response(
        method="head_object",
        service_response={
            "ETag": "fffffffe",
            "ContentLength": 0,
            # 关键：必须显式指定HTTP状态码
            "ResponseMetadata": {"HTTPStatusCode": 200}
        },
        expected_params={
            "Bucket": testing_bucket,
            "Key": testing_key,
        },
    )
    s3_stub.add_response(
        method="put_object",
        service_response={},
        expected_params={
            "Bucket": testing_bucket,
            "Key": testing_key,
            "Body": botocore.stub.ANY,
        },
    )
    # waiter调用的head_object桩，返回200判定等待成功
    s3_stub.add_response(
        method="head_object",
        service_response={
            "ETag": "ffffffff",
            "AcceptRanges": "bytes",
            "ContentLength": len(testing_data),
            "LastModified": datetime.now(),
            "Metadata": {},
            "VersionId": "null",
            # 关键：必须显式指定HTTP状态码，waiter才能识别为成功并终止轮询
            "ResponseMetadata": {"HTTPStatusCode": 200}
        },
        expected_params={
            "Bucket": testing_bucket,
            "Key": testing_key,
            "IfNoneMatch": "fffffffe",
        },
    )
    # 关键：补o.content_length触发的懒加载head_object桩
    s3_stub.add_response(
        method="head_object",
        service_response={
            "ETag": "ffffffff",
            "ContentLength": len(testing_data),
            "ResponseMetadata": {"HTTPStatusCode": 200}
        },
        expected_params={
            "Bucket": testing_bucket,
            "Key": testing_key,
        },
    )
    print(put())  # 正常输出4

如果不想为最后一次懒加载额外加桩，也可以在业务代码waiter返回后手动给o.content_length赋值，跳过自动加载逻辑，但单元测试场景下直接补桩更符合真实调用逻辑，不需要侵入业务代码。

如果要测试waiter重试逻辑，比如第一次调用head_object返回404触发重试，第二次返回200等待成功，只需要在200的响应桩之前加一个404的错误桩即可，注意错误桩也要显式指定HTTP状态码：

# 第一次waiter调用返回404，触发重试
s3_stub.add_client_error(
    method="head_object",
    service_error_code="NoSuchKey",
    http_status_code=404,
    expected_params={
        "Bucket": testing_bucket,
        "Key": testing_key,
        "IfNoneMatch": "fffffffe",
    }
)
# 第二次调用返回200，等待成功
s3_stub.add_response(
    # 同上文中200的响应配置
)

内容的提问来源于stack exchange，提问作者ljmc