如何在 S3 中模拟 ZipFile 对象？答案

【问题标题】：How to mock a ZipFile object in S3?如何在 S3 中模拟 ZipFile 对象？
【发布时间】：2022-01-11 12:46:57
【问题描述】：

我想测试将档案上传到 S3 存储桶。该存档已解压缩，因此我也想测试提取的 JSON 文件。这是我要测试的内容：

def get_files_from_s3(bucket_name, s3_prefix):
    files = []
    s3_resource = boto3.resource("s3")
    bucket = s3_resource.Bucket(bucket_name)
    response = bucket.objects.filter(Prefix=s3_prefix)
    for obj in response:
        if obj.key.endswidth('.zip'):
            buffer = BytesIO(obj.get()["Body"].read())
            print("Before unzipping the file")
            z = ZipFile(buffer)
            for filename in z.namelist():
                print("After unzipping the file")
                if filename == "files/file1.json":
                    for dict in self.parse_json(z, filename):
                    # store some data from the json
            data.append(...)
    return data

def parse_json(obj, file):
    with obj.open(file, "r") as f:
        data = json.loads(f.read())
    return data["some_key"]

这是我在收到my previous question here 的答复后在单元测试中尝试过的：

from moto import mock_s3

@pytest.fixture(scope='function')
def aws_credentials():
    """Mocked AWS Credentials, to ensure we're not touching AWS directly"""
    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'


@mock_s3
def test_get_files_from_s3(self, aws_credentials):
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(self.bucket_name)
    # Create the bucket first, as we're interacting with an empty mocked 'AWS account'
    bucket.create()

    # Create some example files that are representative of what the S3 bucket would look like in production
    client = boto3.client('s3')
    client.put_object(Bucket=self.bucket_name, Key=s3_prefix + "file.zip", Body=open('local_file.zip', 'r'))
    client.put_object(Bucket=self.bucket_name, Key=s3_prefix + "file.nonzip", Body="...")

    # Retrieve the files again using whatever logic
    files = module.get_files_from_s3(BUCKET_NAME, S3_PREFIX)
    self.assertEqual(['file.zip'], files)

所以我关心的是如何在传递给 moto 时模拟 ZipFile 对象？因为当我将 Body 参数与要发送到 S3 客户端的实际 zip 文件一起传递时，我发现了这个错误：E UnicodeDecodeError: 'utf-8' codec can't decode byte 0x9b in position 10: invalid start byte

另一个问题是如何将该对象传递给另一个函数，如parse_json()？

谢谢

【问题讨论】：

标签： python unit-testing amazon-s3 mocking boto3

【解决方案1】：

Body-参数只接受字节，这意味着open-方法需要使用rb参数返回文件的字节内容：

client.put_object(Bucket=self.bucket_name, Key=s3_prefix + "file.zip", Body=open('local_file.zip', 'rb'))

请注意，这不是 Zipfile/mocking/moto 问题，而是 boto3 限制。
如果第二个参数只是r，则内容将作为字符串返回。尝试上传时，AWS 将失败并显示以下错误消息：TypeError: Unicode-objects must be encoded before hashing

【讨论】：

非常感谢！