|
1 import os, unittest |
|
2 from django.contrib.gis.geos import * |
|
3 from django.contrib.gis.db.backend import SpatialBackend |
|
4 from django.contrib.gis.db.models import Collect, Count, Extent, F, Union |
|
5 from django.contrib.gis.tests.utils import no_mysql, no_oracle, no_spatialite |
|
6 from django.conf import settings |
|
7 from models import City, Location, DirectoryEntry, Parcel, Book, Author |
|
8 |
|
9 cities = (('Aurora', 'TX', -97.516111, 33.058333), |
|
10 ('Roswell', 'NM', -104.528056, 33.387222), |
|
11 ('Kecksburg', 'PA', -79.460734, 40.18476), |
|
12 ) |
|
13 |
|
14 class RelatedGeoModelTest(unittest.TestCase): |
|
15 |
|
16 def test01_setup(self): |
|
17 "Setting up for related model tests." |
|
18 for name, state, lon, lat in cities: |
|
19 loc = Location.objects.create(point=Point(lon, lat)) |
|
20 c = City.objects.create(name=name, state=state, location=loc) |
|
21 |
|
22 @no_oracle # TODO: Fix select_related() problems w/Oracle and pagination. |
|
23 def test02_select_related(self): |
|
24 "Testing `select_related` on geographic models (see #7126)." |
|
25 qs1 = City.objects.all() |
|
26 qs2 = City.objects.select_related() |
|
27 qs3 = City.objects.select_related('location') |
|
28 |
|
29 for qs in (qs1, qs2, qs3): |
|
30 for ref, c in zip(cities, qs): |
|
31 nm, st, lon, lat = ref |
|
32 self.assertEqual(nm, c.name) |
|
33 self.assertEqual(st, c.state) |
|
34 self.assertEqual(Point(lon, lat), c.location.point) |
|
35 |
|
36 @no_mysql |
|
37 @no_oracle # Pagination problem is implicated in this test as well. |
|
38 def test03_transform_related(self): |
|
39 "Testing the `transform` GeoQuerySet method on related geographic models." |
|
40 # All the transformations are to state plane coordinate systems using |
|
41 # US Survey Feet (thus a tolerance of 0 implies error w/in 1 survey foot). |
|
42 tol = 0 |
|
43 |
|
44 def check_pnt(ref, pnt): |
|
45 self.assertAlmostEqual(ref.x, pnt.x, tol) |
|
46 self.assertAlmostEqual(ref.y, pnt.y, tol) |
|
47 self.assertEqual(ref.srid, pnt.srid) |
|
48 |
|
49 # Each city transformed to the SRID of their state plane coordinate system. |
|
50 transformed = (('Kecksburg', 2272, 'POINT(1490553.98959621 314792.131023984)'), |
|
51 ('Roswell', 2257, 'POINT(481902.189077221 868477.766629735)'), |
|
52 ('Aurora', 2276, 'POINT(2269923.2484839 7069381.28722222)'), |
|
53 ) |
|
54 |
|
55 for name, srid, wkt in transformed: |
|
56 # Doing this implicitly sets `select_related` select the location. |
|
57 # TODO: Fix why this breaks on Oracle. |
|
58 qs = list(City.objects.filter(name=name).transform(srid, field_name='location__point')) |
|
59 check_pnt(GEOSGeometry(wkt, srid), qs[0].location.point) |
|
60 |
|
61 @no_mysql |
|
62 @no_spatialite |
|
63 def test04a_related_extent_aggregate(self): |
|
64 "Testing the `extent` GeoQuerySet aggregates on related geographic models." |
|
65 # This combines the Extent and Union aggregates into one query |
|
66 aggs = City.objects.aggregate(Extent('location__point')) |
|
67 |
|
68 # One for all locations, one that excludes Roswell. |
|
69 all_extent = (-104.528060913086, 33.0583305358887,-79.4607315063477, 40.1847610473633) |
|
70 txpa_extent = (-97.51611328125, 33.0583305358887,-79.4607315063477, 40.1847610473633) |
|
71 e1 = City.objects.extent(field_name='location__point') |
|
72 e2 = City.objects.exclude(name='Roswell').extent(field_name='location__point') |
|
73 e3 = aggs['location__point__extent'] |
|
74 |
|
75 # The tolerance value is to four decimal places because of differences |
|
76 # between the Oracle and PostGIS spatial backends on the extent calculation. |
|
77 tol = 4 |
|
78 for ref, e in [(all_extent, e1), (txpa_extent, e2), (all_extent, e3)]: |
|
79 for ref_val, e_val in zip(ref, e): self.assertAlmostEqual(ref_val, e_val, tol) |
|
80 |
|
81 @no_mysql |
|
82 def test04b_related_union_aggregate(self): |
|
83 "Testing the `unionagg` GeoQuerySet aggregates on related geographic models." |
|
84 # This combines the Extent and Union aggregates into one query |
|
85 aggs = City.objects.aggregate(Union('location__point')) |
|
86 |
|
87 # These are the points that are components of the aggregate geographic |
|
88 # union that is returned. |
|
89 p1 = Point(-104.528056, 33.387222) |
|
90 p2 = Point(-97.516111, 33.058333) |
|
91 p3 = Point(-79.460734, 40.18476) |
|
92 |
|
93 # Creating the reference union geometry depending on the spatial backend, |
|
94 # as Oracle will have a different internal ordering of the component |
|
95 # geometries than PostGIS. The second union aggregate is for a union |
|
96 # query that includes limiting information in the WHERE clause (in other |
|
97 # words a `.filter()` precedes the call to `.unionagg()`). |
|
98 if SpatialBackend.oracle: |
|
99 ref_u1 = MultiPoint(p3, p1, p2, srid=4326) |
|
100 ref_u2 = MultiPoint(p3, p2, srid=4326) |
|
101 else: |
|
102 ref_u1 = MultiPoint(p1, p2, p3, srid=4326) |
|
103 ref_u2 = MultiPoint(p2, p3, srid=4326) |
|
104 |
|
105 u1 = City.objects.unionagg(field_name='location__point') |
|
106 u2 = City.objects.exclude(name='Roswell').unionagg(field_name='location__point') |
|
107 u3 = aggs['location__point__union'] |
|
108 |
|
109 self.assertEqual(ref_u1, u1) |
|
110 self.assertEqual(ref_u2, u2) |
|
111 self.assertEqual(ref_u1, u3) |
|
112 |
|
113 def test05_select_related_fk_to_subclass(self): |
|
114 "Testing that calling select_related on a query over a model with an FK to a model subclass works" |
|
115 # Regression test for #9752. |
|
116 l = list(DirectoryEntry.objects.all().select_related()) |
|
117 |
|
118 def test06_f_expressions(self): |
|
119 "Testing F() expressions on GeometryFields." |
|
120 # Constructing a dummy parcel border and getting the City instance for |
|
121 # assigning the FK. |
|
122 b1 = GEOSGeometry('POLYGON((-97.501205 33.052520,-97.501205 33.052576,-97.501150 33.052576,-97.501150 33.052520,-97.501205 33.052520))', srid=4326) |
|
123 pcity = City.objects.get(name='Aurora') |
|
124 |
|
125 # First parcel has incorrect center point that is equal to the City; |
|
126 # it also has a second border that is different from the first as a |
|
127 # 100ft buffer around the City. |
|
128 c1 = pcity.location.point |
|
129 c2 = c1.transform(2276, clone=True) |
|
130 b2 = c2.buffer(100) |
|
131 p1 = Parcel.objects.create(name='P1', city=pcity, center1=c1, center2=c2, border1=b1, border2=b2) |
|
132 |
|
133 # Now creating a second Parcel where the borders are the same, just |
|
134 # in different coordinate systems. The center points are also the |
|
135 # the same (but in different coordinate systems), and this time they |
|
136 # actually correspond to the centroid of the border. |
|
137 c1 = b1.centroid |
|
138 c2 = c1.transform(2276, clone=True) |
|
139 p2 = Parcel.objects.create(name='P2', city=pcity, center1=c1, center2=c2, border1=b1, border2=b1) |
|
140 |
|
141 # Should return the second Parcel, which has the center within the |
|
142 # border. |
|
143 qs = Parcel.objects.filter(center1__within=F('border1')) |
|
144 self.assertEqual(1, len(qs)) |
|
145 self.assertEqual('P2', qs[0].name) |
|
146 |
|
147 if not SpatialBackend.mysql: |
|
148 # This time center2 is in a different coordinate system and needs |
|
149 # to be wrapped in transformation SQL. |
|
150 qs = Parcel.objects.filter(center2__within=F('border1')) |
|
151 self.assertEqual(1, len(qs)) |
|
152 self.assertEqual('P2', qs[0].name) |
|
153 |
|
154 # Should return the first Parcel, which has the center point equal |
|
155 # to the point in the City ForeignKey. |
|
156 qs = Parcel.objects.filter(center1=F('city__location__point')) |
|
157 self.assertEqual(1, len(qs)) |
|
158 self.assertEqual('P1', qs[0].name) |
|
159 |
|
160 if not SpatialBackend.mysql: |
|
161 # This time the city column should be wrapped in transformation SQL. |
|
162 qs = Parcel.objects.filter(border2__contains=F('city__location__point')) |
|
163 self.assertEqual(1, len(qs)) |
|
164 self.assertEqual('P1', qs[0].name) |
|
165 |
|
166 def test07_values(self): |
|
167 "Testing values() and values_list() and GeoQuerySets." |
|
168 # GeoQuerySet and GeoValuesQuerySet, and GeoValuesListQuerySet respectively. |
|
169 gqs = Location.objects.all() |
|
170 gvqs = Location.objects.values() |
|
171 gvlqs = Location.objects.values_list() |
|
172 |
|
173 # Incrementing through each of the models, dictionaries, and tuples |
|
174 # returned by the different types of GeoQuerySets. |
|
175 for m, d, t in zip(gqs, gvqs, gvlqs): |
|
176 # The values should be Geometry objects and not raw strings returned |
|
177 # by the spatial database. |
|
178 self.failUnless(isinstance(d['point'], SpatialBackend.Geometry)) |
|
179 self.failUnless(isinstance(t[1], SpatialBackend.Geometry)) |
|
180 self.assertEqual(m.point, d['point']) |
|
181 self.assertEqual(m.point, t[1]) |
|
182 |
|
183 def test08_defer_only(self): |
|
184 "Testing defer() and only() on Geographic models." |
|
185 qs = Location.objects.all() |
|
186 def_qs = Location.objects.defer('point') |
|
187 for loc, def_loc in zip(qs, def_qs): |
|
188 self.assertEqual(loc.point, def_loc.point) |
|
189 |
|
190 def test09_pk_relations(self): |
|
191 "Ensuring correct primary key column is selected across relations. See #10757." |
|
192 # Adding two more cities, but this time making sure that their location |
|
193 # ID values do not match their City ID values. |
|
194 loc1 = Location.objects.create(point='POINT (-95.363151 29.763374)') |
|
195 loc2 = Location.objects.create(point='POINT (-96.801611 32.782057)') |
|
196 dallas = City.objects.create(name='Dallas', state='TX', location=loc2) |
|
197 houston = City.objects.create(name='Houston', state='TX', location=loc1) |
|
198 |
|
199 # The expected ID values -- notice the last two location IDs |
|
200 # are out of order. We want to make sure that the related |
|
201 # location ID column is selected instead of ID column for |
|
202 # the city. |
|
203 city_ids = (1, 2, 3, 4, 5) |
|
204 loc_ids = (1, 2, 3, 5, 4) |
|
205 ids_qs = City.objects.order_by('id').values('id', 'location__id') |
|
206 for val_dict, c_id, l_id in zip(ids_qs, city_ids, loc_ids): |
|
207 self.assertEqual(val_dict['id'], c_id) |
|
208 self.assertEqual(val_dict['location__id'], l_id) |
|
209 |
|
210 def test10_combine(self): |
|
211 "Testing the combination of two GeoQuerySets. See #10807." |
|
212 buf1 = City.objects.get(name='Aurora').location.point.buffer(0.1) |
|
213 buf2 = City.objects.get(name='Kecksburg').location.point.buffer(0.1) |
|
214 qs1 = City.objects.filter(location__point__within=buf1) |
|
215 qs2 = City.objects.filter(location__point__within=buf2) |
|
216 combined = qs1 | qs2 |
|
217 names = [c.name for c in combined] |
|
218 self.assertEqual(2, len(names)) |
|
219 self.failUnless('Aurora' in names) |
|
220 self.failUnless('Kecksburg' in names) |
|
221 |
|
222 def test11_geoquery_pickle(self): |
|
223 "Ensuring GeoQuery objects are unpickled correctly. See #10839." |
|
224 import pickle |
|
225 from django.contrib.gis.db.models.sql import GeoQuery |
|
226 qs = City.objects.all() |
|
227 q_str = pickle.dumps(qs.query) |
|
228 q = pickle.loads(q_str) |
|
229 self.assertEqual(GeoQuery, q.__class__) |
|
230 |
|
231 # TODO: fix on Oracle -- get the following error because the SQL is ordered |
|
232 # by a geometry object, which Oracle apparently doesn't like: |
|
233 # ORA-22901: cannot compare nested table or VARRAY or LOB attributes of an object type |
|
234 @no_oracle |
|
235 def test12a_count(self): |
|
236 "Testing `Count` aggregate use with the `GeoManager` on geo-fields." |
|
237 # Creating a new City, 'Fort Worth', that uses the same location |
|
238 # as Dallas. |
|
239 dallas = City.objects.get(name='Dallas') |
|
240 ftworth = City.objects.create(name='Fort Worth', state='TX', location=dallas.location) |
|
241 |
|
242 # Count annotation should be 2 for the Dallas location now. |
|
243 loc = Location.objects.annotate(num_cities=Count('city')).get(id=dallas.location.id) |
|
244 self.assertEqual(2, loc.num_cities) |
|
245 |
|
246 def test12b_count(self): |
|
247 "Testing `Count` aggregate use with the `GeoManager` on non geo-fields. See #11087." |
|
248 # Creating some data for the Book/Author non-geo models that |
|
249 # use GeoManager. See #11087. |
|
250 tp = Author.objects.create(name='Trevor Paglen') |
|
251 Book.objects.create(title='Torture Taxi', author=tp) |
|
252 Book.objects.create(title='I Could Tell You But Then You Would Have to be Destroyed by Me', author=tp) |
|
253 Book.objects.create(title='Blank Spots on the Map', author=tp) |
|
254 wp = Author.objects.create(name='William Patry') |
|
255 Book.objects.create(title='Patry on Copyright', author=wp) |
|
256 |
|
257 # Should only be one author (Trevor Paglen) returned by this query, and |
|
258 # the annotation should have 3 for the number of books. Also testing |
|
259 # with a `GeoValuesQuerySet` (see #11489). |
|
260 qs = Author.objects.annotate(num_books=Count('books')).filter(num_books__gt=1) |
|
261 vqs = Author.objects.values('name').annotate(num_books=Count('books')).filter(num_books__gt=1) |
|
262 self.assertEqual(1, len(qs)) |
|
263 self.assertEqual(3, qs[0].num_books) |
|
264 self.assertEqual(1, len(vqs)) |
|
265 self.assertEqual(3, vqs[0]['num_books']) |
|
266 |
|
267 # TODO: The phantom model does appear on Oracle. |
|
268 @no_oracle |
|
269 def test13_select_related_null_fk(self): |
|
270 "Testing `select_related` on a nullable ForeignKey via `GeoManager`. See #11381." |
|
271 no_author = Book.objects.create(title='Without Author') |
|
272 b = Book.objects.select_related('author').get(title='Without Author') |
|
273 # Should be `None`, and not a 'dummy' model. |
|
274 self.assertEqual(None, b.author) |
|
275 |
|
276 @no_mysql |
|
277 @no_oracle |
|
278 @no_spatialite |
|
279 def test14_collect(self): |
|
280 "Testing the `collect` GeoQuerySet method and `Collect` aggregate." |
|
281 # Reference query: |
|
282 # SELECT AsText(ST_Collect("relatedapp_location"."point")) FROM "relatedapp_city" LEFT OUTER JOIN |
|
283 # "relatedapp_location" ON ("relatedapp_city"."location_id" = "relatedapp_location"."id") |
|
284 # WHERE "relatedapp_city"."state" = 'TX'; |
|
285 ref_geom = fromstr('MULTIPOINT(-97.516111 33.058333,-96.801611 32.782057,-95.363151 29.763374,-96.801611 32.782057)') |
|
286 |
|
287 c1 = City.objects.filter(state='TX').collect(field_name='location__point') |
|
288 c2 = City.objects.filter(state='TX').aggregate(Collect('location__point'))['location__point__collect'] |
|
289 |
|
290 for coll in (c1, c2): |
|
291 # Even though Dallas and Ft. Worth share same point, Collect doesn't |
|
292 # consolidate -- that's why 4 points in MultiPoint. |
|
293 self.assertEqual(4, len(coll)) |
|
294 self.assertEqual(ref_geom, coll) |
|
295 |
|
296 # TODO: Related tests for KML, GML, and distance lookups. |
|
297 |
|
298 def suite(): |
|
299 s = unittest.TestSuite() |
|
300 s.addTest(unittest.makeSuite(RelatedGeoModelTest)) |
|
301 return s |