test_collection.py 文件源码

def test_index_2dsphere(self, test_db):
        assert 'geo_2dsphere' == await test_db.test.create_index([('geo', GEOSPHERE)])

        for dummy, info in (await test_db.test.index_information()).items():
            field, idx_type = info['key'][0]
            if field == 'geo' and idx_type == '2dsphere':
                break
        else:
            pytest.fail('2dsphere index not found.')

        poly = {'type': 'Polygon',
                'coordinates': [[[40, 5], [40, 6], [41, 6], [41, 5], [40, 5]]]}
        query = {'geo': {'$within': {'$geometry': poly}}}

        # This query will error without a 2dsphere index.
        async with test_db.test.find(query) as cursor:
            async for _ in cursor:
                pass