test_elasticsearch_driver.py 文件源码

def test_elasticsearch_running(es):
    i = 0
    while i < 5:
        try:
            es.ping()
            assert True
            return
        except ConnectionError:
            i += 1
            sleep(2)

    pytest.fail('Elasticsearch not running (failed to connect after {} tries)'
                .format(str(i)))