python类VERSION的实例源码

def _build_search_filters_dwithin(self, dwithin):
        lng, lat = dwithin['point'].get_coords()
        # NB: the 1.0.0 release of elasticsearch introduce an
        #     incompatible change on the distance filter formating
        if elasticsearch.VERSION >= (1, 0, 0):
            distance = "%(dist).6f%(unit)s" % {
                'dist': dwithin['distance'].km,
                'unit': "km"
            }
        else:
            distance = dwithin['distance'].km
        dwithin_filter = {
            "geo_distance": {
                "distance": distance,
                dwithin['field']: {
                    "lat": lat,
                    "lon": lng
                }
            }
        }
        return dwithin_filter

def setup(self):
        """
        Defers loading until needed.
        """
        # Get the existing mapping & cache it. We'll compare it
        # during the ``update`` & if it doesn't match, we'll put the new
        # mapping.
        try:
            self.existing_mapping = self.conn.indices.get_mapping(index=self.index_name)
        except NotFoundError:
            pass
        except Exception:
            if not self.silently_fail:
                raise

        unified_index = haystack.connections[self.connection_alias].get_unified_index()
        self.content_field_name, field_mapping = self.build_schema(unified_index.all_searchfields())
        # fixing ES 1.x/2.x compatible `_boost`
        current_mapping = {
            'modelresult': {
                'properties': field_mapping,
            }
        }
        if elasticsearch.VERSION < (2, 0, 0):
            current_mapping['modelresult']['_boost'] = {
                'name': 'boost',
                'null_value': 1.0
            }
        # end fixing ES 1.x/2.x compatible `_boost`

        if current_mapping != self.existing_mapping:
            try:
                # Make sure the index is there first.
                self.conn.indices.create(index=self.index_name, body=self.DEFAULT_SETTINGS, ignore=400)
                self.conn.indices.put_mapping(index=self.index_name, doc_type='modelresult', body=current_mapping)
                self.existing_mapping = current_mapping
            except Exception:
                if not self.silently_fail:
                    raise

        self.setup_complete = True

def setUp(self):
        super(ElasticsearchSearchQuerySpatialBeforeReleaseTestCase, self).setUp()
        self.backend = connections['default'].get_backend()
        self._elasticsearch_version = elasticsearch.VERSION
        elasticsearch.VERSION = (0,9,9)

def tearDown(self):
        elasticsearch.VERSION = self._elasticsearch_version

def setUp(self):
        super(ElasticsearchSearchQuerySpatialAfterReleaseTestCase, self).setUp()
        self.backend = connections['default'].get_backend()
        self._elasticsearch_version = elasticsearch.VERSION
        elasticsearch.VERSION = (1,0,0)

def tearDown(self):
        elasticsearch.VERSION = self._elasticsearch_version

def log_request_fail(self, method, full_url, path, body, duration,
                         status_code=None, response=None, exception=None):
        if es.VERSION < (5, 0):
            super().log_request_fail(method, full_url, body, duration,
                                     status_code, response, exception)
        else:
            super().log_request_fail(method, full_url, path, body, duration,
                                     status_code, response, exception)

def test_query__in_empty_list(self):
        """Confirm that an empty list avoids a Elasticsearch exception"""
        sqs = SearchQuerySet(using='default').filter(id__in=[])
        self.assertEqual(sqs.query.build_query(), u'id:(!*:*)')


# class ElasticsearchSearchQuerySpatialBeforeReleaseTestCase(TestCase):
#     def setUp(self):
#         super(ElasticsearchSearchQuerySpatialBeforeReleaseTestCase, self).setUp()
#         self.backend = connections['default'].get_backend()
#         self._elasticsearch_version = elasticsearch.VERSION
#         elasticsearch.VERSION = (0,9,9)

#     def tearDown(self):
#         elasticsearch.VERSION = self._elasticsearch_version

#     def test_build_query_with_dwithin_range(self):
#         """
#         Test build_search_kwargs with dwithin range for Elasticsearch versions < 1.0.0
#         """
#         search_kwargs = self.backend.build_search_kwargs('where', dwithin={
#             'field': "location_field",
#             'point': Point(1.2345678, 2.3456789),
#             'distance': D(m=500)
#         })
#         self.assertEqual(search_kwargs['query']['filtered']['filter']['bool']['must'][1]['geo_distance'], {'distance': 0.5, 'location_field': {'lat': 2.3456789, 'lon': 1.2345678}})



# class ElasticsearchSearchQuerySpatialAfterReleaseTestCase(TestCase):
#     def setUp(self):
#         super(ElasticsearchSearchQuerySpatialAfterReleaseTestCase, self).setUp()
#         self.backend = connections['elasticsearch'].get_backend()
#         self._elasticsearch_version = elasticsearch.VERSION
#         elasticsearch.VERSION = (1,0,0)

#     def tearDown(self):
#         elasticsearch.VERSION = self._elasticsearch_version

#     def test_build_query_with_dwithin_range(self):
#         """
#         Test build_search_kwargs with dwithin range for Elasticsearch versions >= 1.0.0
#         """
#         search_kwargs = self.backend.build_search_kwargs('where', dwithin={
#             'field': "location_field",
#             'point': Point(1.2345678, 2.3456789),
#             'distance': D(m=500)
#         })
#         self.assertEqual(search_kwargs['query']['filtered']['filter']['bool']['must'][1]['geo_distance'], {'distance': "0.500000km", 'location_field': {'lat': 2.3456789, 'lon': 1.2345678}})