Python版AWS Lambda单元测试：Mock S3调用解决文件找不到问题

问题根源 #

你当前的Mock仅替换了boto3模块，但未模拟boto3.resource("s3")返回的资源实例、S3 Bucket对象、文件列表以及download_file方法的行为。代码执行时，download_s3_folder无法获取到模拟的S3文件，自然不会在/tmp/records下创建目标文件，最终触发FileNotFoundError。

解决方案 #

下面提供两种可行的Mock方案，按需选择：

方案一：直接Mock下载函数（简单高效，聚焦业务逻辑） #

跳过S3调用细节，直接Mock下载函数，手动创建测试所需文件：

import os
import sys
import unittest
import mock

# Mock boto3模块
boto3_mock = mock.MagicMock()
sys.modules["boto3"] = boto3_mock

from testing_lambda import testing_lambda

class TestingLambdaTests(unittest.TestCase):
    def setUp(self):
        # 创建测试目录和文件
        self.test_dir = "/tmp/records"
        self.test_file = os.path.join(self.test_dir, "go_message.json.j2")
        os.makedirs(self.test_dir, exist_ok=True)
        # 写入测试内容
        with open(self.test_file, "w") as f:
            f.write('{"test": "data"}')

    def tearDown(self):
        # 清理测试文件和目录
        if os.path.exists(self.test_file):
            os.remove(self.test_file)
        if os.path.exists(self.test_dir):
            os.rmdir(self.test_dir)

    @mock.patch.object(testing_lambda, "download_records")
    def test_lambda_handler(self, mock_download):
        # Mock下载函数无实际操作
        mock_download.return_value = None
        # 执行Lambda处理器
        result = testing_lambda.lambda_handler(None, context="")
        # 验证下载函数被调用
        mock_download.assert_called_once()

    def test_package_for_promo(self):
        self.test_lambda_handler()

if __name__ == "__main__":
    unittest.main()

方案二：完整Mock S3调用链（适合测试下载逻辑） #

如果需要验证download_s3_folder的逻辑，需逐层Mock S3的各个调用环节：

import os
import sys
import unittest
import mock

sys.modules["boto3"] = mock.MagicMock()

from testing_lambda import testing_lambda

class TestingLambdaTests(unittest.TestCase):
    def setUp(self):
        # 创建测试目录
        self.test_dir = "/tmp/records"
        os.makedirs(self.test_dir, exist_ok=True)

    def tearDown(self):
        # 清理测试目录
        if os.path.exists(self.test_dir):
            os.rmdir(self.test_dir)

    def test_lambda_handler_with_s3_mock(self):
        # 模拟S3资源实例
        s3_resource_mock = mock.MagicMock()
        testing_lambda.s3 = s3_resource_mock

        # 模拟Bucket对象
        bucket_mock = mock.MagicMock()
        s3_resource_mock.Bucket.return_value = bucket_mock

        # 模拟S3中的目标文件对象
        mock_obj = mock.MagicMock()
        mock_obj.key = "subscription/go_message.json.j2"
        # 模拟文件列表迭代器
        bucket_mock.objects.filter.return_value = [mock_obj]

        # 模拟download_file行为：手动创建本地文件
        def mock_download_file(key, target):
            with open(target, "w") as f:
                f.write('{"test": "data"}')
        bucket_mock.download_file.side_effect = mock_download_file

        # 执行Lambda处理器
        result = testing_lambda.lambda_handler(None, context="")

        # 验证S3调用链
        s3_resource_mock.Bucket.assert_called_once_with("sss-records-dev")
        bucket_mock.objects.filter.assert_called_once_with(Prefix="subscription")
        bucket_mock.download_file.assert_called_once_with(
            "subscription/go_message.json.j2",
            "/tmp/records/go_message.json.j2"
        )

    def test_package_for_promo(self):
        self.test_lambda_handler_with_s3_mock()

if __name__ == "__main__":
    unittest.main()

额外优化建议 #

• 原lambda_handler仅捕获ValueError，但FileNotFoundError会直接抛出，建议调整异常捕获范围：

  def lambda_handler(event, context):
      try:
          download_records()
          with open(os.path.join("/tmp/records", "go_message.json.j2")) as f:
              record = f.read()
      except (ValueError, FileNotFoundError):
          return "record could not be found"
  

• 路径拼接使用os.path.join，避免硬编码拼接出错。

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