测试在其主体中调用 boto3 的函数时如何模拟 boto3 调用

Question

我正在尝试使用 pytest 测试一个名为 get_date_from_s3(bucket, table) 的函数。在这个函数中，我想在测试期间模拟一个 boto3.client("s3").list_objects_v2() 调用，但我似乎无法弄清楚它是如何工作的。

这是我的目录设置：

my_project/
  glue/
      continuous.py
  tests/
      glue/
          test_continuous.py
          conftest.py
      conftest.py

代码 continuous.py 将在 AWS 粘合作业中执行，但我正在本地对其进行测试。

my_project/glue/continuous.py

​​>

import boto3

def get_date_from_s3(bucket, table):
    s3_client = boto3.client("s3")
    result = s3_client.list_objects_v2(Bucket=bucket, Prefix="Foo/{}/".format(table))

    # [the actual thing I want to test]
    latest_date = datetime_date(1, 1, 1)
    output = None
    for content in result.get("Contents"):
        date = key.split("/")
        output = [some logic to get the latest date from the file name in s3]

    return output

def main(argv):
    date = get_date_from_s3(argv[1], argv[2])

if __name__ == "__main__":
    main(sys.argv[1:])

my_project/tests/glue/test_continuous.py

​​>

这就是我想要的：我想通过模拟 s3_client.list_objects_v2() 并将响应值显式设置为 example_response 来测试 get_date_from_s3()。我尝试做类似下面的事情，但它不起作用：

from glue import continuous
import mock

def test_get_date_from_s3(mocker):
    example_response = {
        "ResponseMetadata": "somethingsomething",
        "IsTruncated": False,
        "Contents": [
            {
                "Key": "/year=2021/month=01/day=03/some_file.parquet",
                "LastModified": "datetime.datetime(2021, 2, 5, 17, 5, 11, tzinfo=tzlocal())",
                ...
            },
            {
                "Key": "/year=2021/month=01/day=02/some_file.parquet",
                "LastModified": ...,
            },
            ...
        ]
    }
    
    mocker.patch(
        'continuous.boto3.client.list_objects_v2',
        return_value=example_response
    )

   expected = "20210102"
   actual = get_date_from_s3(bucket, table)
    
   assert actual == expected

注意

我注意到许多模拟示例都具有作为类的一部分进行测试的功能。因为 Continuous.py 是一项粘合工作，所以我没有找到创建类的实用程序，我只有函数和调用它的 main()，这是一种不好的做法吗？函数之前的模拟装饰器似乎仅用于作为类的一部分的函数。 我还阅读了有关moto 的信息，但似乎无法弄清楚如何在此处应用它。

Answer 1

模拟和修补的想法是人们想要模拟/修补特定的东西。因此，要进行正确的修补，必须准确指定要模拟/修补的东西。在给定的示例中，要修补的东西位于：glue > Continuous > boto3 > client instance > list_objects_v2。

正如您指出的那样，您希望调用 list_objects_v2() 以返回准备好的数据。因此，这意味着您必须先模拟“glue.continuous.boto3.client”，然后使用后者模拟“list_objects_v2”。

在实践中，您需要按照以下方式做一些事情：

from glue import continuous_deduplicate
from unittest.mock import Mock, patch

@patch("glue.continuous.boto3.client")
def test_get_date_from_s3(mocked_client):
    mocked_response = Mock()
    mocked_response.return_value = { ... }
    mocked_client.list_objects_v2 = mocked_response

    # Run other setup and function under test:

Answer 2

最后，感谢@Gros Lalo，我发现我的修补目标值是错误的。应该是'glue.continuous.boto3.client.list_objects_v'。但这仍然不起作用，它给我带来了错误AttributeError: <function client at 0x7fad6f1b2af0> does not have the attribute 'list_objects_v'。

所以我做了一些重构，将整个 boto3.client 包装在一个更容易模拟的函数中。这是我的新my_project/glue/continuous.py 文件：

import boto3

def get_s3_objects(bucket, table):
    s3_client = boto3.client("s3")
    return s3_client.list_objects_v2(Bucket=bucket, Prefix="Foo/{}/".format(table))

def get_date_from_s3(bucket, table):
    result = get_s3_objects(bucket, table)

    # [the actual thing I want to test]
    latest_date = datetime_date(1, 1, 1)
    output = None
    for content in result.get("Contents"):
        date = key.split("/")
        output = [some logic to get the latest date from the file name in s3]

    return output

def main(argv):
    date = get_date_from_s3(argv[1], argv[2])

if __name__ == "__main__":
    main(sys.argv[1:])

因此，我的新test_get_latest_date_from_s3() 是：

def test_get_latest_date_from_s3(mocker):
    example_response = {
        "ResponseMetadata": "somethingsomething",
        "IsTruncated": False,
        "Contents": [
            {
                "Key": "/year=2021/month=01/day=03/some_file.parquet",
                "LastModified": "datetime.datetime(2021, 2, 5, 17, 5, 11, tzinfo=tzlocal())",
                ...
            },
            {
                "Key": "/year=2021/month=01/day=02/some_file.parquet",
                "LastModified": ...,
            },
            ...
        ]
    }
    mocker.patch('glue.continuous_deduplicate.get_s3_objects', return_value=example_response)

    expected_date = "20190823"
    actual_date = continuous_deduplicate.get_latest_date_from_s3("some_bucket", "some_table")

    assert expected_date == actual_date

重构对我有用，但是如果有办法直接模拟 list_objects_v2() 而无需将其包装在另一个函数中，我仍然感兴趣！

Answer 3

为了使用 moto 实现此结果，您必须使用 boto3-sdk 正常创建数据。换句话说：创建一个对 AWS 本身成功的测试用例，然后在其上添加 moto-decorator。

对于您的用例，我想它看起来像：

from moto import mock_s3

@mock_s3
def test_glue:
    # create test data
    s3 = boto3.client("s3")
    for d in range(5):
        s3.put_object(Bucket="", Key=f"year=2021/month=01/day={d}/some_file.parquet", Body="asdf")
    # test
    result = get_date_from_s3(bucket, table)
    # assert result is as expected
    ...