227 lines
10 KiB
Python
227 lines
10 KiB
Python
import os, os.path, unittest
|
|
from django.contrib.gis.gdal import DataSource, Envelope, OGRGeometry, OGRException, OGRIndexError, GDAL_VERSION
|
|
from django.contrib.gis.gdal.field import OFTReal, OFTInteger, OFTString
|
|
from django.contrib.gis.geometry.test_data import get_ds_file, TestDS
|
|
|
|
# List of acceptable data sources.
|
|
ds_list = (TestDS('test_point', nfeat=5, nfld=3, geom='POINT', gtype=1, driver='ESRI Shapefile',
|
|
fields={'dbl' : OFTReal, 'int' : OFTInteger, 'str' : OFTString,},
|
|
extent=(-1.35011,0.166623,-0.524093,0.824508), # Got extent from QGIS
|
|
srs_wkt='GEOGCS["GCS_WGS_1984",DATUM["WGS_1984",SPHEROID["WGS_1984",6378137,298.257223563]],PRIMEM["Greenwich",0],UNIT["Degree",0.017453292519943295]]',
|
|
field_values={'dbl' : [float(i) for i in range(1, 6)], 'int' : range(1, 6), 'str' : [str(i) for i in range(1, 6)]},
|
|
fids=range(5)),
|
|
TestDS('test_vrt', ext='vrt', nfeat=3, nfld=3, geom='POINT', gtype='Point25D', driver='VRT',
|
|
fields={'POINT_X' : OFTString, 'POINT_Y' : OFTString, 'NUM' : OFTString}, # VRT uses CSV, which all types are OFTString.
|
|
extent=(1.0, 2.0, 100.0, 523.5), # Min/Max from CSV
|
|
field_values={'POINT_X' : ['1.0', '5.0', '100.0'], 'POINT_Y' : ['2.0', '23.0', '523.5'], 'NUM' : ['5', '17', '23']},
|
|
fids=range(1,4)),
|
|
TestDS('test_poly', nfeat=3, nfld=3, geom='POLYGON', gtype=3,
|
|
driver='ESRI Shapefile',
|
|
fields={'float' : OFTReal, 'int' : OFTInteger, 'str' : OFTString,},
|
|
extent=(-1.01513,-0.558245,0.161876,0.839637), # Got extent from QGIS
|
|
srs_wkt='GEOGCS["GCS_WGS_1984",DATUM["WGS_1984",SPHEROID["WGS_1984",6378137,298.257223563]],PRIMEM["Greenwich",0],UNIT["Degree",0.017453292519943295]]'),
|
|
)
|
|
|
|
bad_ds = (TestDS('foo'),
|
|
)
|
|
|
|
class DataSourceTest(unittest.TestCase):
|
|
|
|
def test01_valid_shp(self):
|
|
"Testing valid SHP Data Source files."
|
|
|
|
for source in ds_list:
|
|
# Loading up the data source
|
|
ds = DataSource(source.ds)
|
|
|
|
# Making sure the layer count is what's expected (only 1 layer in a SHP file)
|
|
self.assertEqual(1, len(ds))
|
|
|
|
# Making sure GetName works
|
|
self.assertEqual(source.ds, ds.name)
|
|
|
|
# Making sure the driver name matches up
|
|
self.assertEqual(source.driver, str(ds.driver))
|
|
|
|
# Making sure indexing works
|
|
try:
|
|
ds[len(ds)]
|
|
except OGRIndexError:
|
|
pass
|
|
else:
|
|
self.fail('Expected an IndexError!')
|
|
|
|
def test02_invalid_shp(self):
|
|
"Testing invalid SHP files for the Data Source."
|
|
for source in bad_ds:
|
|
self.assertRaises(OGRException, DataSource, source.ds)
|
|
|
|
def test03a_layers(self):
|
|
"Testing Data Source Layers."
|
|
print "\nBEGIN - expecting out of range feature id error; safe to ignore.\n"
|
|
for source in ds_list:
|
|
ds = DataSource(source.ds)
|
|
|
|
# Incrementing through each layer, this tests DataSource.__iter__
|
|
for layer in ds:
|
|
# Making sure we get the number of features we expect
|
|
self.assertEqual(len(layer), source.nfeat)
|
|
|
|
# Making sure we get the number of fields we expect
|
|
self.assertEqual(source.nfld, layer.num_fields)
|
|
self.assertEqual(source.nfld, len(layer.fields))
|
|
|
|
# Testing the layer's extent (an Envelope), and it's properties
|
|
if source.driver == 'VRT' and (GDAL_VERSION > (1, 7, 0) and GDAL_VERSION < (1, 7, 3)):
|
|
# There's a known GDAL regression with retrieving the extent
|
|
# of a VRT layer in versions 1.7.0-1.7.2:
|
|
# http://trac.osgeo.org/gdal/ticket/3783
|
|
pass
|
|
else:
|
|
self.assertEqual(True, isinstance(layer.extent, Envelope))
|
|
self.assertAlmostEqual(source.extent[0], layer.extent.min_x, 5)
|
|
self.assertAlmostEqual(source.extent[1], layer.extent.min_y, 5)
|
|
self.assertAlmostEqual(source.extent[2], layer.extent.max_x, 5)
|
|
self.assertAlmostEqual(source.extent[3], layer.extent.max_y, 5)
|
|
|
|
# Now checking the field names.
|
|
flds = layer.fields
|
|
for f in flds: self.assertEqual(True, f in source.fields)
|
|
|
|
# Negative FIDs are not allowed.
|
|
self.assertRaises(OGRIndexError, layer.__getitem__, -1)
|
|
self.assertRaises(OGRIndexError, layer.__getitem__, 50000)
|
|
|
|
if hasattr(source, 'field_values'):
|
|
fld_names = source.field_values.keys()
|
|
|
|
# Testing `Layer.get_fields` (which uses Layer.__iter__)
|
|
for fld_name in fld_names:
|
|
self.assertEqual(source.field_values[fld_name], layer.get_fields(fld_name))
|
|
|
|
# Testing `Layer.__getitem__`.
|
|
for i, fid in enumerate(source.fids):
|
|
feat = layer[fid]
|
|
self.assertEqual(fid, feat.fid)
|
|
# Maybe this should be in the test below, but we might as well test
|
|
# the feature values here while in this loop.
|
|
for fld_name in fld_names:
|
|
self.assertEqual(source.field_values[fld_name][i], feat.get(fld_name))
|
|
print "\nEND - expecting out of range feature id error; safe to ignore."
|
|
|
|
def test03b_layer_slice(self):
|
|
"Test indexing and slicing on Layers."
|
|
# Using the first data-source because the same slice
|
|
# can be used for both the layer and the control values.
|
|
source = ds_list[0]
|
|
ds = DataSource(source.ds)
|
|
|
|
sl = slice(1, 3)
|
|
feats = ds[0][sl]
|
|
|
|
for fld_name in ds[0].fields:
|
|
test_vals = [feat.get(fld_name) for feat in feats]
|
|
control_vals = source.field_values[fld_name][sl]
|
|
self.assertEqual(control_vals, test_vals)
|
|
|
|
def test03c_layer_references(self):
|
|
"Test to make sure Layer access is still available without the DataSource."
|
|
source = ds_list[0]
|
|
|
|
# See ticket #9448.
|
|
def get_layer():
|
|
# This DataSource object is not accessible outside this
|
|
# scope. However, a reference should still be kept alive
|
|
# on the `Layer` returned.
|
|
ds = DataSource(source.ds)
|
|
return ds[0]
|
|
|
|
# Making sure we can call OGR routines on the Layer returned.
|
|
lyr = get_layer()
|
|
self.assertEqual(source.nfeat, len(lyr))
|
|
self.assertEqual(source.gtype, lyr.geom_type.num)
|
|
|
|
def test04_features(self):
|
|
"Testing Data Source Features."
|
|
for source in ds_list:
|
|
ds = DataSource(source.ds)
|
|
|
|
# Incrementing through each layer
|
|
for layer in ds:
|
|
# Incrementing through each feature in the layer
|
|
for feat in layer:
|
|
# Making sure the number of fields, and the geometry type
|
|
# are what's expected.
|
|
self.assertEqual(source.nfld, len(list(feat)))
|
|
self.assertEqual(source.gtype, feat.geom_type)
|
|
|
|
# Making sure the fields match to an appropriate OFT type.
|
|
for k, v in source.fields.items():
|
|
# Making sure we get the proper OGR Field instance, using
|
|
# a string value index for the feature.
|
|
self.assertEqual(True, isinstance(feat[k], v))
|
|
|
|
# Testing Feature.__iter__
|
|
for fld in feat: self.assertEqual(True, fld.name in source.fields.keys())
|
|
|
|
def test05_geometries(self):
|
|
"Testing Geometries from Data Source Features."
|
|
for source in ds_list:
|
|
ds = DataSource(source.ds)
|
|
|
|
# Incrementing through each layer and feature.
|
|
for layer in ds:
|
|
for feat in layer:
|
|
g = feat.geom
|
|
|
|
# Making sure we get the right Geometry name & type
|
|
self.assertEqual(source.geom, g.geom_name)
|
|
self.assertEqual(source.gtype, g.geom_type)
|
|
|
|
# Making sure the SpatialReference is as expected.
|
|
if hasattr(source, 'srs_wkt'):
|
|
self.assertEqual(source.srs_wkt, g.srs.wkt)
|
|
|
|
def test06_spatial_filter(self):
|
|
"Testing the Layer.spatial_filter property."
|
|
ds = DataSource(get_ds_file('cities', 'shp'))
|
|
lyr = ds[0]
|
|
|
|
# When not set, it should be None.
|
|
self.assertEqual(None, lyr.spatial_filter)
|
|
|
|
# Must be set a/an OGRGeometry or 4-tuple.
|
|
self.assertRaises(TypeError, lyr._set_spatial_filter, 'foo')
|
|
|
|
# Setting the spatial filter with a tuple/list with the extent of
|
|
# a buffer centering around Pueblo.
|
|
self.assertRaises(ValueError, lyr._set_spatial_filter, range(5))
|
|
filter_extent = (-105.609252, 37.255001, -103.609252, 39.255001)
|
|
lyr.spatial_filter = (-105.609252, 37.255001, -103.609252, 39.255001)
|
|
self.assertEqual(OGRGeometry.from_bbox(filter_extent), lyr.spatial_filter)
|
|
feats = [feat for feat in lyr]
|
|
self.assertEqual(1, len(feats))
|
|
self.assertEqual('Pueblo', feats[0].get('Name'))
|
|
|
|
# Setting the spatial filter with an OGRGeometry for buffer centering
|
|
# around Houston.
|
|
filter_geom = OGRGeometry('POLYGON((-96.363151 28.763374,-94.363151 28.763374,-94.363151 30.763374,-96.363151 30.763374,-96.363151 28.763374))')
|
|
lyr.spatial_filter = filter_geom
|
|
self.assertEqual(filter_geom, lyr.spatial_filter)
|
|
feats = [feat for feat in lyr]
|
|
self.assertEqual(1, len(feats))
|
|
self.assertEqual('Houston', feats[0].get('Name'))
|
|
|
|
# Clearing the spatial filter by setting it to None. Now
|
|
# should indicate that there are 3 features in the Layer.
|
|
lyr.spatial_filter = None
|
|
self.assertEqual(3, len(lyr))
|
|
|
|
def suite():
|
|
s = unittest.TestSuite()
|
|
s.addTest(unittest.makeSuite(DataSourceTest))
|
|
return s
|
|
|
|
def run(verbosity=2):
|
|
unittest.TextTestRunner(verbosity=verbosity).run(suite())
|