python-3.x - 如何模拟 elasticsearch.helpers.bulk

Question

我正在测试一个bulk用于索引某些文档的类。

这是我的代码：

import mock
import unittest
import json

from elasticsearch.helpers import bulk

from ingestion.ingestor import Ingestor

class TestIngestor(unittest.TestCase):

    def setUp(self):
        self.ingestor = Ingestor()

    @mock.patch("elasticsearch.helpers.bulk", mock.MagicMock(return_value=True))
    def test_ingestor(self):
        with open("tests/data/sample_payload.json", "r") as reader:
            sample_messages = json.loads(reader.read())["Messages"]

        actions = self.ingestor.ingest(sample_messages)

        self.assertEqual(len(actions), 10)

但是，模拟似乎不起作用......当我运行它时，我得到一长串连接被拒绝的错误。

我如何解决它？

标签： python-3.xunit-testingelasticsearchmocking

Answer 1

事实证明我的补丁是错误的......这是我修复它的方法：

@mock.patch("elasticsearch.Elasticsearch.bulk", 
             mock.MagicMock(return_value={"items":[]}))