s3 模拟 lambda_handler 测试返回 botocore.exceptions.ClientError

Question

我正在尝试为我的 lambda 处理程序编写一个测试，它使用 boto3 从存储桶下载文件并将其存储在本地：

s3_resource = boto3.resource('s3')
TEMP_FILE = '/tmp/file.csv'

def lambda_handler(event, context):
    bucket_name = event['detail']['bucket']['name']
    file_name = event['detail']['object']['key']
    s3_resource.Bucket(bucket_name).download_file(Key=file_name, Filename=TEMP_FILE)

因为我不想在我的测试中实际与 s3 交互，所以我创建了一个虚拟 s3_upload_event 来传递给函数调用。我还使用 moto 创建了一个模拟的 s3 存储桶，并在其中放入了一些虚拟 test_data，以及一个具有 s3 权限的模拟 iam 用户：

TEST_BUCKET = "test_bucket_name"
S3_TEST_FILE_KEY = 'path/to/test.csv'

@pytest.fixture
def s3_upload_event():
    return {"detail":{"bucket":{"name": TEST_BUCKET}, "object": {"key": S3_TEST_FILE_KEY}}}

@pytest.fixture
def context():
    return object()  

@pytest.fixture
def test_data():
    return b'col_1,col_2\n1,2\n3,4\n'

@pytest.fixture
@mock_iam
def mock_user(user_name="test-user"):
    # create user
    client = boto3.client("iam", region_name="us-west-2")
    client.create_user(UserName=user_name)

    # create and attach policy
    policy_document = {
        "Version": "2012-10-17",
        "Statement": [{
            "Effect": "Allow",
            "Action": ["s3:*", "s3-object-lambda:*"],
            "Resource": "*"
        }]
    }
    policy_arn = client.create_policy(
        PolicyName="test",
        PolicyDocument=json.dumps(policy_document))["policy"]["Arn"]
    client.attach_user_policy(UserName=user_name, PolicyArn=policy_arn)

    # Return access keys
    yield client.create_access_key(UserName=user_name)["AccessKey"]

@pytest.fixture
@mock_s3
def mock_s3(test_data, mock_user):
    s3 = boto3.client(
        "s3",
        region_name="us-west-2",
        aws_access_key_id=mock_user["AccessKeyId"],
        aws_secret_access_key=mock_user["SecretAccessKey"])
    s3.create_bucket(Bucket=TEST_BUCKET)
    s3.put_object(Bucket=TEST_BUCKET, Key=S3_TEST_FILE_KEY, Body=test_data)
    yield s3

我将这些模拟的固定装置注入到我的测试中，如下所示：

class TestLambdaHandler:        
    def test_lambda_handler(self, mock_user, mock_s3, s3_upload_event, context):

        response = lambda_handler(event = s3_upload_event, context = context)

        assert response["statusCode"] == 200  

但是当我运行测试时，botocore 在到达这行代码时会抛出异常：s3_resource.Bucket(bucket_name).download_file(Key=file_name, Filename=TEMP_FILE):

botocore.exceptions.ClientError: An error occurred (403) when calling the HeadObject operation: Forbidden

当我用谷歌搜索这个错误时，这似乎与缺少 IAM 权限有关。我为模拟用户使用的 PolicyDocument 与我在代码中使用的实际策略相同，所以我不明白为什么它能够在现实生活中下载文件但在测试中失败。我的模拟用户有什么遗漏吗？

Answer 1

在重新阅读 Moto (http://docs.getmoto.org/en/latest/docs/getting_started.html) 的设置说明后，我设法让 mock 工作，上面的代码存在一些问题。

1. 没有必要模拟 iam 凭证和用户，但我确实需要模拟 AWS 凭证并将它们传递给 s3 模拟。
2. 我将装饰器 (@s3_mock) 更改为 s3 模拟上的上下文管理器模式。我的理解是，您不应该将装饰器用于应用模拟的自定义夹具。
3. 将我的模拟从 mock_s3 重命名为 s3，以避免与 moto 出售的夹具混淆。
4. 更新了 s3 夹具以仅创建模拟客户端

更新的赛程：

@pytest.fixture(scope='function')
def aws_credentials():
    """Mocked AWS Credentials for moto."""
    os.environ['AWS_ACCESS_KEY_ID'] = 'testing'
    os.environ['AWS_SECRET_ACCESS_KEY'] = 'testing'
    os.environ['AWS_SECURITY_TOKEN'] = 'testing'
    os.environ['AWS_SESSION_TOKEN'] = 'testing'

@pytest.fixture(scope='function')
def s3(aws_credentials):
    with mock_s3():
        yield boto3.client('s3', region_name="us-west-2")

然后我更新了测试以采用 s3 夹具并创建测试“桶”并将测试“对象”作为测试设置的一部分。

VALID_TEST_FILE_KEY = "path/to/test.csv"
VALID_DATA = "tests/data/valid_data.csv"
TEST_BUCKET = "test_bucket_name"

    def test_lambda_handler(self, s3_upload_valid_file, context, s3):
        #arrange
        self._create_bucket_and_add_file(s3, VALID_DATA, VALID_TEST_FILE_KEY)

        #act
        response = lambda_handler(event = s3_upload_valid_file, context = context)

        assert response["statusCode"] == 200 

    ### Helper Method for Test ###
    def _create_bucket_and_add_file(self, s3, data_file, key):
        with open(data_file, 'r') as f:
            test_data = f.read()
        s3.create_bucket(Bucket=TEST_BUCKET, CreateBucketConfiguration={'LocationConstraint': "us-west-2"})
        s3.put_object(Bucket=TEST_BUCKET, Key=key, Body=test_data)