Блог

Я тестирую класс, который использует 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)
  

Однако, похоже, что издевательство не работает … когда я запускаю его, я получаю длинный список ошибок отказа в подключении.

Как мне это исправить?

Оказывается, мой патч был неправильным… Вот как я это исправил:

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