Python Elasticsearch Mock for test purposes
MIT License
Python Elasticsearch Mock for test purposes
pip install ElasticMock
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())
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)
server_failure
: Will make all calls to ElasticSearch returns the following error message:
{
'status_code': 500,
'error': 'Internal Server Error'
}
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)
python term + 1
. If not, the suggestion will be formatted as python {0}_suggestion.format(term)
. suggestion_body = {
'suggestion-string': {
'text': 'test_text',
'term': {
'field': 'string'
}
},
'suggestion-id': {
'text': 1234567,
'term': {
'field': 'id'
}
}
}
{
'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
}
],
}
python setup.py test