test_controllers.py 文件源码

def setUp(self):

        # Clean index
        self.es = Elasticsearch(hosts=[LOCAL_ELASTICSEARCH])
        try:
            self.es.indices.delete(index='datahub')
            self.es.indices.delete(index='events')
        except NotFoundError:
            pass
        self.es.indices.create('datahub')
        mapping = {'dataset': {'properties': self.MAPPING}}
        self.es.indices.put_mapping(doc_type='dataset',
                                    index='datahub',
                                    body=mapping)

        self.es.indices.create('events')
        mapping = {'event': {'properties': {'timestamp': {'type': 'date'}}}}
        self.es.indices.put_mapping(doc_type='event',
                                    index='events',
                                    body=mapping)