#!/usr/bin/env pytest
# -*- coding: utf-8 -*-
###############################################################################
# $Id: ogr_rfc41.py 2f4e0fa743974d54a54b5c694595454a0b672917 2019-11-30 16:18:05 +0100 Even Rouault $
#
# Project:  GDAL/OGR Test Suite
# Purpose:  Test RFC41
# Author:   Even Rouault <even dot rouault at spatialys.com>
#
###############################################################################
# Copyright (c) 2013, Even Rouault <even dot rouault at spatialys.com>
#
# Permission is hereby granted, free of charge, to any person obtaining a
# copy of this software and associated documentation files (the "Software"),
# to deal in the Software without restriction, including without limitation
# the rights to use, copy, modify, merge, publish, distribute, sublicense,
# and/or sell copies of the Software, and to permit persons to whom the
# Software is furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included
# in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS
# OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL
# THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
# FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
# DEALINGS IN THE SOFTWARE.
###############################################################################


from osgeo import ogr
from osgeo import osr
from osgeo import gdal
import pytest

from ogr.ogr_sql_sqlite import require_ogr_sql_sqlite  # noqa
require_ogr_sql_sqlite; # to make pyflakes happy

###############################################################################
# Test OGRGeomFieldDefn class


def test_ogr_rfc41_1():

    gfld_defn = ogr.GeomFieldDefn()

    # Check default values
    assert gfld_defn.GetName() == ''
    assert gfld_defn.GetType() == ogr.wkbUnknown
    assert gfld_defn.GetSpatialRef() is None
    assert gfld_defn.IsIgnored() == 0

    # Test SetName() / GetName()
    gfld_defn.SetName('foo')
    assert gfld_defn.GetName() == 'foo'

    # Test SetType() / GetType()
    gfld_defn.SetType(ogr.wkbPoint)
    assert gfld_defn.GetType() == ogr.wkbPoint

    # Test SetSpatialRef() / GetSpatialRef()
    sr = osr.SpatialReference()
    gfld_defn.SetSpatialRef(sr)
    got_sr = gfld_defn.GetSpatialRef()
    assert got_sr.IsSame(sr) != 0

    gfld_defn.SetSpatialRef(None)
    assert gfld_defn.GetSpatialRef() is None

    gfld_defn.SetSpatialRef(sr)

    # Test SetIgnored() / IsIgnored()
    gfld_defn.SetIgnored(1)
    assert gfld_defn.IsIgnored() == 1

    # Test setting invalid value
    old_val = gfld_defn.GetType()
    gdal.PushErrorHandler('CPLQuietErrorHandler')
    gfld_defn.SetType(-3)
    gdal.PopErrorHandler()
    assert gfld_defn.GetType() == old_val

    gfld_defn = None


###############################################################################
# Test OGRFeatureDefn methods related to OGRGeomFieldDefn class

def test_ogr_rfc41_2():

    # Check implicit geometry field creation
    feature_defn = ogr.FeatureDefn()
    assert feature_defn.GetGeomFieldCount() == 1
    assert feature_defn.GetGeomType() == ogr.wkbUnknown

    # Test IsSame()
    assert feature_defn.IsSame(feature_defn) == 1
    other_feature_defn = ogr.FeatureDefn()
    assert feature_defn.IsSame(other_feature_defn) == 1
    other_feature_defn.GetGeomFieldDefn(0).SetSpatialRef(osr.SpatialReference())
    assert feature_defn.IsSame(other_feature_defn) == 0
    feature_defn.GetGeomFieldDefn(0).SetSpatialRef(osr.SpatialReference())
    assert feature_defn.IsSame(other_feature_defn) == 1
    other_feature_defn.GetGeomFieldDefn(0).SetSpatialRef(None)
    assert feature_defn.IsSame(other_feature_defn) == 0

    feature_defn = None
    feature_defn = ogr.FeatureDefn()

    # Check changing geometry type
    feature_defn.SetGeomType(ogr.wkbPoint)
    assert feature_defn.GetGeomType() == ogr.wkbPoint
    assert feature_defn.GetGeomFieldDefn(0).GetType() == ogr.wkbPoint

    # Check setting to wkbNone and implicitly destroying the field.
    for _ in range(2):
        feature_defn.SetGeomType(ogr.wkbNone)
        assert feature_defn.GetGeomFieldCount() == 0
        assert feature_defn.GetGeomType() == ogr.wkbNone

    # Recreate the field
    for t in [ogr.wkbPoint, ogr.wkbLineString]:
        feature_defn.SetGeomType(t)
        assert feature_defn.GetGeomFieldCount() == 1
        assert feature_defn.GetGeomType() == t
        assert feature_defn.GetGeomFieldDefn(0).GetType() == t

    # Test setting invalid value
    old_val = feature_defn.GetGeomType()
    gdal.PushErrorHandler('CPLQuietErrorHandler')
    feature_defn.SetGeomType(-3)
    gdal.PopErrorHandler()
    assert feature_defn.GetGeomType() == old_val

    # Test SetIgnored() / IsIgnored()
    assert feature_defn.IsGeometryIgnored() == 0
    assert feature_defn.GetGeomFieldDefn(0).IsIgnored() == 0
    feature_defn.SetGeometryIgnored(1)
    assert feature_defn.IsGeometryIgnored() == 1
    assert feature_defn.GetGeomFieldDefn(0).IsIgnored() == 1

    # Test wrong index values for GetGeomFieldDefn()
    for idx in [-1, 1]:
        gdal.PushErrorHandler('CPLQuietErrorHandler')
        ret = feature_defn.GetGeomFieldDefn(idx)
        gdal.PopErrorHandler()
        assert ret is None

    # Test GetGeomFieldIndex()
    assert feature_defn.GetGeomFieldIndex("") == 0
    assert feature_defn.GetGeomFieldIndex("invalid") == -1

    # Test AddGeomFieldDefn()
    gfld_defn = ogr.GeomFieldDefn('polygon_field', ogr.wkbPolygon)
    feature_defn.AddGeomFieldDefn(gfld_defn)
    assert feature_defn.GetGeomFieldCount() == 2
    assert feature_defn.GetGeomFieldIndex("polygon_field") == 1
    assert feature_defn.GetGeomFieldDefn(1).GetName() == 'polygon_field'

    # Test DeleteGeomFieldDefn() : error cases
    assert feature_defn.DeleteGeomFieldDefn(-1) != 0
    assert feature_defn.DeleteGeomFieldDefn(2) != 0
    assert feature_defn.GetGeomFieldCount() == 2

    # Test DeleteGeomFieldDefn() : valid cases
    assert feature_defn.DeleteGeomFieldDefn(0) == 0
    assert feature_defn.GetGeomFieldCount() == 1
    assert feature_defn.GetGeomFieldIndex("polygon_field") == 0

    assert feature_defn.DeleteGeomFieldDefn(0) == 0
    assert feature_defn.GetGeomFieldCount() == 0

    assert feature_defn.IsSame(feature_defn) == 1
    assert feature_defn.IsSame(ogr.FeatureDefn()) == 0

    feature_defn = None

###############################################################################
# Test OGRFeature methods


def test_ogr_rfc41_3():

    # Test with just one geometry field
    feature_defn = ogr.FeatureDefn()
    feature = ogr.Feature(feature_defn)
    assert feature.GetGeomFieldCount() == 1
    assert feature.GetGeomFieldDefnRef(0).GetName() == ''
    assert feature.GetGeomFieldDefnRef(0).GetType() == ogr.wkbUnknown
    assert feature.GetGeomFieldIndex('') == 0
    assert feature.GetGeomFieldIndex('non_existing') == -1
    assert feature.GetGeomFieldRef(-1) is None
    assert feature.GetGeomFieldRef(0) is None
    assert feature.GetGeomFieldRef(1) is None
    feature_clone_without_geom = feature.Clone()
    assert feature.Equal(feature_clone_without_geom)
    assert feature.SetGeomField(0, ogr.Geometry(ogr.wkbPoint)) == 0
    assert feature.GetGeomFieldRef(0).ExportToWkt() == 'POINT EMPTY'
    assert feature.Equal(feature.Clone())
    assert not feature.Equal(feature_clone_without_geom)
    feature_clone_with_other_geom = feature.Clone()
    feature_clone_with_other_geom.SetGeometry(ogr.Geometry(ogr.wkbLineString))
    assert not feature.Equal(feature_clone_with_other_geom)
    assert feature.SetGeomFieldDirectly(-1, None) != 0
    assert feature.SetGeomFieldDirectly(0, ogr.Geometry(ogr.wkbLineString)) == 0
    assert feature.GetGeomFieldRef(0).ExportToWkt() == 'LINESTRING EMPTY'
    feature_clone_with_geom = feature.Clone()
    assert feature.SetGeomFieldDirectly(0, None) == 0
    assert feature.GetGeomFieldRef(0) is None
    assert not feature.Equal(feature_clone_with_geom)
    feature = None

    # Test one a feature with 0 geometry field
    feature_defn = ogr.FeatureDefn()
    feature_defn.SetGeomType(ogr.wkbNone)
    feature = ogr.Feature(feature_defn)
    assert feature.Equal(feature.Clone())
    assert feature.GetGeomFieldCount() == 0
    # This used to work before RFC 41, but it no longer will
    assert feature.SetGeometry(ogr.Geometry(ogr.wkbPoint)) != 0
    assert feature.SetGeomField(0, ogr.Geometry(ogr.wkbPoint)) != 0
    assert feature.GetGeometryRef() is None
    assert feature.GetGeomFieldRef(0) is None
    assert feature.SetGeometryDirectly(ogr.Geometry(ogr.wkbPoint)) != 0
    assert feature.SetGeomFieldDirectly(0, ogr.Geometry(ogr.wkbPoint)) != 0
    feature = None

    # Test one a feature with several geometry fields
    feature_defn = ogr.FeatureDefn()
    feature_defn.SetGeomType(ogr.wkbNone)
    gfld_defn = ogr.GeomFieldDefn('polygon_field', ogr.wkbPolygon)
    feature_defn.AddGeomFieldDefn(gfld_defn)
    gfld_defn = ogr.GeomFieldDefn('point_field', ogr.wkbPoint)
    feature_defn.AddGeomFieldDefn(gfld_defn)
    feature = ogr.Feature(feature_defn)
    feature.SetGeomField(0, ogr.Geometry(ogr.wkbPolygon))
    feature.SetGeomField(1, ogr.Geometry(ogr.wkbPoint))
    assert feature.GetGeomFieldRef(0).ExportToWkt() == 'POLYGON EMPTY'
    assert feature.GetGeomFieldRef(1).ExportToWkt() == 'POINT EMPTY'
    assert feature.Equal(feature.Clone())
    other_feature = ogr.Feature(feature_defn)
    assert not feature.Equal(other_feature)
    other_feature.SetFrom(feature)
    assert feature.Equal(other_feature)

    # Test that in SetFrom() where target has a single geometry field,
    # we get the first geometry of the source even if we cannot find a
    # source geometry field with the right name.
    feature_defn_default = ogr.FeatureDefn()
    feature_default = ogr.Feature(feature_defn_default)
    feature_default.SetFrom(feature)
    assert feature_default.GetGeomFieldRef(0).ExportToWkt() == 'POLYGON EMPTY'

###############################################################################
# Test OGRLayer methods


def test_ogr_rfc41_4():

    ds = ogr.GetDriverByName('memory').CreateDataSource('')
    assert ds.TestCapability(ogr.ODsCCreateGeomFieldAfterCreateLayer) != 0
    sr = osr.SpatialReference()
    lyr = ds.CreateLayer('test', geom_type=ogr.wkbPoint, srs=sr)
    assert lyr.TestCapability(ogr.OLCCreateGeomField) != 0
    assert lyr.GetSpatialRef().IsSame(sr) != 0
    assert lyr.GetLayerDefn().GetGeomFieldDefn(0).GetSpatialRef().IsSame(sr) != 0
    lyr.GetLayerDefn().GetGeomFieldDefn(0).SetName('a_name')
    feat = ogr.Feature(lyr.GetLayerDefn())
    feat.SetGeometry(ogr.CreateGeometryFromWkt('POINT (1 2)'))
    lyr.CreateFeature(feat)
    lyr.ResetReading()
    feat = lyr.GetNextFeature()
    geom = feat.GetGeometryRef()
    assert geom.GetSpatialReference().IsSame(sr) != 0
    feat = None
    lyr.CreateGeomField(ogr.GeomFieldDefn('another_geom_field', ogr.wkbPolygon))
    lyr.ResetReading()
    feat = lyr.GetNextFeature()
    feat.SetGeomField(1, ogr.CreateGeometryFromWkt('POLYGON ((10 10,10 11,11 11,11 10,10 10))'))
    lyr.SetFeature(feat)
    lyr.ResetReading()
    feat = lyr.GetNextFeature()
    geom = feat.GetGeomFieldRef(0)
    assert geom.ExportToWkt() == 'POINT (1 2)'
    geom = feat.GetGeomFieldRef('another_geom_field')
    assert geom.ExportToWkt() == 'POLYGON ((10 10,10 11,11 11,11 10,10 10))'

    # Test GetExtent()
    got_extent = lyr.GetExtent(geom_field=1)
    assert got_extent == (10.0, 11.0, 10.0, 11.0)
    # Test invalid geometry field index
    gdal.ErrorReset()
    gdal.PushErrorHandler('CPLQuietErrorHandler')
    got_extent = lyr.GetExtent(geom_field=2)
    gdal.PopErrorHandler()
    assert gdal.GetLastErrorMsg() != ''

    # Test SetSpatialFilter()
    lyr.SetSpatialFilter(1, ogr.CreateGeometryFromWkt('POLYGON ((-10 10,-10 11,-11 11,-11 10,-10 10))'))
    lyr.ResetReading()
    feat = lyr.GetNextFeature()
    assert feat is None
    lyr.SetSpatialFilter(1, ogr.CreateGeometryFromWkt('POLYGON ((10 10,10 11,11 11,11 10,10 10))'))
    lyr.ResetReading()
    feat = lyr.GetNextFeature()
    assert feat is not None
    lyr.SetSpatialFilterRect(1, 10, 10, 11, 11)
    lyr.ResetReading()
    feat = lyr.GetNextFeature()
    assert feat is not None
    # Test invalid spatial filter index
    gdal.ErrorReset()
    gdal.PushErrorHandler('CPLQuietErrorHandler')
    lyr.SetSpatialFilterRect(2, 0, 0, 0, 0)
    gdal.PopErrorHandler()
    assert gdal.GetLastErrorMsg() != ''

    lyr.SetSpatialFilter(None)
    another_lyr = ds.CopyLayer(lyr, 'dup_test')
    dup_feat = another_lyr.GetNextFeature()
    geom = dup_feat.GetGeomFieldRef('a_name')
    assert geom.ExportToWkt() == 'POINT (1 2)'
    geom = dup_feat.GetGeomFieldRef('another_geom_field')
    assert geom.ExportToWkt() == 'POLYGON ((10 10,10 11,11 11,11 10,10 10))'

###############################################################################
# Test Python field accessors facilities


def test_ogr_rfc41_5():

    feature_defn = ogr.FeatureDefn()
    field_defn = ogr.FieldDefn('strfield', ogr.OFTString)
    feature_defn.AddFieldDefn(field_defn)
    feature_defn.GetGeomFieldDefn(0).SetName('geomfield')

    f = ogr.Feature(feature_defn)

    assert f['strfield'] is None
    assert f.strfield is None

    assert f['geomfield'] is None
    assert f.geomfield is None

    with pytest.raises(KeyError):
        f['nonexistent_field']
    

    with pytest.raises(AttributeError):
        f.nonexistent_field
    

    with pytest.raises(KeyError):
        f['nonexistent_field'] = 'foo'
    

    # This works.  Default Python behaviour. Stored in a dictionary
    f.nonexistent_field = 'bar'
    assert f.nonexistent_field == 'bar'

    f['strfield'] = 'foo'
    assert f['strfield'] == 'foo'
    assert f.strfield == 'foo'

    f.strfield = 'bar'
    assert f['strfield'] == 'bar'
    assert f.strfield == 'bar'

    wkt = 'POINT EMPTY'
    f['geomfield'] = ogr.CreateGeometryFromWkt(wkt)
    assert f['geomfield'].ExportToWkt() == wkt
    assert f.geomfield.ExportToWkt() == wkt

    wkt2 = 'POLYGON EMPTY'
    f.geomfield = ogr.CreateGeometryFromWkt(wkt2)
    assert f['geomfield'].ExportToWkt() == wkt2
    assert f.geomfield.ExportToWkt() == wkt2

###############################################################################
# Test OGRSQL with geometries


def test_ogr_rfc41_6():

    ds = ogr.GetDriverByName('memory').CreateDataSource('')
    sr = osr.SpatialReference()
    lyr = ds.CreateLayer('poly', geom_type=ogr.wkbPolygon, srs=sr)
    lyr.GetLayerDefn().GetGeomFieldDefn(0).SetName('geomfield')
    lyr.CreateField(ogr.FieldDefn('intfield', ogr.OFTInteger))
    lyr.CreateField(ogr.FieldDefn('wkt', ogr.OFTString))
    feat = ogr.Feature(lyr.GetLayerDefn())
    feat.SetField('intfield', 1)
    feat.SetField('wkt', 'POINT (0 0)')
    feat.SetGeometryDirectly(ogr.CreateGeometryFromWkt('POLYGON EMPTY'))
    lyr.CreateFeature(feat)
    feat = ogr.Feature(lyr.GetLayerDefn())
    lyr.CreateFeature(feat)
    feat = None

    # Test implicit geometry column (since poly has one single geometry column)
    # then explicit geometry column
    for sql in ['SELECT intfield FROM poly',
                'SELECT * FROM poly',
                'SELECT intfield, geomfield FROM poly',
                'SELECT geomfield, intfield FROM poly']:
        sql_lyr = ds.ExecuteSQL(sql)
        assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetType() == ogr.wkbPolygon
        assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetSpatialRef() is not None
        feat = sql_lyr.GetNextFeature()
        assert feat.GetField('intfield') == 1
        assert feat.GetGeomFieldRef('geomfield') is not None
        feat = sql_lyr.GetNextFeature()
        assert feat.GetGeomFieldRef('geomfield') is None
        feat = None
        ds.ReleaseResultSet(sql_lyr)

    # Test CAST(geometry_field AS GEOMETRY)
    sql_lyr = ds.ExecuteSQL('SELECT CAST(geomfield AS GEOMETRY) AS mygeom FROM poly WHERE CAST(geomfield AS GEOMETRY) IS NOT NULL')
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetType() == ogr.wkbUnknown
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetSpatialRef() is None
    feat = sql_lyr.GetNextFeature()
    assert feat.GetGeomFieldRef('mygeom') is not None
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test CAST(xxx AS GEOMETRY(POLYGON))
    sql_lyr = ds.ExecuteSQL('SELECT CAST(geomfield AS GEOMETRY(POLYGON)) AS mygeom FROM poly WHERE CAST(geomfield AS GEOMETRY(POLYGON)) IS NOT NULL')
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetType() == ogr.wkbPolygon
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetSpatialRef() is None
    feat = sql_lyr.GetNextFeature()
    assert feat.GetGeomFieldRef('mygeom') is not None
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test CAST(xxx AS GEOMETRY(POLYGON,4326))
    sql_lyr = ds.ExecuteSQL('SELECT CAST(geomfield AS GEOMETRY(POLYGON,4326)) AS mygeom FROM poly WHERE CAST(geomfield AS GEOMETRY(POLYGON,4326)) IS NOT NULL')
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetType() == ogr.wkbPolygon
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetSpatialRef().ExportToWkt().find('4326') >= 0
    feat = sql_lyr.GetNextFeature()
    assert feat.GetGeomFieldRef('mygeom') is not None
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test CAST(a_multipolygon AS GEOMETRY(POLYGON))
    sql_lyr = ds.ExecuteSQL("SELECT CAST('MULTIPOLYGON (((0 0,0 1,1 1,1 0,0 0)))' AS GEOMETRY(POLYGON)) AS mygeom FROM poly")
    feat = sql_lyr.GetNextFeature()
    assert feat.GetGeomFieldRef('mygeom').ExportToWkt() == 'POLYGON ((0 0,0 1,1 1,1 0,0 0))'
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test CAST(a_polygon AS GEOMETRY(MULTIPOLYGON))
    sql_lyr = ds.ExecuteSQL("SELECT CAST('POLYGON ((0 0,0 1,1 1,1 0,0 0))' AS GEOMETRY(MULTIPOLYGON)) AS mygeom FROM poly")
    feat = sql_lyr.GetNextFeature()
    assert feat.GetGeomFieldRef('mygeom').ExportToWkt() == 'MULTIPOLYGON (((0 0,0 1,1 1,1 0,0 0)))'
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test CAST(a_multilinestring AS GEOMETRY(LINESTRING))
    sql_lyr = ds.ExecuteSQL("SELECT CAST('MULTILINESTRING ((0 0,0 1,1 1,1 0,0 0))' AS GEOMETRY(LINESTRING)) AS mygeom FROM poly")
    feat = sql_lyr.GetNextFeature()
    assert feat.GetGeomFieldRef('mygeom').ExportToWkt() == 'LINESTRING (0 0,0 1,1 1,1 0,0 0)'
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test CAST(a_linestring AS GEOMETRY(MULTILINESTRING))
    sql_lyr = ds.ExecuteSQL("SELECT CAST('LINESTRING (0 0,0 1,1 1,1 0,0 0)' AS GEOMETRY(MULTILINESTRING)) AS mygeom FROM poly")
    feat = sql_lyr.GetNextFeature()
    assert feat.GetGeomFieldRef('mygeom').ExportToWkt() == 'MULTILINESTRING ((0 0,0 1,1 1,1 0,0 0))'
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test expression with cast CHARACTER <--> GEOMETRY
    sql_lyr = ds.ExecuteSQL('SELECT CAST(CAST(geomfield AS CHARACTER) AS GEOMETRY) AS mygeom, intfield FROM poly')
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetType() == ogr.wkbUnknown
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetSpatialRef() is None
    feat = sql_lyr.GetNextFeature()
    assert feat.GetField('intfield') == 1
    assert feat.GetGeomFieldRef('mygeom') is not None
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test CAST(NULL AS GEOMETRY)
    sql_lyr = ds.ExecuteSQL('SELECT CAST(NULL AS GEOMETRY) FROM poly')
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetType() == ogr.wkbUnknown
    feat = sql_lyr.GetNextFeature()
    assert feat.GetGeomFieldRef('') is None
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test CAST(stringfield AS GEOMETRY)
    sql_lyr = ds.ExecuteSQL('SELECT CAST(wkt AS GEOMETRY) FROM poly')
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetType() == ogr.wkbUnknown
    feat = sql_lyr.GetNextFeature()
    assert feat.GetGeomFieldRef('wkt').ExportToWkt() == 'POINT (0 0)'
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test COUNT(geometry)
    sql_lyr = ds.ExecuteSQL('SELECT COUNT(geomfield) FROM poly')
    feat = sql_lyr.GetNextFeature()
    assert feat is not None
    assert feat.GetField(0) == 1
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    wrong_sql_list = [
        ('SELECT DISTINCT geomfield FROM poly', 'SELECT DISTINCT on a geometry not supported'),
        ('SELECT COUNT(DISTINCT geomfield) FROM poly', 'SELECT COUNT DISTINCT on a geometry not supported'),
        ('SELECT MAX(geomfield) FROM poly', 'Use of field function MAX() on geometry field'),
        ('SELECT CAST(5 AS GEOMETRY) FROM poly', 'Cannot cast integer to geometry'),
        ('SELECT CAST(geomfield AS integer) FROM poly', 'Cannot cast geometry to integer'),
        ('SELECT CAST(geomfield AS GEOMETRY(2)) FROM poly', 'First argument of CAST operator should be a geometry type identifier'),
        ('SELECT CAST(geomfield AS GEOMETRY(UNSUPPORTED_TYPE)) FROM poly', 'SQL Expression Parsing Error: syntax error'),
        ('SELECT CAST(geomfield AS GEOMETRY(UNSUPPORTED_TYPE,5)) FROM poly', 'SQL Expression Parsing Error: syntax error'),
    ]

    for (sql, error_msg) in wrong_sql_list:
        gdal.ErrorReset()
        gdal.PushErrorHandler('CPLQuietErrorHandler')
        sql_lyr = ds.ExecuteSQL(sql)
        gdal.PopErrorHandler()
        assert gdal.GetLastErrorMsg().find(error_msg) == 0, \
            ('For %s, expected error %s, got %s' % (sql, error_msg, gdal.GetLastErrorMsg()))
        assert sql_lyr is None

    # Test invalid expressions with geometry
    for sql in ["SELECT geomfield + 'a' FROM poly",
                "SELECT geomfield * 'a' FROM poly",
                "SELECT geomfield + 'a' FROM poly",
                "SELECT geomfield - 'a' FROM poly",
                "SELECT geomfield % 'a' FROM poly",
                "SELECT CONCAT(geomfield, 'a') FROM poly",
                "SELECT SUBSTR(geomfield, 0, 1) FROM poly",
                "SELECT * FROM poly WHERE geomfield = CAST('POINT EMPTY' AS GEOMETRY)",
                "SELECT * FROM poly WHERE geomfield LIKE 'a'",
                "SELECT * FROM poly WHERE geomfield IN( 'a' )"]:
        gdal.ErrorReset()
        gdal.PushErrorHandler('CPLQuietErrorHandler')
        sql_lyr = ds.ExecuteSQL(sql)
        gdal.PopErrorHandler()
        assert gdal.GetLastErrorMsg().find('Cannot use geometry field in this operation') == 0
        assert sql_lyr is None

    # Test expression with geometry in WHERE
    sql_lyr = ds.ExecuteSQL('SELECT * FROM poly WHERE geomfield IS NOT NULL')
    feat = sql_lyr.GetNextFeature()
    assert feat.GetField('intfield') == 1
    feat = sql_lyr.GetNextFeature()
    assert feat is None
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    sql_lyr = ds.ExecuteSQL('SELECT * FROM poly WHERE geomfield IS NULL')
    feat = sql_lyr.GetNextFeature()
    assert not feat.IsFieldSet(0)
    feat = sql_lyr.GetNextFeature()
    assert feat is None
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    sql_lyr = ds.ExecuteSQL("SELECT * FROM poly WHERE CAST(geomfield AS CHARACTER) = 'POLYGON EMPTY'")
    feat = sql_lyr.GetNextFeature()
    assert feat is not None
    feat = sql_lyr.GetNextFeature()
    assert feat is None
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    sql_lyr = ds.ExecuteSQL('SELECT count(*) FROM poly WHERE geomfield IS NULL')
    feat = sql_lyr.GetNextFeature()
    assert feat.GetField(0) == 1
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    sql_lyr = ds.ExecuteSQL('SELECT count(*) FROM poly WHERE geomfield IS NOT NULL')
    feat = sql_lyr.GetNextFeature()
    assert feat.GetField(0) == 1
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test spatial filter
    feat = lyr.GetFeature(0)
    feat.SetGeometryDirectly(ogr.CreateGeometryFromWkt('POINT(1 2)'))
    lyr.SetFeature(feat)
    feat = None

    lyr.DeleteFeature(1)

    sql_lyr = ds.ExecuteSQL("SELECT * FROM poly")
    sql_lyr.SetSpatialFilterRect(0, 0, 0, 0)
    feat = sql_lyr.GetNextFeature()
    assert feat is None
    feat = None

    sql_lyr.SetSpatialFilterRect(0, 1, 2, 1, 2)
    feat = sql_lyr.GetNextFeature()
    assert feat is not None
    feat = None

    # Test invalid spatial filter index
    gdal.ErrorReset()
    gdal.PushErrorHandler('CPLQuietErrorHandler')
    sql_lyr.SetSpatialFilterRect(2, 0, 0, 0, 0)
    gdal.PopErrorHandler()
    assert gdal.GetLastErrorMsg() != ''

    # Test invalid geometry field index
    gdal.ErrorReset()
    gdal.PushErrorHandler('CPLQuietErrorHandler')
    sql_lyr.GetExtent(geom_field=2)
    gdal.PopErrorHandler()
    assert gdal.GetLastErrorMsg() != ''

    ds.ReleaseResultSet(sql_lyr)

    # Test querying several geometry fields
    sql_lyr = ds.ExecuteSQL('SELECT geomfield as geom1, geomfield as geom2 FROM poly')
    feat = sql_lyr.GetNextFeature()
    assert feat is not None
    assert feat.GetGeomFieldRef('geom1') is not None
    assert feat.GetGeomFieldRef('geom2') is not None
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test querying a layer with several geometry fields
    lyr.CreateGeomField(ogr.GeomFieldDefn('secondarygeom', ogr.wkbPoint))
    lyr.ResetReading()
    feat = lyr.GetNextFeature()
    feat.SetGeomField('secondarygeom', ogr.CreateGeometryFromWkt('POINT (10 100)'))
    lyr.SetFeature(feat)
    feat = None

    for sql in ['SELECT * FROM poly',
                'SELECT geomfield, secondarygeom FROM poly',
                'SELECT secondarygeom, geomfield FROM poly']:
        sql_lyr = ds.ExecuteSQL(sql)
        feat = sql_lyr.GetNextFeature()
        assert feat.GetGeomFieldRef('geomfield').ExportToWkt() == 'POINT (1 2)'
        assert feat.GetGeomFieldRef('secondarygeom').ExportToWkt() == 'POINT (10 100)'
        feat = None
        ds.ReleaseResultSet(sql_lyr)

    # Check that we don't get an implicit geometry field
    sql_lyr = ds.ExecuteSQL('SELECT intfield FROM poly')
    assert sql_lyr.GetLayerDefn().GetGeomFieldCount() == 0
    ds.ReleaseResultSet(sql_lyr)

    # Check GetExtent() and SetSpatialFilter()
    sql_lyr = ds.ExecuteSQL('SELECT * FROM poly')
    assert sql_lyr.GetExtent(geom_field=0) == (1.0, 1.0, 2.0, 2.0)
    assert sql_lyr.GetExtent(geom_field=1) == (10.0, 10.0, 100.0, 100.0)
    sql_lyr.SetSpatialFilterRect(0, 0.5, 1.5, 1.5, 2.5)
    assert sql_lyr.GetFeatureCount() == 1
    sql_lyr.SetSpatialFilterRect(0, 0, 0, 0.5, 0.5)
    assert sql_lyr.GetFeatureCount() == 0
    sql_lyr.SetSpatialFilterRect(1, 9, 99, 11, 101)
    assert sql_lyr.GetFeatureCount() == 1
    sql_lyr.SetSpatialFilterRect(1, 0, 0, 0.5, 0.5)
    assert sql_lyr.GetFeatureCount() == 0
    ds.ReleaseResultSet(sql_lyr)

    ds = None

###############################################################################
# Test crazy OGRSQL


def test_ogr_rfc41_7():

    ds = ogr.Open('data')
    sql = "select eas_id, \"_ogr_geometry_\" as geom1, cast(null as geometry) as geom2, " + \
          "'a', cast('POINT(3 4)' as geometry) as geom3, fid, \"_ogr_geometry_\" as geom4, " +\
          "'c', p.eas_id, cast(area as integer) as area_int, \"_ogr_geometry_\", area from " +\
          "poly join \"data\".poly p on poly.eas_id = p.eas_id"
    sql_lyr = ds.ExecuteSQL(sql)
    feat = sql_lyr.GetNextFeature()
    if feat.eas_id != 168 or \
       feat.FIELD_2 != 'a' or \
       feat.fid != 0 or \
       feat.FIELD_4 != 'c' or \
       feat['p.eas_id'] != 168 or \
       feat.area_int != 215229 or \
       feat.area != pytest.approx(215229.266, abs=1e-5) or \
       feat.geom1.GetGeometryType() != ogr.wkbPolygon or \
       feat.geom2 is not None or \
       feat.geom3.GetGeometryType() != ogr.wkbPoint or \
       feat.geom4.GetGeometryType() != ogr.wkbPolygon or \
       feat['_ogr_geometry_'].GetGeometryType() != ogr.wkbPolygon:
        feat.DumpReadable()
        pytest.fail()
    ds.ReleaseResultSet(sql_lyr)

    ds = None

###############################################################################
# Test SQLite dialect


def test_ogr_rfc41_8(require_ogr_sql_sqlite):  # noqa
    ds = ogr.GetDriverByName('memory').CreateDataSource('')
    lyr = ds.CreateLayer('mytable', geom_type=ogr.wkbPolygon)
    lyr.GetLayerDefn().GetGeomFieldDefn(0).SetName('geomfield')
    gfld_defn = ogr.GeomFieldDefn('geomfield2', ogr.wkbPoint25D)
    sr = osr.SpatialReference()
    sr.ImportFromEPSG(4326)
    gfld_defn.SetSpatialRef(sr)
    lyr.CreateGeomField(gfld_defn)

    # Check that we get the geometry columns, even with no features
    sql_lyr = ds.ExecuteSQL('SELECT * FROM mytable', dialect='SQLite')
    assert sql_lyr.GetLayerDefn().GetGeomFieldCount() == 2
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetType() == ogr.wkbPolygon
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(0).GetSpatialRef() is None
    assert sql_lyr.GetLayerDefn().GetGeomFieldDefn(1).GetType() == ogr.wkbPoint25D
    srs = sql_lyr.GetLayerDefn().GetGeomFieldDefn(1).GetSpatialRef()
    assert srs.GetAuthorityCode(None) == '4326'
    ds.ReleaseResultSet(sql_lyr)

    # Test INSERT INTO request
    ds.ExecuteSQL("INSERT INTO mytable (geomfield, geomfield2) VALUES (" +
                  "GeomFromText('POLYGON ((0 0,0 1,1 1,1 0,0 0))'), " +
                  "GeomFromText('POINT Z(0 1 2)') )", dialect='SQLite')

    # Check output
    sql_lyr = ds.ExecuteSQL('SELECT geomfield2, geomfield FROM mytable', dialect='SQLite')
    feat = sql_lyr.GetNextFeature()
    geom = feat.GetGeomFieldRef('geomfield')
    if geom.ExportToWkt() != 'POLYGON ((0 0,0 1,1 1,1 0,0 0))':
        feat.DumpReadable()
        pytest.fail()
    geom = feat.GetGeomFieldRef('geomfield2')
    if geom.ExportToWkt() != 'POINT (0 1 2)':
        feat.DumpReadable()
        pytest.fail()
    feat = None
    ds.ReleaseResultSet(sql_lyr)

    # Test UPDATE
    ds.ExecuteSQL("UPDATE mytable SET geomfield2 = " +
                  "GeomFromText('POINT Z(3 4 5)')", dialect='SQLite')

    # Check output
    sql_lyr = ds.ExecuteSQL('SELECT geomfield2, geomfield FROM mytable', dialect='SQLite')
    feat = sql_lyr.GetNextFeature()
    geom = feat.GetGeomFieldRef('geomfield')
    if geom.ExportToWkt() != 'POLYGON ((0 0,0 1,1 1,1 0,0 0))':
        feat.DumpReadable()
        pytest.fail()
    geom = feat.GetGeomFieldRef('geomfield2')
    if geom.ExportToWkt() != 'POINT (3 4 5)':
        feat.DumpReadable()
        pytest.fail()
    feat = None
    ds.ReleaseResultSet(sql_lyr)



