388 lines
16 KiB
Python
388 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
import unittest
|
|
|
|
from mongoengine import *
|
|
from mongoengine.connection import get_db
|
|
|
|
__all__ = ("GeoFieldTest", )
|
|
|
|
|
|
class GeoFieldTest(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
connect(db='mongoenginetest')
|
|
self.db = get_db()
|
|
|
|
def _test_for_expected_error(self, Cls, loc, expected):
|
|
try:
|
|
Cls(loc=loc).validate()
|
|
self.fail('Should not validate the location {0}'.format(loc))
|
|
except ValidationError as e:
|
|
self.assertEqual(expected, e.to_dict()['loc'])
|
|
|
|
def test_geopoint_validation(self):
|
|
class Location(Document):
|
|
loc = GeoPointField()
|
|
|
|
invalid_coords = [{"x": 1, "y": 2}, 5, "a"]
|
|
expected = 'GeoPointField can only accept tuples or lists of (x, y)'
|
|
|
|
for coord in invalid_coords:
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
invalid_coords = [[], [1], [1, 2, 3]]
|
|
for coord in invalid_coords:
|
|
expected = "Value (%s) must be a two-dimensional point" % repr(coord)
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
invalid_coords = [[{}, {}], ("a", "b")]
|
|
for coord in invalid_coords:
|
|
expected = "Both values (%s) in point must be float or int" % repr(coord)
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
def test_point_validation(self):
|
|
class Location(Document):
|
|
loc = PointField()
|
|
|
|
invalid_coords = {"x": 1, "y": 2}
|
|
expected = 'PointField can only accept a valid GeoJson dictionary or lists of (x, y)'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "MadeUp", "coordinates": []}
|
|
expected = 'PointField type must be "Point"'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "Point", "coordinates": [1, 2, 3]}
|
|
expected = "Value ([1, 2, 3]) must be a two-dimensional point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [5, "a"]
|
|
expected = "PointField can only accept lists of [x, y]"
|
|
for coord in invalid_coords:
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
invalid_coords = [[], [1], [1, 2, 3]]
|
|
for coord in invalid_coords:
|
|
expected = "Value (%s) must be a two-dimensional point" % repr(coord)
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
invalid_coords = [[{}, {}], ("a", "b")]
|
|
for coord in invalid_coords:
|
|
expected = "Both values (%s) in point must be float or int" % repr(coord)
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
Location(loc=[1, 2]).validate()
|
|
Location(loc={
|
|
"type": "Point",
|
|
"coordinates": [
|
|
81.4471435546875,
|
|
23.61432859499169
|
|
]}).validate()
|
|
|
|
def test_linestring_validation(self):
|
|
class Location(Document):
|
|
loc = LineStringField()
|
|
|
|
invalid_coords = {"x": 1, "y": 2}
|
|
expected = 'LineStringField can only accept a valid GeoJson dictionary or lists of (x, y)'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "MadeUp", "coordinates": [[]]}
|
|
expected = 'LineStringField type must be "LineString"'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "LineString", "coordinates": [[1, 2, 3]]}
|
|
expected = "Invalid LineString:\nValue ([1, 2, 3]) must be a two-dimensional point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [5, "a"]
|
|
expected = "Invalid LineString must contain at least one valid point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[1]]
|
|
expected = "Invalid LineString:\nValue (%s) must be a two-dimensional point" % repr(invalid_coords[0])
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[1, 2, 3]]
|
|
expected = "Invalid LineString:\nValue (%s) must be a two-dimensional point" % repr(invalid_coords[0])
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[{}, {}]], [("a", "b")]]
|
|
for coord in invalid_coords:
|
|
expected = "Invalid LineString:\nBoth values (%s) in point must be float or int" % repr(coord[0])
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
Location(loc=[[1, 2], [3, 4], [5, 6], [1, 2]]).validate()
|
|
|
|
def test_polygon_validation(self):
|
|
class Location(Document):
|
|
loc = PolygonField()
|
|
|
|
invalid_coords = {"x": 1, "y": 2}
|
|
expected = 'PolygonField can only accept a valid GeoJson dictionary or lists of (x, y)'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "MadeUp", "coordinates": [[]]}
|
|
expected = 'PolygonField type must be "Polygon"'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "Polygon", "coordinates": [[[1, 2, 3]]]}
|
|
expected = "Invalid Polygon:\nValue ([1, 2, 3]) must be a two-dimensional point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[5, "a"]]]
|
|
expected = "Invalid Polygon:\nBoth values ([5, 'a']) in point must be float or int"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[]]]
|
|
expected = "Invalid Polygon must contain at least one valid linestring"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[1, 2, 3]]]
|
|
expected = "Invalid Polygon:\nValue ([1, 2, 3]) must be a two-dimensional point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[{}, {}]], [("a", "b")]]
|
|
expected = "Invalid Polygon:\nBoth values ([{}, {}]) in point must be float or int, Both values (('a', 'b')) in point must be float or int"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[1, 2], [3, 4]]]
|
|
expected = "Invalid Polygon:\nLineStrings must start and end at the same point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
Location(loc=[[[1, 2], [3, 4], [5, 6], [1, 2]]]).validate()
|
|
|
|
def test_multipoint_validation(self):
|
|
class Location(Document):
|
|
loc = MultiPointField()
|
|
|
|
invalid_coords = {"x": 1, "y": 2}
|
|
expected = 'MultiPointField can only accept a valid GeoJson dictionary or lists of (x, y)'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "MadeUp", "coordinates": [[]]}
|
|
expected = 'MultiPointField type must be "MultiPoint"'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "MultiPoint", "coordinates": [[1, 2, 3]]}
|
|
expected = "Value ([1, 2, 3]) must be a two-dimensional point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[]]
|
|
expected = "Invalid MultiPoint must contain at least one valid point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[1]], [[1, 2, 3]]]
|
|
for coord in invalid_coords:
|
|
expected = "Value (%s) must be a two-dimensional point" % repr(coord[0])
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
invalid_coords = [[[{}, {}]], [("a", "b")]]
|
|
for coord in invalid_coords:
|
|
expected = "Both values (%s) in point must be float or int" % repr(coord[0])
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
Location(loc=[[1, 2]]).validate()
|
|
Location(loc={
|
|
"type": "MultiPoint",
|
|
"coordinates": [
|
|
[1, 2],
|
|
[81.4471435546875, 23.61432859499169]
|
|
]}).validate()
|
|
|
|
def test_multilinestring_validation(self):
|
|
class Location(Document):
|
|
loc = MultiLineStringField()
|
|
|
|
invalid_coords = {"x": 1, "y": 2}
|
|
expected = 'MultiLineStringField can only accept a valid GeoJson dictionary or lists of (x, y)'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "MadeUp", "coordinates": [[]]}
|
|
expected = 'MultiLineStringField type must be "MultiLineString"'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "MultiLineString", "coordinates": [[[1, 2, 3]]]}
|
|
expected = "Invalid MultiLineString:\nValue ([1, 2, 3]) must be a two-dimensional point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [5, "a"]
|
|
expected = "Invalid MultiLineString must contain at least one valid linestring"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[1]]]
|
|
expected = "Invalid MultiLineString:\nValue (%s) must be a two-dimensional point" % repr(invalid_coords[0][0])
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[1, 2, 3]]]
|
|
expected = "Invalid MultiLineString:\nValue (%s) must be a two-dimensional point" % repr(invalid_coords[0][0])
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[[{}, {}]]], [[("a", "b")]]]
|
|
for coord in invalid_coords:
|
|
expected = "Invalid MultiLineString:\nBoth values (%s) in point must be float or int" % repr(coord[0][0])
|
|
self._test_for_expected_error(Location, coord, expected)
|
|
|
|
Location(loc=[[[1, 2], [3, 4], [5, 6], [1, 2]]]).validate()
|
|
|
|
def test_multipolygon_validation(self):
|
|
class Location(Document):
|
|
loc = MultiPolygonField()
|
|
|
|
invalid_coords = {"x": 1, "y": 2}
|
|
expected = 'MultiPolygonField can only accept a valid GeoJson dictionary or lists of (x, y)'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "MadeUp", "coordinates": [[]]}
|
|
expected = 'MultiPolygonField type must be "MultiPolygon"'
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = {"type": "MultiPolygon", "coordinates": [[[[1, 2, 3]]]]}
|
|
expected = "Invalid MultiPolygon:\nValue ([1, 2, 3]) must be a two-dimensional point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[[5, "a"]]]]
|
|
expected = "Invalid MultiPolygon:\nBoth values ([5, 'a']) in point must be float or int"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[[]]]]
|
|
expected = "Invalid MultiPolygon must contain at least one valid Polygon"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[[1, 2, 3]]]]
|
|
expected = "Invalid MultiPolygon:\nValue ([1, 2, 3]) must be a two-dimensional point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[[{}, {}]]], [[("a", "b")]]]
|
|
expected = "Invalid MultiPolygon:\nBoth values ([{}, {}]) in point must be float or int, Both values (('a', 'b')) in point must be float or int"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
invalid_coords = [[[[1, 2], [3, 4]]]]
|
|
expected = "Invalid MultiPolygon:\nLineStrings must start and end at the same point"
|
|
self._test_for_expected_error(Location, invalid_coords, expected)
|
|
|
|
Location(loc=[[[[1, 2], [3, 4], [5, 6], [1, 2]]]]).validate()
|
|
|
|
def test_indexes_geopoint(self):
|
|
"""Ensure that indexes are created automatically for GeoPointFields.
|
|
"""
|
|
class Event(Document):
|
|
title = StringField()
|
|
location = GeoPointField()
|
|
|
|
geo_indicies = Event._geo_indices()
|
|
self.assertEqual(geo_indicies, [{'fields': [('location', '2d')]}])
|
|
|
|
def test_geopoint_embedded_indexes(self):
|
|
"""Ensure that indexes are created automatically for GeoPointFields on
|
|
embedded documents.
|
|
"""
|
|
class Venue(EmbeddedDocument):
|
|
location = GeoPointField()
|
|
name = StringField()
|
|
|
|
class Event(Document):
|
|
title = StringField()
|
|
venue = EmbeddedDocumentField(Venue)
|
|
|
|
geo_indicies = Event._geo_indices()
|
|
self.assertEqual(geo_indicies, [{'fields': [('venue.location', '2d')]}])
|
|
|
|
def test_indexes_2dsphere(self):
|
|
"""Ensure that indexes are created automatically for GeoPointFields.
|
|
"""
|
|
class Event(Document):
|
|
title = StringField()
|
|
point = PointField()
|
|
line = LineStringField()
|
|
polygon = PolygonField()
|
|
|
|
geo_indicies = Event._geo_indices()
|
|
self.assertIn({'fields': [('line', '2dsphere')]}, geo_indicies)
|
|
self.assertIn({'fields': [('polygon', '2dsphere')]}, geo_indicies)
|
|
self.assertIn({'fields': [('point', '2dsphere')]}, geo_indicies)
|
|
|
|
def test_indexes_2dsphere_embedded(self):
|
|
"""Ensure that indexes are created automatically for GeoPointFields.
|
|
"""
|
|
class Venue(EmbeddedDocument):
|
|
name = StringField()
|
|
point = PointField()
|
|
line = LineStringField()
|
|
polygon = PolygonField()
|
|
|
|
class Event(Document):
|
|
title = StringField()
|
|
venue = EmbeddedDocumentField(Venue)
|
|
|
|
geo_indicies = Event._geo_indices()
|
|
self.assertIn({'fields': [('venue.line', '2dsphere')]}, geo_indicies)
|
|
self.assertIn({'fields': [('venue.polygon', '2dsphere')]}, geo_indicies)
|
|
self.assertIn({'fields': [('venue.point', '2dsphere')]}, geo_indicies)
|
|
|
|
def test_geo_indexes_recursion(self):
|
|
|
|
class Location(Document):
|
|
name = StringField()
|
|
location = GeoPointField()
|
|
|
|
class Parent(Document):
|
|
name = StringField()
|
|
location = ReferenceField(Location)
|
|
|
|
Location.drop_collection()
|
|
Parent.drop_collection()
|
|
|
|
Parent(name='Berlin').save()
|
|
info = Parent._get_collection().index_information()
|
|
self.assertNotIn('location_2d', info)
|
|
info = Location._get_collection().index_information()
|
|
self.assertIn('location_2d', info)
|
|
|
|
self.assertEqual(len(Parent._geo_indices()), 0)
|
|
self.assertEqual(len(Location._geo_indices()), 1)
|
|
|
|
def test_geo_indexes_auto_index(self):
|
|
|
|
# Test just listing the fields
|
|
class Log(Document):
|
|
location = PointField(auto_index=False)
|
|
datetime = DateTimeField()
|
|
|
|
meta = {
|
|
'indexes': [[("location", "2dsphere"), ("datetime", 1)]]
|
|
}
|
|
|
|
self.assertEqual([], Log._geo_indices())
|
|
|
|
Log.drop_collection()
|
|
Log.ensure_indexes()
|
|
|
|
info = Log._get_collection().index_information()
|
|
self.assertEqual(info["location_2dsphere_datetime_1"]["key"],
|
|
[('location', '2dsphere'), ('datetime', 1)])
|
|
|
|
# Test listing explicitly
|
|
class Log(Document):
|
|
location = PointField(auto_index=False)
|
|
datetime = DateTimeField()
|
|
|
|
meta = {
|
|
'indexes': [
|
|
{'fields': [("location", "2dsphere"), ("datetime", 1)]}
|
|
]
|
|
}
|
|
|
|
self.assertEqual([], Log._geo_indices())
|
|
|
|
Log.drop_collection()
|
|
Log.ensure_indexes()
|
|
|
|
info = Log._get_collection().index_information()
|
|
self.assertEqual(info["location_2dsphere_datetime_1"]["key"],
|
|
[('location', '2dsphere'), ('datetime', 1)])
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|