Używam w pythonie elasticsearch-dsl do komunikacji z Elasticsearch'em.
Próbuję przetestować zapis dokumentu, bez używania rzeczywistego serwera Elasticsearch.
Kiedy uruchamiam pytest konsola zwraca poniższy błąd:
mock_client = <Mock id='66229920'>
def test_article_save(mock_client):
data = init_test_data()
> data.get('articleone').save(using='mock')
tddtest.py:70:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
tdd.py:34: in save
return super(Article, self).save(** kwargs)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = Article(id=1), using = 'mock', index = None, validate = True, skip_empty = True, kwargs = {}, es = <Mock id='66229920'>, doc_meta = {'id': 1}, meta = <Mock name='mock.index()' id='66396040'>
k = 'version'
def save(self, using=None, index=None, validate=True, skip_empty=True, **kwargs):
"""
Save the document into elasticsearch. If the document doesn't exist it
is created, it is overwritten otherwise. Returns ``True`` if this
operations resulted in new document being created.
:arg index: elasticsearch index to use, if the ``Document`` is
associated with an index this can be omitted.
:arg using: connection alias to use, defaults to ``'default'``
:arg validate: set to ``False`` to skip validating the document
:arg skip_empty: if set to ``False`` will cause empty values (``None``,
``[]``, ``{}``) to be left on the document. Those values will be
stripped out otherwise as they make no difference in elasticsearch.
Any additional keyword arguments will be passed to
``Elasticsearch.index`` unchanged.
:return operation result created/updated
"""
if validate:
self.full_clean()
es = self._get_connection(using)
# extract routing etc from meta
doc_meta = {
k: self.meta[k]
for k in DOC_META_FIELDS
if k in self.meta
}
# Optimistic concurrency control
if 'seq_no' in self.meta and 'primary_term' in self.meta:
doc_meta['if_seq_no'] = self.meta['seq_no']
doc_meta['if_primary_term'] = self.meta['primary_term']
doc_meta.update(kwargs)
meta = es.index(
index=self._get_index(index),
body=self.to_dict(skip_empty=skip_empty),
**doc_meta
)
# update meta information from ES
for k in META_FIELDS:
> if '_' + k in meta:
E TypeError: argument of type 'Mock' is not iterable
..\venv\lib\site-packages\elasticsearch_dsl\document.py:460: TypeError
======================================================================================= short test summary info ========================================================================================
FAILED tddtest.py::test_article_save - TypeError: argument of type 'Mock' is not iterable
Moje pliki
tdd.py
from copy import deepcopy
from datetime import datetime
from pytz import timezone
from elasticsearch_dsl import Text, Keyword, Date, Integer, Document
from elasticsearch_dsl.exceptions import IllegalOperation
from elasticsearch_dsl.connections import connections, add_connection
from pytest import fixture
class Article(Document):
title = Text(analyzer='snowball', fields={'raw': Keyword()})
author = Text(analyzer='snowball')
body = Text(analyzer='snowball')
tags = Keyword()
published_from = Date()
lines = Integer()
def to_dict(self, include_meta=False, skip_empty=True):
self.title = ' '.join(self.title.split()) if self.title else self.title
self.author = ' '.join(self.author.split()) if self.author else self.author
self.body = ' '.join(self.body.strip().split(' ')) if self.body else self.body
self.published_from = self.published_from.isoformat().replace('+00:00', 'Z') if self.published_from else self.published_from
self.lines = len(self.body.strip().split('\n')) if self.body else 0
return super(Article, self).to_dict(include_meta, skip_empty)
class Index:
name = 'blog'
settings = {
'number_of_shards': 2,
}
def save(self, ** kwargs):
return super(Article, self).save(** kwargs)
def is_published(self):
return datetime.now() >= self.published_from
conftest.py
import os
from elasticsearch_dsl.connections import connections, add_connection
from pytest import fixture
from unittest import mock
#from mock import Mock
@fixture
def write_client(client):
yield client
client.indices.delete('test-*', ignore=404)
client.indices.delete_template('test-template', ignore=404)
@fixture
def mock_client(dummy_response):
client = mock.Mock()
client.search.return_value = dummy_response
add_connection('mock', client)
yield client
connections._conn = {}
connections._kwargs = {}
@fixture
def dummy_response():
return {
'_shards' : {
'failed': 0,
'successful': 10,
'total': 10
},
'hits': {
'hits': [
{
'_index': 'test-index',
'_type': '_doc',
'_id': '1',
'_score': '10.114',
'_source': {
'title': 'Test elasticsearch',
'body':
'''
Lorem Ipsum
''',
'published_from': '2013-02-10T10:31:07.851688',
'tags': [
'g1',
'g2'
],
'lines': '1'
},
'highlight': {
'title': ['<em>Test</em> elasticsearch']
}
},
{
'_index': 'test-index',
'_type': '_doc',
'_id': '2',
'_score': '12.0',
'_source': {
'title': 'Test elasticsearch numer 2',
'body':
'''
Lorem Ipsum
''',
'published_from': '2014-02-10T10:31:07.851688',
'tags': [
'g1',
'g2'
],
'lines': '1'
},
'highlight': {
'title': ['<em>Test</em> elasticsearch numer 2']
}
},
]
},
"timed_out": False,
"took": 123
}
tddtest.py
from datetime import datetime
from pytz import timezone
from elasticsearch_dsl.connections import connections, add_connection
from pytest import fixture
from tdd import Article
def init_test_data():
date3may = datetime(2020,5,3,23,59,59)
zero_timezone = timezone('Etc/UTC')
warsaw_timezone = timezone('Europe/Warsaw')
zero_localized_date3may = zero_timezone.localize(date3may)
warsaw_localized_date3may = warsaw_timezone.localize(date3may)
zero_string3may = '2020-05-03T23:59:59Z'
warsaw_zero_string3may = '2020-05-03T23:59:59+02:00'
test_article_one = Article(
meta={'id': 1},
title=
'''
Titleone one
''',
body=
'''
Id: 1;
Title: Titleone one;
Published_from_time: 00;
tags: ['g1', 'g2']
''',
published_from=zero_localized_date3may,
tags=['g1', 'g2']
)
test_article_two = Article(
meta={'id': 2},
title=
'''
Titletwo two
''',
body=
'''
Id: 2;
Title: Titletwo two;
Published_from_time: +01;
tags: ['g1', 'g2']
''',
published_from=warsaw_localized_date3may,
tags=['g1', 'g2']
)
return {"articleone": test_article_one, "articletwo": test_article_two, "timezero": zero_string3may, "timewarsaw": warsaw_zero_string3may}
def test_article():
data = init_test_data()
aone = data.get('articleone').to_dict(include_meta=True)
atwo = data.get('articletwo').to_dict(include_meta=True)
assert aone['_id'] == 1
assert aone['_source']['title'] == '''Titleone one'''
assert aone['_source']['body'] == '''Id: 1;\nTitle: Titleone one;\nPublished_from_time: 00;\ntags: ['g1', 'g2']'''
assert aone['_source']['published_from'] == data.get('timezero')
assert atwo['_source']['published_from'] == data.get('timewarsaw')
assert atwo['_source']['tags'][-1] == 'g2'
assert atwo['_source']['lines'] == 4
def test_article_save(mock_client):
data = init_test_data()
data.get('articleone').save(using='mock')
assert 1 != 2
requirements.txt
wheel==0.34.2
pytz==2020.1
rfc3339==6.2
elasticsearch-dsl==7.2.0
elasticsearch==7.7.0
pytest-mock==3.1.0
pytest-cov==2.9.0
pytest==5.4.2
mock==4.0.2