using a regex because of small differences depending on gdal versions list of acceptable data sources loading up the data source the layer count is what s expected only 1 layer in a shp file making sure getname works making sure the driver name matches up making sure indexing works incrementing through each layer this tests datasource iter making sure we get the number of features we expect making sure we get the number of fields we expect testing the layer s extent an envelope and its properties now checking the field names negative fids are not allowed testing layer getfields which uses layer iter testing layer getitem maybe this should be in the test below but we might as well test the feature values here while in this loop using the first datasource because the same slice can be used for both the layer and the control values ensure ogr objects keep references to the objects they belong to see ticket 9448 this datasource object is not accessible outside this scope however a reference should still be kept alive on the layer returned making sure we can call ogr routines on the layer returned same issue for featurefield objects see 18640 incrementing through each layer incrementing through each feature in the layer making sure the number of fields and the geometry type are what s expected making sure the fields match to an appropriate oft type making sure we get the proper ogr field instance using a string value index for the feature testing feature iter incrementing through each layer and feature making sure we get the right geometry name type making sure the spatialreference is as expected when not set it should be none must be set aan ogrgeometry or 4tuple setting the spatial filter with a tuplelist with the extent of a buffer centering around pueblo setting the spatial filter with an ogrgeometry for buffer centering around houston clearing the spatial filter by setting it to none now should indicate that there are 3 features in the layer using dbf from census 2010 tiger shapefile for texas which has land area aland10 stored in a real field with no precision reference value obtained using ogrinfo using a regex because of small differences depending on gdal versions list of acceptable data sources got extent from qgis vrt uses csv which all types are oftstring min max from csv got extent from qgis got extent from qgis loading up the data source the layer count is what s expected only 1 layer in a shp file making sure getname works making sure the driver name matches up making sure indexing works incrementing through each layer this tests datasource __iter__ making sure we get the number of features we expect making sure we get the number of fields we expect testing the layer s extent an envelope and its properties now checking the field names negative fids are not allowed testing layer get_fields which uses layer __iter__ testing layer __getitem__ maybe this should be in the test below but we might as well test the feature values here while in this loop using the first data source because the same slice can be used for both the layer and the control values ensure ogr objects keep references to the objects they belong to see ticket 9448 this datasource object is not accessible outside this scope however a reference should still be kept alive on the layer returned making sure we can call ogr routines on the layer returned same issue for feature field objects see 18640 incrementing through each layer incrementing through each feature in the layer making sure the number of fields and the geometry type are what s expected making sure the fields match to an appropriate oft type making sure we get the proper ogr field instance using a string value index for the feature testing feature __iter__ incrementing through each layer and feature making sure we get the right geometry name type making sure the spatialreference is as expected when not set it should be none must be set a an ogrgeometry or 4 tuple setting the spatial filter with a tuple list with the extent of a buffer centering around pueblo setting the spatial filter with an ogrgeometry for buffer centering around houston clearing the spatial filter by setting it to none now should indicate that there are 3 features in the layer using dbf from census 2010 tiger shapefile for texas which has land area aland10 stored in a real field with no precision reference value obtained using ogrinfo
import os
import re
from datetime import datetime
from pathlib import Path

from django.contrib.gis.gdal import DataSource, Envelope, GDALException, OGRGeometry
from django.contrib.gis.gdal.field import OFTDateTime, OFTInteger, OFTReal, OFTString
from django.contrib.gis.geos import GEOSGeometry
from django.test import SimpleTestCase

from ..test_data import TEST_DATA, TestDS, get_ds_file

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

wgs_84_wkt_regex = r'^GEOGCS\["(GCS_)?WGS[ _](19)?84".*$'

datetime_format = "%Y-%m-%dT%H:%M:%S"


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),  
        srs_wkt=wgs_84_wkt,
        field_values={
            "dbl": [float(i) for i in range(1, 6)],
            "int": list(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="OGR_VRT",
        fields={
            "POINT_X": OFTString,
            "POINT_Y": OFTString,
            "NUM": OFTString,
        },  
        extent=(1.0, 2.0, 100.0, 523.5),  
        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),  
        srs_wkt=wgs_84_wkt,
    ),
    TestDS(
        "has_nulls",
        nfeat=3,
        nfld=6,
        geom="POLYGON",
        gtype=3,
        driver="GeoJSON",
        ext="geojson",
        fields={
            "uuid": OFTString,
            "name": OFTString,
            "num": OFTReal,
            "integer": OFTInteger,
            "datetime": OFTDateTime,
            "boolean": OFTInteger,
        },
        extent=(-75.274200, 39.846504, -74.959717, 40.119040),  
        field_values={
            "uuid": [
                "1378c26f-cbe6-44b0-929f-eb330d4991f5",
                "fa2ba67c-a135-4338-b924-a9622b5d869f",
                "4494c1f3-55ab-4256-b365-12115cb388d5",
            ],
            "name": ["Philadelphia", None, "north"],
            "num": [1.001, None, 0.0],
            "integer": [5, None, 8],
            "boolean": [True, None, False],
            "datetime": [
                datetime.strptime("1994-08-14T11:32:14", datetime_format),
                None,
                datetime.strptime("2018-11-29T03:02:52", datetime_format),
            ],
        },
        fids=range(3),
    ),
)

bad_ds = (TestDS("foo"),)


class DataSourceTest(SimpleTestCase):
    def test01_valid_shp(self):
        "Testing valid SHP Data Source files."

        for source in ds_list:
            
            ds = DataSource(source.ds)

            
            self.assertEqual(1, len(ds))

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

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

            
            msg = "Index out of range when accessing layers in a datasource: %s."
            with self.assertRaisesMessage(IndexError, msg % len(ds)):
                ds.__getitem__(len(ds))

            with self.assertRaisesMessage(
                IndexError, "Invalid OGR layer name given: invalid."
            ):
                ds.__getitem__("invalid")

    def test_ds_input_pathlib(self):
        test_shp = Path(get_ds_file("test_point", "shp"))
        ds = DataSource(test_shp)
        self.assertEqual(len(ds), 1)

    def test02_invalid_shp(self):
        "Testing invalid SHP files for the Data Source."
        for source in bad_ds:
            with self.assertRaises(GDALException):
                DataSource(source.ds)

    def test03a_layers(self):
        "Testing Data Source Layers."
        for source in ds_list:
            ds = DataSource(source.ds)

            
            for layer in ds:
                self.assertEqual(layer.name, source.name)
                self.assertEqual(str(layer), source.name)
                
                self.assertEqual(len(layer), source.nfeat)

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

                
                self.assertIsInstance(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)

                
                flds = layer.fields
                for f in flds:
                    self.assertIn(f, source.fields)

                
                with self.assertRaisesMessage(
                    IndexError, "Negative indices are not allowed on OGR Layers."
                ):
                    layer.__getitem__(-1)
                with self.assertRaisesMessage(IndexError, "Invalid feature id: 50000."):
                    layer.__getitem__(50000)

                if hasattr(source, "field_values"):
                    
                    for fld_name, fld_value in source.field_values.items():
                        self.assertEqual(fld_value, layer.get_fields(fld_name))

                    
                    for i, fid in enumerate(source.fids):
                        feat = layer[fid]
                        self.assertEqual(fid, feat.fid)
                        
                        
                        
                        for fld_name, fld_value in source.field_values.items():
                            self.assertEqual(fld_value[i], feat.get(fld_name))

                        msg = (
                            "Index out of range when accessing field in a feature: %s."
                        )
                        with self.assertRaisesMessage(IndexError, msg % len(feat)):
                            feat.__getitem__(len(feat))

                        with self.assertRaisesMessage(
                            IndexError, "Invalid OFT field name given: invalid."
                        ):
                            feat.__getitem__("invalid")

    def test03b_layer_slice(self):
        "Test indexing and slicing on Layers."
        
        
        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):
        
        source = ds_list[0]

        
        def get_layer():
            
            
            
            ds = DataSource(source.ds)
            return ds[0]

        
        lyr = get_layer()
        self.assertEqual(source.nfeat, len(lyr))
        self.assertEqual(source.gtype, lyr.geom_type.num)

        
        self.assertEqual(str(lyr[0]["str"]), "1")

    def test04_features(self):
        "Testing Data Source Features."
        for source in ds_list:
            ds = DataSource(source.ds)

            
            for layer in ds:
                
                for feat in layer:
                    
                    
                    self.assertEqual(source.nfld, len(list(feat)))
                    self.assertEqual(source.gtype, feat.geom_type)

                    
                    for k, v in source.fields.items():
                        
                        
                        self.assertIsInstance(feat[k], v)
                    self.assertIsInstance(feat.fields[0], str)

                    
                    for fld in feat:
                        self.assertIn(fld.name, source.fields)

    def test05_geometries(self):
        "Testing Geometries from Data Source Features."
        for source in ds_list:
            ds = DataSource(source.ds)

            
            for layer in ds:
                geoms = layer.get_geoms()
                geos_geoms = layer.get_geoms(geos=True)
                self.assertEqual(len(geoms), len(geos_geoms))
                self.assertEqual(len(geoms), len(layer))
                for feat, geom, geos_geom in zip(layer, geoms, geos_geoms):
                    g = feat.geom
                    self.assertEqual(geom, g)
                    self.assertIsInstance(geos_geom, GEOSGeometry)
                    self.assertEqual(g, geos_geom.ogr)
                    
                    self.assertEqual(source.geom, g.geom_name)
                    self.assertEqual(source.gtype, g.geom_type)

                    
                    if hasattr(source, "srs_wkt"):
                        self.assertIsNotNone(re.match(wgs_84_wkt_regex, g.srs.wkt))

    def test06_spatial_filter(self):
        "Testing the Layer.spatial_filter property."
        ds = DataSource(get_ds_file("cities", "shp"))
        lyr = ds[0]

        
        self.assertIsNone(lyr.spatial_filter)

        
        with self.assertRaises(TypeError):
            lyr._set_spatial_filter("foo")

        
        
        with self.assertRaises(ValueError):
            lyr._set_spatial_filter(list(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"))

        
        
        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"))

        
        
        lyr.spatial_filter = None
        self.assertEqual(3, len(lyr))

    def test07_integer_overflow(self):
        "Testing that OFTReal fields, treated as OFTInteger, do not overflow."
        
        
        
        ds = DataSource(os.path.join(TEST_DATA, "texas.dbf"))
        feat = ds[0][0]
        
        self.assertEqual(676586997978, feat.get("ALAND10"))

    def test_nonexistent_field(self):
        source = ds_list[0]
        ds = DataSource(source.ds)
        msg = "invalid field name: nonexistent"
        with self.assertRaisesMessage(GDALException, msg):
            ds[0].get_fields("nonexistent")