无法模拟 AWS S3 客户端

问题描述 投票:0回答:1

我正在尝试模拟对 Python AWS S3 客户端上的 get_paginator 函数的调用。这是我的生产代码:

处理程序.py

import boto3

class RealClass:
    def __init__(self):
        self.s3_client = boto3.client("s3")

    def get_unprocessed_files(self) -> list[str]:
        paginator = self.s3_client.get_paginator("list_objects_v2")
        operation_parameters = {"Bucket": self.bronze_bucket, "Prefix": self.prefix}
        page_iterator = paginator.paginate(**operation_parameters)

        un_processed_files = []
        for page in page_iterator:
            for obj in page.get("Contents", []):
                key = obj["Key"]
                if key.endswith(".content.txt") or key.endswith(".metadata.json"):
                    un_processed_files.append(key)

        return un_processed_files

test_handler.py

import unittest
from unittest.mock import patch, MagicMock

from handler import RealClass

class TestRealClass(unittest.TestCase):
    def setUp(self) -> None:
        self.real = RealClass()

    @patch("boto3.client")
    def test_get_unprocessed_files(self, mock_boto3_client):
        response = [
            {
                "Contents": [
                    {"Key": "files/1001000284.txt"},
                ]
            }
        ]

        # What to do here?

        result = self.pii.get_unprocessed_files()
        self.assertIsInstance(result, list)
        self.assertTrue(len(result) > 0)
        self.assertTrue(result[0].find("1001000284") > -1)

我得到的只是提供的令牌已过期,这意味着我猜真正的功能没有被嘲笑。

谢谢!

python amazon-s3 mocking boto3 python-unittest
1个回答
0
投票

模拟设置

您缺少模拟的设置。几乎在所有语言中,在模拟对象之后,您需要根据您的情况设置所需的行为:

当调用 get_paginator 方法时,返回这个 fake 响应

也许用更描述性的语言,你就能明白。

JAVA : getBook() 方法执行时,返回一本假书

enter image description here

C#在下面的示例中,模拟被设置为仅在执行 GetPeople() 方法时返回人员列表(而不是真正的数据库)

enter image description here

mock_boto3_client.get_paginator.return_value

我没有测试这段代码,但我认为这是你所需要的

import unittest
from datetime import datetime
from unittest.mock import patch
# real handler
import real_code

class TestRealClass(unittest.TestCase):

    @patch('boto3.client')
    def test_get_unprocessed_files(self, mock_boto3_client):

        desired_response = {...}

        mock_boto3_client.return_value = mock_boto3_client
        mock_boto3_client.get_paginator.return_value = desired_response

        result_value = real_code.get_unprocessed_files("MyTestParameterName") 

参考文献

建议

如果您可以与我们分享一个可复制的 hello world 代码来帮助您

© www.soinside.com 2019 - 2024. All rights reserved.