You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

无法Mock AWS S3 Client:Python中get_paginator函数模拟问题

解决Python Mock AWS S3 get_paginator的问题

问题根源

  1. Mock时机错误:你在setUp中提前实例化了RealClass,此时真实的boto3.client("s3")已经被调用,后续的@patch("boto3.client")无法覆盖这个已创建的客户端,导致测试时调用了真实的S3接口,出现token过期错误。
  2. Mock目标不对:需要patch被测试模块(handler.py)中导入的boto3.client,而非全局的boto3.client。
  3. 未完整Mock分页调用链:生产代码中依赖get_paginator→paginate→页面迭代的完整流程,需要逐一Mock这些环节的返回值。
  4. 测试数据不匹配业务逻辑:生产代码只筛选.content.txt或.metadata.json后缀的文件,你的测试用.txt不会被纳入结果列表。

修正后的测试代码

import unittest
from unittest.mock import patch, MagicMock

from handler import RealClass

class TestRealClass(unittest.TestCase):
    @patch("handler.boto3.client")  # 修正:patch handler模块内的boto3.client
    def test_get_unprocessed_files(self, mock_boto3_client):
        # 1. 构造Mock分页器及返回页面数据
        mock_paginator = MagicMock()
        # 让paginate返回可迭代的页面,注意文件后缀匹配生产筛选规则
        mock_paginator.paginate.return_value = [
            {
                "Contents": [
                    {"Key": "files/1001000284.content.txt"},
                    {"Key": "files/ignore.txt"},  # 该文件会被过滤
                    {"Key": "files/meta.metadata.json"}
                ]
            }
        ]
        # 让Mock的S3客户端返回Mock分页器
        mock_boto3_client.return_value.get_paginator.return_value = mock_paginator

        # 2. 实例化RealClass并补充必要属性(生产代码依赖bronze_bucket和prefix)
        real_instance = RealClass()
        real_instance.bronze_bucket = "test-bucket"
        real_instance.prefix = "files/"

        # 3. 执行测试方法
        result = real_instance.get_unprocessed_files()

        # 4. 断言结果正确性
        self.assertIsInstance(result, list)
        self.assertEqual(len(result), 2)  # 仅两个符合后缀的文件会被保留
        self.assertIn("files/1001000284.content.txt", result)
        self.assertIn("files/meta.metadata.json", result)

        # 可选:验证调用是否符合预期
        mock_boto3_client.assert_called_once_with("s3")
        mock_boto3_client.return_value.get_paginator.assert_called_once_with("list_objects_v2")
        mock_paginator.paginate.assert_called_once_with(
            Bucket="test-bucket",
            Prefix="files/"
        )

额外优化建议

在RealClass的__init__中初始化bronze_bucket和prefix属性,避免测试时手动设置:

class RealClass:
    def __init__(self, bronze_bucket="default-bucket", prefix=""):
        self.s3_client = boto3.client("s3")
        self.bronze_bucket = bronze_bucket
        self.prefix = prefix

测试时可直接传入参数:real_instance = RealClass("test-bucket", "files/")

内容的提问来源于stack exchange,提问作者NeoFahrenheit

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.18 19:33:18