ElasticMock

Python Elasticsearch Mock for test purposes

Installation

pip install ElasticMock

Usage

To use ElasticMock, decorate your test method with @elasticmock decorator:

from unittest import TestCase

from elasticmock import elasticmock


class TestClass(TestCase):

    @elasticmock
    def test_should_return_something_from_elasticsearch(self):
        self.assertIsNotNone(some_function_that_uses_elasticsearch())

Custom Behaviours

You can also force the behaviour of the ElasticSearch instance by importing the elasticmock.behaviour module:

from unittest import TestCase

from elasticmock import behaviour


class TestClass(TestCase):

    ...

    def test_should_return_internal_server_error_when_simulate_server_error_is_true(self):
        behaviour.server_failure.enable()
        ...
        behaviour.server_failure.disable()

You can also disable all behaviours by calling behaviour.disable_all() (Consider put this in your def tearDown(self) method)

Available Behaviours

• server_failure: Will make all calls to ElasticSearch returns the following error message:

  {
      'status_code': 500,
      'error': 'Internal Server Error'
  }
  

Code example

Let's say you have a prod code snippet like this one:

import elasticsearch

class FooService:

    def __init__(self):
        self.es = elasticsearch.Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])

    def create(self, index, body):
        es_object = self.es.index(index, body)
        return es_object.get('_id')

    def read(self, index, id):
        es_object = self.es.get(index, id)
        return es_object.get('_source')

Than you should be able to test this class by mocking ElasticSearch using the following test class:

from unittest import TestCase
from elasticmock import elasticmock
from foo.bar import FooService

class FooServiceTest(TestCase):

    @elasticmock
    def should_create_and_read_object(self):
        # Variables used to test
        index = 'test-index'
        expected_document = {
            'foo': 'bar'
        }

        # Instantiate service
        service = FooService()

        # Index document on ElasticSearch
        id = service.create(index, expected_document)
        self.assertIsNotNone(id)

        # Retrive dpcument from ElasticSearch
        document = service.read(index, id)
        self.assertEquals(expected_document, document)

Notes:

• The mocked search method returns all available documents indexed on the index with the requested document type.
• The mocked suggest method returns the exactly suggestions dictionary passed as body serialized in Elasticsearch.suggest response. Atention: If the term is an int, the suggestion will be python term + 1. If not, the suggestion will be formatted as python {0}_suggestion.format(term) . Example:
  • Suggestion Body:

   suggestion_body = {
       'suggestion-string': {
           'text': 'test_text',
           'term': {
               'field': 'string'
           }
       },
       'suggestion-id': {
           'text': 1234567,
           'term': {
               'field': 'id'
           }
       }
   }
  

  • Suggestion Response:

  {
      'suggestion-string': [
          {
              'text': 'test_text',
              'length': 1,
              'options': [
                  {
                      'text': 'test_text_suggestion',
                      'freq': 1,
                      'score': 1.0
                  }
              ],
              'offset': 0
          }
      ],
      'suggestion-id': [
          {
              'text': 1234567,
              'length': 1,
              'options': [
                  {
                      'text': 1234568,
                      'freq': 1,
                      'score': 1.0
                  }
              ],
              'offset': 0
          }
      ],
  }
  

Testing

python setup.py test

Changelog

1.5.1:

• Fix tests for es > 7 (Thanks @chesstrian)

1.5.0:

• FakeElasticSearch: Mocked indices property
  • FakeIndicesClient: Mocked create, exists, refresh and delete methods
• FakeElasticSearch: Mocked cluster property
  • FakeClusterClient: Mocked health method

1.4.0

• Fix es.index regression issue
• Add 'Force Server Failure' feature as requested
• Reformat code to be compliant with PEP8
• Add support to Python 3.8

1.3.7

• Adding fix for updating existing doc using index (Thanks @adityaghosh)
• Added bulk method (Thanks @charl-van-niekerk)
• Add default value to doc_type in index method as it is by default set to '_doc' (Thanks @mohantyashish109)
• Add support for Python 3.7 (Thanks @asherf)

1.3.6

• Fix installation issue (Thanks @tdhopper)

1.3.5

• Fix 1.3.4 release (Thanks @infinite-Joy)

1.3.4

• Added aggregations to response if requested (Thanks @snakeye)
• Implementing new methods for scrolling (Thanks @tcatrain)

1.3.3

• Search: doc_type can be a list (Thanks @garncarz)
• Exclude tests package (Thanks @jmlw)
• Make the FakeElasticsearch init signature match the one from Elasticsearch (Thanks @xrmx)
• Improve search and count (Thanks @frivoire)

1.3.2

• elasticmock: Python 3 fixes (Thanks @barseghyanartur)
• test: Add information on testing (Thanks @barseghyanartur)
• README.md: Fixed typo (Thanks @bowlofstew)

1.3.1

• elasticmock: Allow the same arguments to the mock that elasticsearch.Elasticsearch allows (Thanks @mattbreeden)

1.3.0:

• FakeElasticSearch: Mocked count method (Thanks @TheoResources)

1.2.0:

• FakeElasticSearch: Mocked suggest method

1.1.1:

• elasticmock: Changing the cleanup older FakeElasticSearch's instances order
• FakeElasticSearch.index: Changing the method signature to correctly overrides the Elasticsearch.index method

1.1.0:

• FakeElasticSearch: Mocked delete method

1.0.1:

• setup.py: Fixed GitHub link

1.0.0:

• elasticmock: Created @elasticmock decorator
• FakeElasticSearch: Mocked exists, get, get_source, index, info, search and ping method