test_aggregations.py 文件源码

def test_date_range(app, es, event_queues, indexed_events):
    aggregate_events(['file-download-agg'])
    current_search_client.indices.refresh(index='*')

    query = Search(using=current_search_client,
                   index='stats-file-download')[0:30].sort('file_id')
    results = query.execute()

    total_count = 0
    for result in results:
        if 'file_id' in result:
            total_count += result.count
    assert total_count == 30