How to mock elasticsearch.helpers.bulk

Question

I am testing a class that uses bulk to index some documents.

Here is my code:

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)

However, the mocking does not seem to work... when I run it, I get a long list of connection refused errors.

How do I fix it?

Leo · Accepted Answer · 2020-09-03 13:32:25Z

1

It turns out that my patch was wrong... Here is how I fixed it:

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

answered Sep 3, 2020 at 13:32

Leo

5,1822 gold badges33 silver badges74 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

zar3bski Over a year ago

Faced similar issues while mocking this lib. Quite hard to get the paths right because of how modules are imported (even more with _async ressources). Say I am using from elasticsearch.helpers import async_bulk what would the path be?

Collectives™ on Stack Overflow

How to mock elasticsearch.helpers.bulk

1 Answer 1

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related