我正在尝试模拟Elasticsearch数据以用于托管CI单元测试。
我准备了一些可以成功地使用bulk()
加载的装置,但是由于未知原因,我无法将与任何相匹配,即使test_index
看似包含数据(因为我可以通过其ID对get()
项进行编码)。fixtures.json
是我从实际生产索引中获取的ES文档的子集。有了真实世界的索引,一切都会按预期工作,并且所有测试都通过了。
下面是一个虚假行为的人工例子:
class MyTestCase(TestCase):
es = Elasticsearch()
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.es.indices.create('test_index', SOME_SCHEMA)
with open('fixtures.json') as fixtures:
bulk(cls.es, json.load(fixtures))
@classmethod
def tearDownClass(cls):
super().tearDownClass()
cls.es.indices.delete('test_index')
def test_something(self):
# check all documents are there:
with open('fixtures.json') as fixtures:
for f in json.load(fixtures):
print(self.es.get(index='test_index', id=f['_id']))
# yes they are!
# BUT:
match_all = {"query": {"match_all": {}}}
print('hits:', self.es.search(index='test_index', body=match_all)['hits']['hits'])
# prints `hits: []` like there was nothing in
print('count:', self.es.count(index='test_index', body=match_all)['count'])
# prints `count: 0`
最佳答案
尽管我可以完全理解您的痛苦(除测试外,其他所有方法都可以),但答案实际上很简单:与您的实验相比,测试太快了。
索引文档与将文档索引之间的延迟最多为1秒
可搜索的。
创建索引和准备索引之间的开销)。
因此,修复方法是
time.sleep()
,为ES提供一些空间来创建需要给您结果的所有魔术。我会这样做:@classmethod
def setUpClass(cls):
super().setUpClass()
cls.es.indices.create('test_index', SOME_SCHEMA)
with open('fixtures.json') as fixtures:
bulk(cls.es, json.load(fixtures))
cls.wait_until_index_ready()
@classmethod
def wait_until_index_ready(cls, timeout=10):
for sec in range(timeout):
time.sleep(1)
if cls.es.cluster.health().get('status') in ('green', 'yellow'):
break
关于unit-testing - Elasticsearch “get by index”返回文档,而 “match_all”不返回结果,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/39492354/