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

如何Mock Boto3 S3的put_bucket_lifecycle_configuration方法做单元测试

我来帮你搞定这个测试问题!既然moto还没支持put_bucket_lifecycle_configuration方法,咱们可以用unittest.mock来mock S3客户端,或者稍微重构代码让测试更顺手。下面给你两种可行的方案:

方案1:直接用unittest.mock mock boto3客户端(无需修改原有代码)

这个方法不用动你的被测代码,直接通过mock拦截boto3.client的调用,捕获后续S3方法的参数即可:

import unittest
from unittest.mock import patch, MagicMock
from sut import create_a_bucket

class TestS3BucketLifecycle(unittest.TestCase):
    @patch('boto3.client')
    def test_put_bucket_lifecycle_with_correct_rules(self, mock_boto_client):
        # 让mock的boto3.client返回一个MagicMock对象,用来捕获所有S3调用
        mock_s3_client = MagicMock()
        mock_boto_client.return_value = mock_s3_client
        
        # 调用要测试的函数
        test_bucket_name = "my-test-bucket"
        create_a_bucket(test_bucket_name)
        
        # 先验证create_bucket是否被正确调用
        mock_s3_client.create_bucket.assert_called_once_with(Bucket=test_bucket_name)
        
        # 获取put_bucket_lifecycle_configuration的调用参数
        lifecycle_call_args = mock_s3_client.put_bucket_lifecycle_configuration.call_args
        
        # 验证Bucket参数正确
        self.assertEqual(lifecycle_call_args[1]['Bucket'], test_bucket_name)
        
        # 这里替换成你从lifecycle_config计算出来的预期规则
        expected_rules = [
            {'ID': 'sample-rule', 'Prefix': 'logs/', 'Status': 'Enabled', 'Expiration': {'Days': 7}}
        ]
        # 验证Rules参数和预期一致
        self.assertEqual(lifecycle_call_args[1]['LifecycleConfiguration']['Rules'], expected_rules)

原理说明:

我们用@patch('boto3.client')装饰测试函数,让所有对boto3.client("s3")的调用都返回我们定义的MagicMock对象。这个mock会记录所有方法调用的参数,我们通过call_args就能拿到put_bucket_lifecycle_configuration的调用细节,和预期值对比即可。

方案2:重构代码实现依赖注入(更推荐,代码更易维护)

如果想让代码的可测试性更好,推荐稍微重构一下,把S3客户端作为可选参数传入S3Bucket类。这样测试时可以直接传入mock客户端,不需要mock整个boto3:

重构后的被测代码(sut.py):

import boto3

class S3Bucket(object):
    def __init__(self, name, lifecycle_config, s3_client=None):
        self.name = name
        self.lifecycle_config = lifecycle_config
        # 如果没传入客户端,就用默认的boto3 S3客户端
        self.s3_client = s3_client or boto3.client("s3")

    def create(self):
        self.s3_client.create_bucket(Bucket=self.name)
        # 把计算规则的逻辑抽成单独方法,方便单独测试
        rules = self._calculate_lifecycle_rules()
        self.s3_client.put_bucket_lifecycle_configuration(
            Bucket=self.name,
            LifecycleConfiguration={"Rules": rules}
        )
    
    def _calculate_lifecycle_rules(self):
        # 这里放你原来从lifecycle_config生成rules的逻辑
        # 举个示例:
        return [
            {
                'ID': 'auto-cleanup',
                'Prefix': '',
                'Status': 'Enabled',
                'Expiration': {'Days': self.lifecycle_config.get('expire_after_days')}
            }
        ]

def create_a_bucket(name):
    lifecycle_policy = {"expire_after_days": 30}  # 你的原有配置逻辑
    bucket = S3Bucket(name, lifecycle_policy)
    bucket.create()
    return bucket

对应的测试代码:

import unittest
from unittest.mock import MagicMock
from sut import S3Bucket

class TestS3BucketLifecycle(unittest.TestCase):
    def test_lifecycle_configuration_parameters(self):
        # 创建一个mock的S3客户端
        mock_s3 = MagicMock()
        test_bucket = "test-bucket-123"
        test_config = {"expire_after_days": 30}
        
        # 实例化S3Bucket时传入mock客户端
        bucket = S3Bucket(test_bucket, test_config, s3_client=mock_s3)
        bucket.create()
        
        # 验证create_bucket调用正确
        mock_s3.create_bucket.assert_called_once_with(Bucket=test_bucket)
        
        # 验证put_bucket_lifecycle_configuration的参数符合预期
        expected_rules = [
            {'ID': 'auto-cleanup', 'Prefix': '', 'Status': 'Enabled', 'Expiration': {'Days': 30}}
        ]
        mock_s3.put_bucket_lifecycle_configuration.assert_called_once_with(
            Bucket=test_bucket,
            LifecycleConfiguration={"Rules": expected_rules}
        )
    
    # 还可以单独测试规则计算的逻辑
    def test_rule_calculation(self):
        test_config = {"expire_after_days": 90}
        bucket = S3Bucket("dummy-name", test_config)
        calculated_rules = bucket._calculate_lifecycle_rules()
        self.assertEqual(calculated_rules[0]['Expiration']['Days'], 90)

为什么推荐这个方案?

  • 符合依赖注入原则,代码的耦合度更低,以后替换S3客户端实现也更方便。
  • 可以单独测试规则计算的逻辑,不用依赖整个S3客户端的调用流程,测试更聚焦。

关于botocore.stub.Stubber的说明:

Stubber需要你提前定义好预期的调用顺序和参数,如果代码中的调用和预期有偏差(比如顺序不对、参数不匹配)就会报错,灵活性不如MagicMock。如果只是想验证参数是否正确,用MagicMock会更简单直接。

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

相关产品推荐
方舟 Agent Plan

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

最近更新时间:2026.05.29 06:59:39