Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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
# flake8: noqa
import hashlib
import time
from six import text_type as unicode
from cubicweb.devtools import testlib
from cubicweb.cwconfig import CubicWebConfiguration
from cubes.elasticsearch.search_helpers import compose_search
from elasticsearch_dsl import Search
from cubes.elasticsearch.es import CUSTOM_ATTRIBUTES
CUSTOM_ATTRIBUTES['Blog'] = ('title',)
class ParentsSearchTC(testlib.CubicWebTC):
def setup_database(self):
super(ParentsSearchTC, 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://localhost:9200'
# TODO unique ID to avoid collision
self.config['index-name'] = 'unittest_index_name'
def test_parent_search(self):
# self.vid_validators['esearch'] = lambda: None
with self.admin_access.cnx() as cnx:
indexer = cnx.vreg['es'].select('indexer', cnx)
mappings = indexer.settings.setdefault('mappings', {})
mappings["BlogEntry"] = {'_parent':{"type":"Blog"}}
indexer.get_connection()
test_structure = {
u'A': [u'Paris ceci', u'Nantes', u'Toulouse'],
u'B': [u'Paris cela'],
u'C': [u'Paris autre', u'Paris plage']
}
for fa_title, facomp_contents in test_structure.items():
blog = cnx.create_entity('Blog',
title=fa_title)
for facomp_content in facomp_contents:
cnx.create_entity('BlogEntry',
entry_of=blog,
title=facomp_content,
content=facomp_content)
cnx.commit()
time.sleep(2) # TODO find a way to have synchronous operations in unittests
for query, number_of_results, first_result in (("Paris", 3, "C"),
("Nantes", 1, "A")):
search = compose_search(Search(),
query,
parents_for="BlogEntry",
fields=['_all'],
fuzzy=True)
import json
print json.dumps(search.to_dict())
self.assertEquals(len(search.execute()), number_of_results)
self.assertEquals(search.execute().to_dict()['hits']['hits'][0]['_source']['title'],
first_result)
def tearDown(self):
with self.admin_access.cnx() as cnx:
indexer = cnx.vreg['es'].select('indexer', cnx)
es = indexer.get_connection()
es.indices.delete(self.config['index-name'])
super(ParentsSearchTC, self).tearDown()
if __name__ == '__main__':
from logilab.common.testlib import unittest_main
unittest_main()