pytest 断言上下文管理器内部的方法是否被调用

Question

我已经根据他们的示例实现了 InfluxDB 的接口。它使用 context_manager 内的 DB 写入：

from influxdb_client import InfluxDBClient, Point, WriteOptions


class InfluxInterface:

    def __init__(self):
        self.__client = None
        self.bucket_name = 'bucket_name'

    def connect(self, url=IFDB_URL, token=IFDB_TOKEN, org=IFDB_ORG, timeout=IFDB_TIMEOUT, debug=False):
        self.__client = InfluxDBClient(url=url, token=token, org=org, timeout=timeout, debug=debug)

    def is_connected(self):
        return self.__client is not None

    def insert_stuff(self, stuff: List[Tag]):
        if not self.is_connected():
            return

        points = []

        for item in stuff:
            point = Point('measurment_name').time(item.item_ts).tag('tag_name', item.tag).field('field_name', item.field)
            points.append(point)

        with self.__client.write_api(write_options=WriteOptions(batch_size=5000, flush_interval=5000),
                                     success_callback=self.batch_write_success,
                                     error_callback=self.batch_write_error,
                                     retry_callback=self.batch_write_retry) as write_api:
            write_api.write(bucket=self.bucket_name, record=points)
            # Write is not returning anything (returning None)

现在，我想使用

pytest

对它进行单元测试，但我失败了..

我想做的第一件事是，嘲笑

is_connected()

，总是返回

True

。我这样做：

mocker.patch.object(InfluxInterface, "is_connected").return_value = True

我的目标是模拟

write_api.write(bucket=self.bucket_name, record=points)

，因为我不想写入数据库，或连接到数据库，甚至需要将数据库作为对这些东西进行单元测试的前提。我只想断言是否使用正确的参数调用了

write_api.write

。所以我准备了这个测试方法：

import pytest
from my_lib.influx_lib import InfluxInterface

class TestInfluxInterface:
    def test_insert_stuff(self, mocker):
        # influx_client = mocker.patch('my_lib.influx_lib.InfluxDBClient', autospec=True)
        # Is connected, return True
        fake_is_connected = mocker.patch.object(InfluxInterface, "is_connected").return_value = True

        # fake_client = mocker.patch.object(InfluxInterface, '__client', autospec=True)

        fake_write_api = mocker.patch.object(InfluxInterface, '_InfluxInterface__client').return_value.__enter__.write

        test_connector = InfluxInterface()
        test_list = [test_item_1, test_item_2, test_item_3]
        result = test_connector.insert_stuff(test_list)
        fake_write_api.assert_called_once_with(bucket=self.bucket_name, record=points)

正如您在上面代码的注释中看到的，我一直在尝试各种不同的方法来正确模拟上下文管理器和

write_api.write

方法。但这些都不起作用。

---

编辑：

这就是我到目前为止所取得的成绩：

def test_insert_tags(self, mocker):
    influx_client = mocker.patch('my_lib.influx_lib.InfluxDBClient')  # , autospec=True)
    # Is connected, return Status.ok
    mocker.patch.object(InfluxInterface, "is_connected").return_value = True

    test_connector = InfluxInterface()
    fake_write_api = mocker.MagicMock()
    test_connector._InfluxInterface__client = influx_client
    # test_connector._InfluxInterface__client.write_api.return_value = mocker.Mock(w_api=fake_write_api, __enter__=influx_client, __exit__=mocker.Mock())
    # test_connector._InfluxInterface__client.write_api.return_value = mocker.Mock(__enter__=fake_write_api, __exit__=mocker.Mock())
    # test_connector._InfluxInterface__client.write_api.return_value.__enter__.w_api.write = fake_write_api

    test_taglist = [analog_test_tag, discrete_test_tag, string_test_tag]
    result = test_connector.insert_tags(test_taglist, RecordType.all_values)

    print(f'influx_client has calls: {influx_client.write_api.mock_calls}')
    print(f'fake_write_api has calls: {fake_write_api.mock_calls}')
    print(f'fake_write_api.write has calls: {fake_write_api.write.mock_calls}')

    fake_write_api.write.assert_called_once()
    assert result.result_status.status_code == Tconf.status_ok.status_code

并且代码实际运行起来，只是assert不满足。 （所有 3 个带注释的设置

return_value

的调用都通过，但在每种情况下都不会调用

fake_write_api

。）

这是按照上面的代码运行测试的结果。

influx_client has calls:
 [call(write_options=<influxdb_client.client.write_api.WriteOptions object at 0x7f8d9e3bf580>, success_callback=<bound method InfluxInterface.batch_write_success of <my_lib.influx_lib.InfluxInterface object at 0x7f8d9e34d6c0>>, error_callback=<bound method InfluxInterface.batch_write_error of <my_lib.influx_lib.InfluxInterface object at 0x7f8d9e34d6c0>>, retry_callback=<bound method InfluxInterface.batch_write_retry of <my_lib.influx_lib.InfluxInterface object at 0x7f8d9e34d6c0>>),
  call().__enter__(),
  call().__enter__().write(bucket='bucket_name', record=[<influxdb_client.client.write.point.Point object at 0x7f8d9e3bd690>, <influxdb_client.client.write.point.Point object at 0x7f8d9e3bd6f0>]),
  call().__exit__(None, None, None)]
fake_write_api has calls: []
fake_write_api.write has calls: []

Answer 1

就我而言，我只需要断言，为此我使用了mock_calls。请参阅下面的示例：

assert influx_client.mock_calls[2] == call().__enter__().write(...)

嘲笑它似乎更难。你成功嘲笑它了吗？