#!/usr/bin/env pytest
###############################################################################
# $Id: ogr_sde.py 30250f784292af8d1f0ae95625b06ac482885d86 2018-12-14 10:16:24 +1300 Craig de Stigter $
#
# Project:  GDAL/OGR Test Suite
# Purpose:  Test OGR ArcSDE driver.
# Author:   Howard Butler <hobu.inc@gmail.com>
#
###############################################################################
# Copyright (c) 2008, Howard Butler <hobu.inc@gmail.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.
###############################################################################



import gdaltest
import ogrtest

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

###############################################################################
# Open ArcSDE datasource.

sde_server = '172.16.1.193'
sde_port = '5151'
sde_db = 'sde'
sde_user = 'sde'
sde_password = 'sde'

gdaltest.sde_dr = None
try:
    gdaltest.sde_dr = ogr.GetDriverByName('SDE')
except:
    pass


def test_ogr_sde_1():
    "Test basic opening of a database"

    if gdaltest.sde_dr is None:
        pytest.skip()

    base = 'SDE:%s,%s,%s,%s,%s' % (sde_server, sde_port, sde_db, sde_user, sde_password)
    ds = ogr.Open(base)
    if ds is None:
        gdaltest.sde_dr = None
        pytest.skip("Could not open %s" % base)
    ds.Destroy()

    ds = ogr.Open(base, update=1)
    ds.Destroy()


def test_ogr_sde_2():
    "Test creation of a layer"
    if gdaltest.sde_dr is None:
        pytest.skip()
    base = 'SDE:%s,%s,%s,%s,%s' % (sde_server, sde_port, sde_db, sde_user, sde_password)

    shp_ds = ogr.Open('data/poly.shp')
    gdaltest.shp_ds = shp_ds
    shp_lyr = shp_ds.GetLayer(0)

    ds = ogr.Open(base, update=1)
    lyr = ds.CreateLayer('SDE.TPOLY', geom_type=ogr.wkbPolygon, srs=shp_lyr.GetSpatialRef(), options=['OVERWRITE=YES'])
#    lyr = ds.CreateLayer( 'SDE.TPOLY' ,geom_type=ogr.wkbPolygon)

    ogrtest.quick_create_layer_def(lyr,
                                   [('AREA', ogr.OFTReal),
                                    ('EAS_ID', ogr.OFTInteger),
                                    ('PRFEDEA', ogr.OFTString),
                                    ('WHEN', ogr.OFTDateTime)])

    #######################################################
    # Copy in poly.shp

    dst_feat = ogr.Feature(feature_def=lyr.GetLayerDefn())

    feat = shp_lyr.GetNextFeature()
    gdaltest.poly_feat = []

    while feat is not None:

        gdaltest.poly_feat.append(feat)

        dst_feat.SetFrom(feat)
        lyr.CreateFeature(dst_feat)

        feat = shp_lyr.GetNextFeature()

    dst_feat.Destroy()


def test_ogr_sde_3():
    "Test basic version locking"
    if gdaltest.sde_dr is None:
        pytest.skip()

    base = 'SDE:%s,%s,%s,%s,%s,SDE.TPOLY,SDE.DEFAULT' % (sde_server, sde_port, sde_db, sde_user, sde_password)
    ds = ogr.Open(base, update=1)

    ds2 = ogr.Open(base, update=1)
    assert ds2 is None, 'A locked version was able to be opened'

    ds.Destroy()


def test_ogr_sde_4():
    "Test basic version creation"

    if gdaltest.sde_dr is None:
        pytest.skip()
    version_name = 'TESTING'
    gdal.SetConfigOption('SDE_VERSIONOVERWRITE', 'TRUE')

    base = 'SDE:%s,%s,%s,%s,%s,SDE.TPOLY,SDE.DEFAULT,%s' % (sde_server, sde_port, sde_db, sde_user, sde_password, version_name)
    ds = ogr.Open(base, update=1)
    ds.Destroy()

    gdal.SetConfigOption('SDE_VERSIONOVERWRITE', 'FALSE')

    base = 'SDE:%s,%s,%s,%s,%s,SDE.TPOLY,SDE.DEFAULT,%s' % (sde_server, sde_port, sde_db, sde_user, sde_password, version_name)
    ds = ogr.Open(base, update=1)
    ds.Destroy()


def test_ogr_sde_5():
    "Test versioned editing"

    if gdaltest.sde_dr is None:
        pytest.skip()
    version_name = 'TESTING'
    gdal.SetConfigOption('SDE_VERSIONOVERWRITE', 'TRUE')

    base = 'SDE:%s,%s,%s,%s,%s,SDE.TPOLY,SDE.DEFAULT,%s' % (sde_server, sde_port, sde_db, sde_user, sde_password, version_name)
    ds = ogr.Open(base, update=1)

    l1 = ds.GetLayerByName('SDE.TPOLY')

    f1 = l1.GetFeature(1)
    f1.SetField("PRFEDEA", 'SDE.TESTING')
    l1.SetFeature(f1)

    ds.Destroy()
    del ds

    default = 'DEFAULT'
    gdal.SetConfigOption('SDE_VERSIONOVERWRITE', 'FALSE')

    default = 'SDE:%s,%s,%s,%s,%s,SDE.TPOLY,SDE.DEFAULT,%s' % (sde_server, sde_port, sde_db, sde_user, sde_password, default)
#    print default
    ds2 = ogr.Open(default, update=1)

    l2 = ds2.GetLayerByName('SDE.TPOLY')

    f2 = l2.GetFeature(1)

    f2.SetField("PRFEDEA", 'SDE.DEFAULT')
    f2.SetField("WHEN", 2008, 3, 19, 16, 15, 00, 0)

    l2.SetFeature(f2)
    ds2.Destroy()
    del ds2

    ds3 = ogr.Open(base)
    l3 = ds3.GetLayerByName('SDE.TPOLY')
    f3 = l3.GetFeature(1)
    assert f3.GetField("PRFEDEA") == "SDE.TESTING", \
        'versioned editing failed for child version SDE.TESTING'

    ds3.Destroy()
    del ds3

    ds4 = ogr.Open(default)
    l4 = ds4.GetLayerByName('SDE.TPOLY')
    f4 = l4.GetFeature(1)
    assert f4.GetField("PRFEDEA") == "SDE.DEFAULT", \
        'versioned editing failed for parent version SDE.DEFAULT'

    idx = f4.GetFieldIndex('WHEN')
    df = f4.GetField(idx)
    if df != '2008/03/19 16:15:00':
        gdaltest.post_reason("datetime handling did not work -- expected '2008/03/19 16:15:00' got '%s' " % df)
    ds4.Destroy()
    del ds4


def test_ogr_sde_6():
    "Extent fetching"

    if gdaltest.sde_dr is None:
        pytest.skip()

    base = 'SDE:%s,%s,%s,%s,%s,SDE.TPOLY,SDE.DEFAULT' % (
        sde_server, sde_port, sde_db, sde_user, sde_password)
    ds = ogr.Open(base, update=1)

    l1 = ds.GetLayerByName('SDE.TPOLY')
    extent = l1.GetExtent(force=0)
    if extent != (0.0, 2147483645.0, 0.0, 2147483645.0):
        gdaltest.post_reason("unforced extent did not equal expected value")

    extent = l1.GetExtent(force=1)
    if extent != (478316.0, 481645.0, 4762881.0, 4765611.0):
        gdaltest.post_reason("forced extent did not equal expected value")
    

def test_ogr_sde_7():
    "Bad layer test"

    if gdaltest.sde_dr is None:
        pytest.skip()

    base = 'SDE:%s,%s,%s,%s,%s,SDE.TPOLY,SDE.DEFAULT' % (
        sde_server, sde_port, sde_db, sde_user, sde_password)
    ds = ogr.Open(base, update=1)

    l1 = ds.GetLayerByName('SDE.TPOLY2')
    if l1:
        gdaltest.post_reason("we got a layer when we should not have")

    ds.Destroy()

    default = 'DEFAULT'
    gdal.SetConfigOption('SDE_VERSIONOVERWRITE', 'FALSE')

    default = 'SDE:%s,%s,%s,%s,%s,SDE.TPOLY,SDE.DEFAULT,%s' % (
        sde_server, sde_port, sde_db, sde_user, sde_password, default)
    ds = ogr.Open(default, update=1)

    l1 = ds.GetLayerByName('SDE.TPOLY2')
    if l1:
        gdaltest.post_reason("we got a layer when we should not have")
    ds.Destroy()

    default = 'DEFAULT'
    gdal.SetConfigOption('SDE_VERSIONOVERWRITE', 'FALSE')

    default = 'SDE:%s,%s,%s,%s,%s' % (
        sde_server, sde_port, sde_db, sde_user, sde_password)
    ds = ogr.Open(default)

    l1 = ds.GetLayerByName('SDE.TPOLY2')
    if l1:
        gdaltest.post_reason("we got a layer when we should not have")
    ds.Destroy()


def test_ogr_sde_8():
    "Test spatial references"
    if gdaltest.sde_dr is None:
        pytest.skip()
    base = 'SDE:%s,%s,%s,%s,%s' % (sde_server, sde_port, sde_db, sde_user, sde_password)

    shp_ds = ogr.Open('data/poly.shp')
    gdaltest.shp_ds = shp_ds
    shp_lyr = shp_ds.GetLayer(0)

    ref = osr.SpatialReference()
    ref.ImportFromWkt('LOCAL_CS["IMAGE"]')

    ds = ogr.Open(base, update=1)
    lyr = ds.CreateLayer('SDE.TPOLY', geom_type=ogr.wkbPolygon, srs=ref, options=['OVERWRITE=YES'])
    ref.ImportFromEPSG(4326)
    lyr = ds.CreateLayer('SDE.TPOLY', geom_type=ogr.wkbPolygon, srs=ref, options=['OVERWRITE=YES'])
    ogrtest.quick_create_layer_def(lyr,
                                   [('AREA', ogr.OFTReal),
                                    ('EAS_ID', ogr.OFTInteger),
                                    ('PRFEDEA', ogr.OFTString),
                                    ('WHEN', ogr.OFTDateTime)])

    #######################################################
    # Copy in poly.shp

    dst_feat = ogr.Feature(feature_def=lyr.GetLayerDefn())

    feat = shp_lyr.GetNextFeature()
    gdaltest.poly_feat = []

    while feat is not None:

        gdaltest.poly_feat.append(feat)

        dst_feat.SetFrom(feat)
        lyr.CreateFeature(dst_feat)

        feat = shp_lyr.GetNextFeature()

    dst_feat.Destroy()


def test_ogr_sde_cleanup():
    if gdaltest.sde_dr is None:
        pytest.skip()
    base = 'SDE:%s,%s,%s,%s,%s' % (sde_server, sde_port, sde_db, sde_user, sde_password)
    ds = ogr.Open(base, update=1)
    ds.DeleteLayer('%s.%s' % (sde_user.upper(), 'TPOLY'))
    ds.Destroy()



