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.8.1:

• Add support for Python 3.9 (Thanks @singingwolfboy)
• use unittest.mock instead of mock (Thanks @singingwolfboy)
• Add must_not for bool search query (Thanks @t-bittarn)

1.8.0:

• Add multi_match (Thanks @carlosgalvez-tiendeo)
• Add mget (Thanks @carlosgalvez-tiendeo)
• Add create, update, and delete to bulk API (Thanks @fenimore)
• Add Should to bool Query (Thanks @fenimore)
• Update Search API return result (Thanks @fenimore)

1.7.0:

• Add shards skipped to search and count (Thanks @philtweir)
• Allow 'match_all' queries in FakeSearch (Thanks @jankislinger)
• Query using nested attributes (Thanks @jankislinger)
• New features: range, size, aggregations (Thanks @jankislinger)
• Adding "should" and "minimum_should_match" to QueryType (Thanks @lunarie16)

1.6.2:

• Add must to query type (Thanks @cuent)
• Add match all query type (Thanks @cuent)

1.6.1:

• Fix Twine README.md

1.6.0:

• Implements several basic search types (Thanks @KyKoPho)
• Allow ignoring of missing documents (404) for get and delete (Thanks @joosterman)

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