1 import os, os.path, unittest

     2 from django.contrib.gis.gdal import DataSource, Envelope, OGRException, OGRIndexError

     3 from django.contrib.gis.gdal.field import OFTReal, OFTInteger, OFTString

     4 from django.contrib import gis

     5 

     6 # Path for SHP files

     7 data_path = os.path.join(os.path.dirname(gis.__file__), 'tests' + os.sep + 'data')

     8 def get_ds_file(name, ext):

     9     

    10 

    11     return os.sep.join([data_path, name, name + '.%s' % ext])

    12 

    13 # Test SHP data source object

    14 class TestDS:

    15     def __init__(self, name, **kwargs):

    16         ext = kwargs.pop('ext', 'shp')

    17         self.ds = get_ds_file(name, ext)

    18         for key, value in kwargs.items():

    19             setattr(self, key, value)

    20 

    21 # List of acceptable data sources.

    22 ds_list = (TestDS('test_point', nfeat=5, nfld=3, geom='POINT', gtype=1, driver='ESRI Shapefile',

    23                   fields={'dbl' : OFTReal, 'int' : OFTInteger, 'str' : OFTString,},

    24                   extent=(-1.35011,0.166623,-0.524093,0.824508), # Got extent from QGIS

    25                   srs_wkt='GEOGCS["GCS_WGS_1984",DATUM["WGS_1984",SPHEROID["WGS_1984",6378137,298.257223563]],PRIMEM["Greenwich",0],UNIT["Degree",0.017453292519943295]]',

    26                   field_values={'dbl' : [float(i) for i in range(1, 6)], 'int' : range(1, 6), 'str' : [str(i) for i in range(1, 6)]},

    27                   fids=range(5)),

    28            TestDS('test_vrt', ext='vrt', nfeat=3, nfld=3, geom='POINT', gtype=1, driver='VRT',

    29                   fields={'POINT_X' : OFTString, 'POINT_Y' : OFTString, 'NUM' : OFTString}, # VRT uses CSV, which all types are OFTString.

    30                   extent=(1.0, 2.0, 100.0, 523.5), # Min/Max from CSV

    31                   field_values={'POINT_X' : ['1.0', '5.0', '100.0'], 'POINT_Y' : ['2.0', '23.0', '523.5'], 'NUM' : ['5', '17', '23']},

    32                   fids=range(1,4)),

    33            TestDS('test_poly', nfeat=3, nfld=3, geom='POLYGON', gtype=3, 

    34                   driver='ESRI Shapefile',

    35                   fields={'float' : OFTReal, 'int' : OFTInteger, 'str' : OFTString,},

    36                   extent=(-1.01513,-0.558245,0.161876,0.839637), # Got extent from QGIS

    37                   srs_wkt='GEOGCS["GCS_WGS_1984",DATUM["WGS_1984",SPHEROID["WGS_1984",6378137,298.257223563]],PRIMEM["Greenwich",0],UNIT["Degree",0.017453292519943295]]'),

    38            )

    39 

    40 bad_ds = (TestDS('foo'),

    41           )

    42 

    43 class DataSourceTest(unittest.TestCase):

    44 

    45     def test01_valid_shp(self):

    46         "Testing valid SHP Data Source files."

    47 

    48         for source in ds_list:

    49             # Loading up the data source

    50             ds = DataSource(source.ds)

    51 

    52             # Making sure the layer count is what's expected (only 1 layer in a SHP file)

    53             self.assertEqual(1, len(ds))

    54 

    55             # Making sure GetName works

    56             self.assertEqual(source.ds, ds.name)

    57 

    58             # Making sure the driver name matches up

    59             self.assertEqual(source.driver, str(ds.driver))

    60 

    61             # Making sure indexing works

    62             try:

    63                 ds[len(ds)]

    64             except OGRIndexError:

    65                 pass

    66             else:

    67                 self.fail('Expected an IndexError!')

    68                         

    69     def test02_invalid_shp(self):

    70         "Testing invalid SHP files for the Data Source."

    71         for source in bad_ds:

    72             self.assertRaises(OGRException, DataSource, source.ds)

    73 

    74     def test03a_layers(self):

    75         "Testing Data Source Layers."

    76         print "\nBEGIN - expecting out of range feature id error; safe to ignore.\n"

    77         for source in ds_list:

    78             ds = DataSource(source.ds)

    79 

    80             # Incrementing through each layer, this tests DataSource.__iter__

    81             for layer in ds:                

    82                 # Making sure we get the number of features we expect

    83                 self.assertEqual(len(layer), source.nfeat)

    84 

    85                 # Making sure we get the number of fields we expect

    86                 self.assertEqual(source.nfld, layer.num_fields)

    87                 self.assertEqual(source.nfld, len(layer.fields))

    88 

    89                 # Testing the layer's extent (an Envelope), and it's properties

    90                 self.assertEqual(True, isinstance(layer.extent, Envelope))

    91                 self.assertAlmostEqual(source.extent[0], layer.extent.min_x, 5)

    92                 self.assertAlmostEqual(source.extent[1], layer.extent.min_y, 5)

    93                 self.assertAlmostEqual(source.extent[2], layer.extent.max_x, 5)

    94                 self.assertAlmostEqual(source.extent[3], layer.extent.max_y, 5)

    95 

    96                 # Now checking the field names.

    97                 flds = layer.fields

    98                 for f in flds: self.assertEqual(True, f in source.fields)

    99                 

   100                 # Negative FIDs are not allowed.

   101                 self.assertRaises(OGRIndexError, layer.__getitem__, -1)

   102                 self.assertRaises(OGRIndexError, layer.__getitem__, 50000)

   103 

   104                 if hasattr(source, 'field_values'):

   105                     fld_names = source.field_values.keys()

   106 

   107                     # Testing `Layer.get_fields` (which uses Layer.__iter__)

   108                     for fld_name in fld_names:

   109                         self.assertEqual(source.field_values[fld_name], layer.get_fields(fld_name))

   110 

   111                     # Testing `Layer.__getitem__`.

   112                     for i, fid in enumerate(source.fids):

   113                         feat = layer[fid]

   114                         self.assertEqual(fid, feat.fid)

   115                         # Maybe this should be in the test below, but we might as well test

   116                         # the feature values here while in this loop.

   117                         for fld_name in fld_names:

   118                             self.assertEqual(source.field_values[fld_name][i], feat.get(fld_name))

   119         print "\nEND - expecting out of range feature id error; safe to ignore."

   120                         

   121     def test03b_layer_slice(self):

   122         "Test indexing and slicing on Layers."

   123         # Using the first data-source because the same slice

   124         # can be used for both the layer and the control values.

   125         source = ds_list[0]

   126         ds = DataSource(source.ds)

   127 

   128         sl = slice(1, 3)

   129         feats = ds[0][sl]

   130 

   131         for fld_name in ds[0].fields:

   132             test_vals = [feat.get(fld_name) for feat in feats]

   133             control_vals = source.field_values[fld_name][sl]

   134             self.assertEqual(control_vals, test_vals)

   135 

   136     def test03c_layer_references(self):

   137         "Test to make sure Layer access is still available without the DataSource."

   138         source = ds_list[0]

   139 

   140         # See ticket #9448.

   141         def get_layer():

   142             # This DataSource object is not accessible outside this

   143             # scope.  However, a reference should still be kept alive

   144             # on the `Layer` returned.

   145             ds = DataSource(source.ds)

   146             return ds[0]

   147 

   148         # Making sure we can call OGR routines on the Layer returned.

   149         lyr = get_layer()

   150         self.assertEqual(source.nfeat, len(lyr))

   151         self.assertEqual(source.gtype, lyr.geom_type.num)        

   152 

   153     def test04_features(self):

   154         "Testing Data Source Features."

   155         for source in ds_list:

   156             ds = DataSource(source.ds)

   157 

   158             # Incrementing through each layer

   159             for layer in ds:

   160                 # Incrementing through each feature in the layer

   161                 for feat in layer:

   162                     # Making sure the number of fields, and the geometry type

   163                     # are what's expected.

   164                     self.assertEqual(source.nfld, len(list(feat)))

   165                     self.assertEqual(source.gtype, feat.geom_type)

   166 

   167                     # Making sure the fields match to an appropriate OFT type.

   168                     for k, v in source.fields.items():

   169                         # Making sure we get the proper OGR Field instance, using

   170                         # a string value index for the feature.

   171                         self.assertEqual(True, isinstance(feat[k], v))

   172 

   173                     # Testing Feature.__iter__

   174                     for fld in feat: self.assertEqual(True, fld.name in source.fields.keys())

   175                         

   176     def test05_geometries(self):

   177         "Testing Geometries from Data Source Features."

   178         for source in ds_list:

   179             ds = DataSource(source.ds)

   180 

   181             # Incrementing through each layer and feature.

   182             for layer in ds:

   183                 for feat in layer:

   184                     g = feat.geom

   185 

   186                     # Making sure we get the right Geometry name & type

   187                     self.assertEqual(source.geom, g.geom_name)

   188                     self.assertEqual(source.gtype, g.geom_type)

   189 

   190                     # Making sure the SpatialReference is as expected.

   191                     if hasattr(source, 'srs_wkt'):

   192                         self.assertEqual(source.srs_wkt, g.srs.wkt)

   193 

   194 

   195 def suite():

   196     s = unittest.TestSuite()

   197     s.addTest(unittest.makeSuite(DataSourceTest))

   198     return s

   199 

   200 def run(verbosity=2):

   201     unittest.TextTestRunner(verbosity=verbosity).run(suite())