test_espandas.py 文件源码

def test_es():
    """
    Before running other tests, ensure connection to ES is established
    """
    es = Elasticsearch()
    try:
        es.indices.create(INDEX)
        es.indices.delete(INDEX)
        return True
    except RequestError:
        print('Index already exists: skipping tests.')
        return False
    except ConnectionError:
        print('The ElasticSearch backend is not running: skipping tests.')
        return False
    except Exception as e:
        print('An unknown error occured connecting to ElasticSearch: %s' % e)
        return False