Newer
Older
# flake8: noqa
from __future__ import print_function
import sys
import unittest
from io import StringIO
from mock import patch
from elasticsearch_dsl.result import Response
from cubicweb.devtools import testlib
from cubicweb.cwconfig import CubicWebConfiguration
from cubes.elasticsearch import ccplugin
from cubes.elasticsearch.es import indexable_types, index_settings
class ExportElasticSearchTC(testlib.AutoPopulateTest):
# ignore ComputedRelations
ignored_relations = set(
('narrower_concept', 'hidden_label', 'preferred_label', 'alternative_label', ))
def setup_database(self):
super(ExportElasticSearchTC, self).setup_database()
self.orig_config_for = CubicWebConfiguration.config_for
config_for = lambda appid: self.config # noqa
CubicWebConfiguration.config_for = staticmethod(config_for)
self.config['elasticsearch-locations'] = 'http://10.1.1.1:9200'
self.config['index-name'] = 'unittest_index_name'
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def to_test_etypes(self):
with self.admin_access.repo_cnx() as cnx:
types = indexable_types(cnx.repo)
return types
def tearDown(self):
CubicWebConfiguration.config_for = self.orig_config_for
super(ExportElasticSearchTC, self).tearDown()
def test_indexable_types(self):
with self.admin_access.repo_cnx() as cnx:
self.assertNotEquals(
len(indexable_types(cnx.repo)),
0)
@patch('elasticsearch.client.Elasticsearch.index', unsafe=True)
@patch('elasticsearch.client.indices.IndicesClient.create', unsafe=True)
def test_ccplugin(self, create, index):
self.auto_populate(10)
index.reset_mock()
cmd = [self.appid, '--dry-run', 'yes']
sys.stdout = out = StringIO()
try:
ccplugin.IndexInES(None).main_run(cmd)
finally:
sys.stdout = sys.__stdout__
self.assertEquals('', out.getvalue())
create.assert_not_called()
index.assert_not_called()
cmd = [self.appid, '--dry-run', 'yes', '--debug', 'yes']
sys.stdout = out = StringIO()
try:
ccplugin.IndexInES(None).main_run(cmd)
finally:
sys.stdout = sys.__stdout__
self.assert_('found ' in out.getvalue())
create.assert_not_called()
index.assert_not_called()
# TODO try wrong option
# cmd = [self.appid, '--wrong-option', 'yes']
cmd = [self.appid]
sys.stdout = StringIO()
try:
ccplugin.IndexInES(None).main_run(cmd)
finally:
sys.stdout = sys.__stdout__
with self.admin_access.repo_cnx() as cnx:
self.assert_(cnx.execute('Any X WHERE X is %(etype)s' %
{'etype': indexable_types(cnx.repo)[0]}))
create.assert_called_with(
ignore=400, index='unittest_index_name', body=index_settings())
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
index.assert_called()
# TODO ? check called data
@patch('elasticsearch.client.Elasticsearch.index', unsafe=True)
def test_es_hooks_create(self, index):
with self.admin_access.cnx() as cnx:
cnx.create_entity('BlogEntry', title=u'Article about stuff',
content=u'content herer')
cnx.commit()
index.assert_called()
@patch('elasticsearch.client.Elasticsearch.index', unsafe=True)
def test_es_hooks_modify(self, index):
with self.admin_access.cnx() as cnx:
entity = cnx.create_entity('BlogEntry', title=u'Article about stuff',
content=u'content herer')
cnx.commit()
index.reset_mock()
entity.cw_set(title=u'Different title')
index.assert_called()
def mock_execute(*args, **kwargs):
result = {'_source': {'description': 'test',
'cwuri': 'http://example.org/123',
'eid': 123,
'title': 'test'},
'_type': 'BaseContent',
'_score': 1}
return Response({'hits': {'hits': [result],
'total': 1
}})
class ElasticSearchViewsTC(testlib.CubicWebTC):
@patch('elasticsearch_dsl.search.Search.execute', new=mock_execute)
def test_search_view(self):
with self.new_access('anon').web_request() as req:
# self._cw.form.get('search'))
self.view('esearch', req=req, template=None)
if __name__ == '__main__':
unittest.main()