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

编程问答 2022-05-25

问题描述

我正在尝试使用 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,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 的文章，但似乎不知道如何在此处应用它。

解决方法

模拟和修补的想法是人们想要模拟/修补特定的东西。因此，要进行正确的修补，必须准确指定要模拟/修补的内容。在给定的例子中，需要修补的东西位于：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:
,

最后，由于@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,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() 而不必将其包装在另一个函数中，我仍然很感兴趣！

,

为了使用 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
    ...
boto3motopytestpytest-mockunit-testing