Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@elasticmock
def setUp(self):
self.es = elasticsearch.Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
@elasticmock
def test_should_return_same_elastic_instance_when_instantiate_more_than_one_instance_with_same_host(self):
es1 = elasticsearch.Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
es2 = elasticsearch.Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
self.assertEqual(es1, es2)
@elasticmock
def test_should_return_same_elastic_instance_when_instantiate_more_than_one_instance_with_same_host(self):
es1 = elasticsearch.Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
es2 = elasticsearch.Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
self.assertEqual(es1, es2)
@elasticmock
def setUp(self):
self.es = elasticsearch.Elasticsearch(hosts=[{'host': 'localhost', 'port': 9200}])
self.index_name = 'test_index'
self.doc_type = 'doc-Type'
self.body = {
'author': 'kimchy',
'text': 'Elasticsearch: cool. bonsai cool.',
'timestamp': datetime.now(),
}
self.updated_body = {'string': 'content-updated', 'id': 1}
@elasticmock
def test_should_return_suggestions(self):
self.es.index(index=self.index_name, doc_type=self.doc_type, body=self.body)
suggestion_body = {
'suggestion-string': {
'text': 'test_text',
'term': {
'field': 'string'
}
},
'suggestion-id': {
'text': 1234567,
'term': {
'field': 'id'
}
}
}
@elasticmock
def test_export(self):
log.ThugOpts.elasticsearch_logging = True
log.configuration_path = configuration_path
assert log.ThugOpts.elasticsearch_logging
with patch('elasticmock.FakeElasticsearch.indices', create=True):
elastic_search = ElasticSearch(thug.__version__)
response = elastic_search.export('sample-dir')
enabled = elastic_search.enabled
assert response
assert enabled
log.ThugOpts.elasticsearch_logging = False
log.configuration_path = thug.__configuration_path__
assert not log.ThugOpts.elasticsearch_logging
@elasticmock
def test_should_raise_notfounderror_when_nonindexed_id_is_used_for_suggest(self):
with self.assertRaises(NotFoundError):
self.es.suggest(body={}, index=self.index_name)