Python单元测试AWS Lambda处理器的S3事件模拟报错问题

报错来自两处不匹配问题：

• 类型不匹配：s3.Bucket()方法要求传入字符串类型的桶名，但测试构造的event里bucket字段是嵌套字典{'name': S3_BUCKET_NAME}，boto3做桶名格式校验时收到dict类型，直接抛出TypeError。
• 逻辑不匹配：现有handler代码会遍历桶下所有对象做处理，完全没用到event里传入的object key，既不符合S3事件触发单文件处理的常规逻辑，测试时也容易因为桶内存在其他无关文件导致用例失败。

方案1：保持现有handler逻辑，仅修正测试事件 #

如果你的handler设计就是要求event里bucket字段直接传桶名字符串，不需要兼容标准S3触发事件，直接改测试用例里的event构造即可：

def test_handler(self):
    from functions.spell_correction.src.index import handler

    event = {
        # 直接传桶名字符串，不要传嵌套字典
        "bucket": S3_BUCKET_NAME
    }

    result = handler(event, {})
    self.assertEqual(result, {"success": True, "response": []})

注意：这个方案下要确保测试桶里只有你上传的那一个测试文件，否则handler遍历全桶时读到其他文件会导致用例不符合预期。

方案2：对齐标准S3事件规范，修正handler逻辑（生产环境推荐） #

如果这个Lambda是配置为S3上传自动触发，标准S3事件的桶名和key都在Records数组的嵌套结构里，不需要遍历全桶，直接读取触发事件的单个文件即可，修改后的handler代码：

def handler(event, context):
    print(f'Event: {event}')
    s3 = boto3.resource('s3')
    word_correction = correction.WordCorrection()

    for record in event.get('Records', []):
        # 从标准S3事件结构中解析桶名和文件key
        bucket_name = record['s3']['bucket']['name']
        file_key = record['s3']['object']['key']
        bucket = s3.Bucket(bucket_name)
        
        # 直接读取目标文件，不需要遍历全桶所有对象
        body = bucket.Object(file_key).get()['Body'].read()
        data = get_data_from_file(body)

        if not file_key.endswith('.json'):
            return {
                "success": False,
                "response": "Invalid file type. File must have .json extension."
            }
        
        try:
            word_correction.create_duplicated_words_file(data)
        except Exception as e:
            print(e)
            return {
                "success": False,
                "response": f"Failed to read file - {e}"
            }
        
        try:
            corrected_word_list = word_correction.spell_words(json.loads(body))
        except Exception as e:
            print(e)
            return {
                "success": False,
                "response": f"Failed to correct words - {e}"
            }
        
        return {
            "success": True,
            "response": corrected_word_list
        }

    return {
        "success": False,
        "response": "No valid S3 record found in event"
    }

对应的测试事件要对齐标准S3事件结构：

def test_handler(self):
    from functions.spell_correction.src.index import handler

    event = {
        "Records": [
            {
                "s3": {
                    "bucket": {"name": S3_BUCKET_NAME},
                    "object": {"key": S3_TEST_FILE_KEY}
                }
            }
        ]
    }

    result = handler(event, {})
    self.assertEqual(result, {"success": True, "response": []})

• 当前setUp逻辑用moto创建模拟桶、上传测试文件的写法是正确的，保持region_name和创建桶的区域一致即可。
• 如果后续测试用的文件key包含空格、中文等特殊字符，记得S3事件里的key是URL编码的，handler里需要先做解码再处理。
• 遍历全桶的写法在生产环境会有性能问题：桶内文件数量大时列举对象耗时长、容易触发Lambda超时，还可能因为IAM权限没有配置s3:ListBucket导致运行失败，建议优先选方案2。

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