tests.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335
  1. # -*- coding: utf-8 -*-
  2. from __future__ import unicode_literals
  3. from copy import copy
  4. from decimal import Decimal
  5. import os
  6. import unittest
  7. from unittest import skipUnless
  8. from django.contrib.gis.gdal import HAS_GDAL
  9. from django.contrib.gis.tests.utils import HAS_SPATIAL_DB, mysql
  10. from django.db import router
  11. from django.conf import settings
  12. from django.test import TestCase
  13. from django.utils._os import upath
  14. if HAS_GDAL:
  15. from django.contrib.gis.utils.layermapping import (LayerMapping,
  16. LayerMapError, InvalidDecimal, MissingForeignKey)
  17. from django.contrib.gis.gdal import DataSource
  18. from .models import (
  19. City, County, CountyFeat, Interstate, ICity1, ICity2, Invalid, State,
  20. city_mapping, co_mapping, cofeat_mapping, inter_mapping)
  21. shp_path = os.path.realpath(os.path.join(os.path.dirname(upath(__file__)), os.pardir, 'data'))
  22. city_shp = os.path.join(shp_path, 'cities', 'cities.shp')
  23. co_shp = os.path.join(shp_path, 'counties', 'counties.shp')
  24. inter_shp = os.path.join(shp_path, 'interstates', 'interstates.shp')
  25. invalid_shp = os.path.join(shp_path, 'invalid', 'emptypoints.shp')
  26. # Dictionaries to hold what's expected in the county shapefile.
  27. NAMES = ['Bexar', 'Galveston', 'Harris', 'Honolulu', 'Pueblo']
  28. NUMS = [1, 2, 1, 19, 1] # Number of polygons for each.
  29. STATES = ['Texas', 'Texas', 'Texas', 'Hawaii', 'Colorado']
  30. @skipUnless(HAS_GDAL and HAS_SPATIAL_DB, "GDAL and spatial db are required.")
  31. class LayerMapTest(TestCase):
  32. def test_init(self):
  33. "Testing LayerMapping initialization."
  34. # Model field that does not exist.
  35. bad1 = copy(city_mapping)
  36. bad1['foobar'] = 'FooField'
  37. # Shapefile field that does not exist.
  38. bad2 = copy(city_mapping)
  39. bad2['name'] = 'Nombre'
  40. # Nonexistent geographic field type.
  41. bad3 = copy(city_mapping)
  42. bad3['point'] = 'CURVE'
  43. # Incrementing through the bad mapping dictionaries and
  44. # ensuring that a LayerMapError is raised.
  45. for bad_map in (bad1, bad2, bad3):
  46. with self.assertRaises(LayerMapError):
  47. LayerMapping(City, city_shp, bad_map)
  48. # A LookupError should be thrown for bogus encodings.
  49. with self.assertRaises(LookupError):
  50. LayerMapping(City, city_shp, city_mapping, encoding='foobar')
  51. def test_simple_layermap(self):
  52. "Test LayerMapping import of a simple point shapefile."
  53. # Setting up for the LayerMapping.
  54. lm = LayerMapping(City, city_shp, city_mapping)
  55. lm.save()
  56. # There should be three cities in the shape file.
  57. self.assertEqual(3, City.objects.count())
  58. # Opening up the shapefile, and verifying the values in each
  59. # of the features made it to the model.
  60. ds = DataSource(city_shp)
  61. layer = ds[0]
  62. for feat in layer:
  63. city = City.objects.get(name=feat['Name'].value)
  64. self.assertEqual(feat['Population'].value, city.population)
  65. self.assertEqual(Decimal(str(feat['Density'])), city.density)
  66. self.assertEqual(feat['Created'].value, city.dt)
  67. # Comparing the geometries.
  68. pnt1, pnt2 = feat.geom, city.point
  69. self.assertAlmostEqual(pnt1.x, pnt2.x, 5)
  70. self.assertAlmostEqual(pnt1.y, pnt2.y, 5)
  71. def test_layermap_strict(self):
  72. "Testing the `strict` keyword, and import of a LineString shapefile."
  73. # When the `strict` keyword is set an error encountered will force
  74. # the importation to stop.
  75. with self.assertRaises(InvalidDecimal):
  76. lm = LayerMapping(Interstate, inter_shp, inter_mapping)
  77. lm.save(silent=True, strict=True)
  78. Interstate.objects.all().delete()
  79. # This LayerMapping should work b/c `strict` is not set.
  80. lm = LayerMapping(Interstate, inter_shp, inter_mapping)
  81. lm.save(silent=True)
  82. # Two interstate should have imported correctly.
  83. self.assertEqual(2, Interstate.objects.count())
  84. # Verifying the values in the layer w/the model.
  85. ds = DataSource(inter_shp)
  86. # Only the first two features of this shapefile are valid.
  87. valid_feats = ds[0][:2]
  88. for feat in valid_feats:
  89. istate = Interstate.objects.get(name=feat['Name'].value)
  90. if feat.fid == 0:
  91. self.assertEqual(Decimal(str(feat['Length'])), istate.length)
  92. elif feat.fid == 1:
  93. # Everything but the first two decimal digits were truncated,
  94. # because the Interstate model's `length` field has decimal_places=2.
  95. self.assertAlmostEqual(feat.get('Length'), float(istate.length), 2)
  96. for p1, p2 in zip(feat.geom, istate.path):
  97. self.assertAlmostEqual(p1[0], p2[0], 6)
  98. self.assertAlmostEqual(p1[1], p2[1], 6)
  99. def county_helper(self, county_feat=True):
  100. "Helper function for ensuring the integrity of the mapped County models."
  101. for name, n, st in zip(NAMES, NUMS, STATES):
  102. # Should only be one record b/c of `unique` keyword.
  103. c = County.objects.get(name=name)
  104. self.assertEqual(n, len(c.mpoly))
  105. self.assertEqual(st, c.state.name) # Checking ForeignKey mapping.
  106. # Multiple records because `unique` was not set.
  107. if county_feat:
  108. qs = CountyFeat.objects.filter(name=name)
  109. self.assertEqual(n, qs.count())
  110. def test_layermap_unique_multigeometry_fk(self):
  111. "Testing the `unique`, and `transform`, geometry collection conversion, and ForeignKey mappings."
  112. # All the following should work.
  113. try:
  114. # Telling LayerMapping that we want no transformations performed on the data.
  115. lm = LayerMapping(County, co_shp, co_mapping, transform=False)
  116. # Specifying the source spatial reference system via the `source_srs` keyword.
  117. lm = LayerMapping(County, co_shp, co_mapping, source_srs=4269)
  118. lm = LayerMapping(County, co_shp, co_mapping, source_srs='NAD83')
  119. # Unique may take tuple or string parameters.
  120. for arg in ('name', ('name', 'mpoly')):
  121. lm = LayerMapping(County, co_shp, co_mapping, transform=False, unique=arg)
  122. except:
  123. self.fail('No exception should be raised for proper use of keywords.')
  124. # Testing invalid params for the `unique` keyword.
  125. for e, arg in ((TypeError, 5.0), (ValueError, 'foobar'), (ValueError, ('name', 'mpolygon'))):
  126. self.assertRaises(e, LayerMapping, County, co_shp, co_mapping, transform=False, unique=arg)
  127. # No source reference system defined in the shapefile, should raise an error.
  128. if not mysql:
  129. self.assertRaises(LayerMapError, LayerMapping, County, co_shp, co_mapping)
  130. # Passing in invalid ForeignKey mapping parameters -- must be a dictionary
  131. # mapping for the model the ForeignKey points to.
  132. bad_fk_map1 = copy(co_mapping)
  133. bad_fk_map1['state'] = 'name'
  134. bad_fk_map2 = copy(co_mapping)
  135. bad_fk_map2['state'] = {'nombre': 'State'}
  136. self.assertRaises(TypeError, LayerMapping, County, co_shp, bad_fk_map1, transform=False)
  137. self.assertRaises(LayerMapError, LayerMapping, County, co_shp, bad_fk_map2, transform=False)
  138. # There exist no State models for the ForeignKey mapping to work -- should raise
  139. # a MissingForeignKey exception (this error would be ignored if the `strict`
  140. # keyword is not set).
  141. lm = LayerMapping(County, co_shp, co_mapping, transform=False, unique='name')
  142. self.assertRaises(MissingForeignKey, lm.save, silent=True, strict=True)
  143. # Now creating the state models so the ForeignKey mapping may work.
  144. State.objects.bulk_create([
  145. State(name='Colorado'), State(name='Hawaii'), State(name='Texas')
  146. ])
  147. # If a mapping is specified as a collection, all OGR fields that
  148. # are not collections will be converted into them. For example,
  149. # a Point column would be converted to MultiPoint. Other things being done
  150. # w/the keyword args:
  151. # `transform=False`: Specifies that no transform is to be done; this
  152. # has the effect of ignoring the spatial reference check (because the
  153. # county shapefile does not have implicit spatial reference info).
  154. #
  155. # `unique='name'`: Creates models on the condition that they have
  156. # unique county names; geometries from each feature however will be
  157. # appended to the geometry collection of the unique model. Thus,
  158. # all of the various islands in Honolulu county will be in in one
  159. # database record with a MULTIPOLYGON type.
  160. lm = LayerMapping(County, co_shp, co_mapping, transform=False, unique='name')
  161. lm.save(silent=True, strict=True)
  162. # A reference that doesn't use the unique keyword; a new database record will
  163. # created for each polygon.
  164. lm = LayerMapping(CountyFeat, co_shp, cofeat_mapping, transform=False)
  165. lm.save(silent=True, strict=True)
  166. # The county helper is called to ensure integrity of County models.
  167. self.county_helper()
  168. def test_test_fid_range_step(self):
  169. "Tests the `fid_range` keyword and the `step` keyword of .save()."
  170. # Function for clearing out all the counties before testing.
  171. def clear_counties():
  172. County.objects.all().delete()
  173. State.objects.bulk_create([
  174. State(name='Colorado'), State(name='Hawaii'), State(name='Texas')
  175. ])
  176. # Initializing the LayerMapping object to use in these tests.
  177. lm = LayerMapping(County, co_shp, co_mapping, transform=False, unique='name')
  178. # Bad feature id ranges should raise a type error.
  179. bad_ranges = (5.0, 'foo', co_shp)
  180. for bad in bad_ranges:
  181. self.assertRaises(TypeError, lm.save, fid_range=bad)
  182. # Step keyword should not be allowed w/`fid_range`.
  183. fr = (3, 5) # layer[3:5]
  184. self.assertRaises(LayerMapError, lm.save, fid_range=fr, step=10)
  185. lm.save(fid_range=fr)
  186. # Features IDs 3 & 4 are for Galveston County, Texas -- only
  187. # one model is returned because the `unique` keyword was set.
  188. qs = County.objects.all()
  189. self.assertEqual(1, qs.count())
  190. self.assertEqual('Galveston', qs[0].name)
  191. # Features IDs 5 and beyond for Honolulu County, Hawaii, and
  192. # FID 0 is for Pueblo County, Colorado.
  193. clear_counties()
  194. lm.save(fid_range=slice(5, None), silent=True, strict=True) # layer[5:]
  195. lm.save(fid_range=slice(None, 1), silent=True, strict=True) # layer[:1]
  196. # Only Pueblo & Honolulu counties should be present because of
  197. # the `unique` keyword. Have to set `order_by` on this QuerySet
  198. # or else MySQL will return a different ordering than the other dbs.
  199. qs = County.objects.order_by('name')
  200. self.assertEqual(2, qs.count())
  201. hi, co = tuple(qs)
  202. hi_idx, co_idx = tuple(map(NAMES.index, ('Honolulu', 'Pueblo')))
  203. self.assertEqual('Pueblo', co.name)
  204. self.assertEqual(NUMS[co_idx], len(co.mpoly))
  205. self.assertEqual('Honolulu', hi.name)
  206. self.assertEqual(NUMS[hi_idx], len(hi.mpoly))
  207. # Testing the `step` keyword -- should get the same counties
  208. # regardless of we use a step that divides equally, that is odd,
  209. # or that is larger than the dataset.
  210. for st in (4, 7, 1000):
  211. clear_counties()
  212. lm.save(step=st, strict=True)
  213. self.county_helper(county_feat=False)
  214. def test_model_inheritance(self):
  215. "Tests LayerMapping on inherited models. See #12093."
  216. icity_mapping = {'name': 'Name',
  217. 'population': 'Population',
  218. 'density': 'Density',
  219. 'point': 'POINT',
  220. 'dt': 'Created',
  221. }
  222. # Parent model has geometry field.
  223. lm1 = LayerMapping(ICity1, city_shp, icity_mapping)
  224. lm1.save()
  225. # Grandparent has geometry field.
  226. lm2 = LayerMapping(ICity2, city_shp, icity_mapping)
  227. lm2.save()
  228. self.assertEqual(6, ICity1.objects.count())
  229. self.assertEqual(3, ICity2.objects.count())
  230. def test_invalid_layer(self):
  231. "Tests LayerMapping on invalid geometries. See #15378."
  232. invalid_mapping = {'point': 'POINT'}
  233. lm = LayerMapping(Invalid, invalid_shp, invalid_mapping,
  234. source_srs=4326)
  235. lm.save(silent=True)
  236. def test_textfield(self):
  237. "Tests that String content fits also in a TextField"
  238. mapping = copy(city_mapping)
  239. mapping['name_txt'] = 'Name'
  240. lm = LayerMapping(City, city_shp, mapping)
  241. lm.save(silent=True, strict=True)
  242. self.assertEqual(City.objects.count(), 3)
  243. self.assertEqual(City.objects.all().order_by('name_txt')[0].name_txt, "Houston")
  244. def test_encoded_name(self):
  245. """ Test a layer containing utf-8-encoded name """
  246. city_shp = os.path.join(shp_path, 'ch-city', 'ch-city.shp')
  247. lm = LayerMapping(City, city_shp, city_mapping)
  248. lm.save(silent=True, strict=True)
  249. self.assertEqual(City.objects.count(), 1)
  250. self.assertEqual(City.objects.all()[0].name, "Zürich")
  251. class OtherRouter(object):
  252. def db_for_read(self, model, **hints):
  253. return 'other'
  254. def db_for_write(self, model, **hints):
  255. return self.db_for_read(model, **hints)
  256. def allow_relation(self, obj1, obj2, **hints):
  257. return None
  258. def allow_migrate(self, db, model):
  259. return True
  260. @skipUnless(HAS_GDAL and HAS_SPATIAL_DB, "GDAL and spatial db are required.")
  261. class LayerMapRouterTest(TestCase):
  262. def setUp(self):
  263. self.old_routers = router.routers
  264. router.routers = [OtherRouter()]
  265. def tearDown(self):
  266. router.routers = self.old_routers
  267. @unittest.skipUnless(len(settings.DATABASES) > 1, 'multiple databases required')
  268. def test_layermapping_default_db(self):
  269. lm = LayerMapping(City, city_shp, city_mapping)
  270. self.assertEqual(lm.using, 'other')