0

我正在尝试测试一个get_date_from_s3(bucket, table)使用 pytest 调用的函数。在这个函数中，有一个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() 并将响应值显式设置为来测试 get_date_from_s3() example_response。我尝试做类似下面的事情，但它不起作用：

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，但似乎无法弄清楚如何在这里应用它。

4

3 回答 3

0

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

于 2021-02-16T17:42:25.230 回答
0

为了使用 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
    ...
于 2021-03-27T08:58:03.243 回答
0

模拟和修补的想法是想模拟/修补特定的东西。因此，要进行正确的修补，必须准确指定要模拟/修补的东西。在给定的示例中，要修补的东西位于：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:
于 2021-02-16T15:41:49.093 回答