如何在moto模拟的S3环境中mock ZipFile对象并解决编码报错

问题背景 #

需要测试S3存储桶压缩包上传、解压后JSON文件处理的完整流程，原业务代码如下：

import boto3
import json
from io import BytesIO
from zipfile import ZipFile

def get_files_from_s3(bucket_name, s3_prefix):
    data = []
    s3_resource = boto3.resource("s3")
    bucket = s3_resource.Bucket(bucket_name)
    response = bucket.objects.filter(Prefix=s3_prefix)
    for obj in response:
        # 原代码笔误：endswidth应为endswith
        if obj.key.endswith('.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 item in parse_json(z, filename):
                        # 存储json中的对应数据
                        data.append(item)
    return data

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

基于moto的mock_s3编写的单元测试运行时遇到两个问题：

1. 向mock S3传递本地zip文件作为Body参数时，抛出UnicodeDecodeError: 'utf-8' codec can't decode byte 0x9b in position 10: invalid start byte报错
2. mock生成的ZipFile对象如何对接parse_json这类后续处理函数

问题1解决：UnicodeDecodeError报错处理 #

报错核心原因是：zip属于二进制格式文件，原测试代码中使用文本模式r读取本地zip文件，Python默认会尝试用UTF-8解码二进制内容，触发编码错误。
推荐两种修复方案：

• 方案一：修改本地文件读取模式为二进制读
  将open('local_file.zip', 'r')改为open('local_file.zip', 'rb')即可
• 方案二：内存生成测试zip包（更适合单元测试，无本地文件依赖）
  直接在测试代码中动态生成符合业务结构的zip包，不需要依赖本地文件，示例代码：

from io import BytesIO
from zipfile import ZipFile
import json

# 内存构造测试zip包
zip_buffer = BytesIO()
with ZipFile(zip_buffer, 'w') as zf:
    # 写入业务逻辑需要的json文件，路径和结构和生产环境保持一致
    test_json_content = {"some_key": [{"test_field": "value1"}, {"test_field": "value2"}]}
    zf.writestr("files/file1.json", json.dumps(test_json_content))
# 重置文件指针到开头，后续读取才能拿到完整内容
zip_buffer.seek(0)

上传到mock S3时直接传入构造好的二进制内容即可：

client.put_object(Bucket=self.bucket_name, Key=s3_prefix + "file.zip", Body=zip_buffer.read())

问题2解决：ZipFile对象对接后续处理函数 #

通过上述方案构造的zip包结构和生产环境完全一致，业务代码从S3读取内容后生成的是原生Python ZipFile 实例，不需要做任何额外适配，直接传给parse_json函数即可正常读取压缩包内的json文件，符合真实业务运行逻辑。

注意：原业务代码中存在笔误endswidth，需改为endswith才能正确筛选zip格式文件。

完整修正后测试代码示例 #

import os
import json
import boto3
import pytest
from io import BytesIO
from zipfile import ZipFile
from moto import mock_s3
# 替换为你自己的业务模块名
import your_business_module

@pytest.fixture(scope='function')
def aws_credentials():
    """Mock AWS凭证，避免测试访问真实AWS服务"""
    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(aws_credentials):
    # 测试参数配置
    test_bucket = "test-bucket-001"
    test_s3_prefix = "test_data/"
    
    # 初始化mock S3并创建测试桶
    s3 = boto3.resource('s3', region_name='us-east-1')
    s3.create_bucket(Bucket=test_bucket)

    # 内存生成测试zip包
    zip_buffer = BytesIO()
    with ZipFile(zip_buffer, 'w') as zf:
        test_json = {"some_key": [{"id": 1, "content": "test1"}, {"id": 2, "content": "test2"}]}
        zf.writestr("files/file1.json", json.dumps(test_json))
    zip_buffer.seek(0)

    # 上传测试文件到mock S3
    s3_client = boto3.client('s3', region_name='us-east-1')
    s3_client.put_object(Bucket=test_bucket, Key=test_s3_prefix + "test.zip", Body=zip_buffer.read())
    s3_client.put_object(Bucket=test_bucket, Key=test_s3_prefix + "normal.txt", Body=b"test content")

    # 执行业务逻辑
    result = your_business_module.get_files_from_s3(test_bucket, test_s3_prefix)
    
    # 断言验证结果
    assert len(result) == 2
    assert result[0]["id"] == 1

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