#!/usr/bin/env pytest # -*- coding: utf-8 -*- ############################################################################### # $Id$ # # Project: GDAL/OGR Test Suite # Purpose: CSW driver testing. # Author: Even Rouault # ############################################################################### # Copyright (c) 2015, Even Rouault # # 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 pytest from osgeo import gdal, ogr ############################################################################### @pytest.fixture(autouse=True, scope="module") def module_disable_exceptions(): with gdaltest.disable_exceptions(): yield ############################################################################### @pytest.fixture(autouse=True, scope="module") def setup_and_cleanup(): with gdal.config_option("CPL_CURL_ENABLE_VSIMEM", "YES"): yield ############################################################################### # Test underlying OGR drivers # pytestmark = pytest.mark.require_driver("CSW") @pytest.fixture(autouse=True, scope="module") def ogr_csw_init(): gml_ds = ogr.Open("data/gml/ionic_wfs.gml") if gml_ds is None: if gdal.GetLastErrorMsg().find("Xerces") != -1: pytest.skip() pytest.skip("failed to open test file.") ############################################################################### # Test reading a pyCSW server @pytest.mark.skip() def test_ogr_csw_pycsw(): ds = ogr.Open("CSW:http://catalog.data.gov/csw") if ds is None: if gdaltest.gdalurlopen("http://catalog.data.gov/csw") is None: pytest.skip("cannot open URL") pytest.skip("did not managed to open CSW datastore") lyr = ds.GetLayer(0) f = lyr.GetNextFeature() assert f is not None, "did not get expected layer name" ############################################################################### def test_ogr_csw_vsimem_fail_because_not_enabled(): with gdal.config_option("CPL_CURL_ENABLE_VSIMEM", "NO"): ds = ogr.Open("CSW:/vsimem/csw_endpoint") assert ds is None ############################################################################### def test_ogr_csw_vsimem_fail_because_no_get_capabilities(): ds = ogr.Open("CSW:/vsimem/csw_endpoint") assert ds is None ############################################################################### def test_ogr_csw_vsimem_fail_because_empty_response(): gdal.FileFromMemBuffer( "/vsimem/csw_endpoint?SERVICE=CSW&REQUEST=GetCapabilities", "" ) with gdaltest.error_handler(): ds = ogr.Open("CSW:/vsimem/csw_endpoint") assert ds is None assert gdal.GetLastErrorMsg().find("Empty content returned by server") >= 0 ############################################################################### def test_ogr_csw_vsimem_fail_because_no_CSW_Capabilities(): gdal.FileFromMemBuffer( "/vsimem/csw_endpoint?SERVICE=CSW&REQUEST=GetCapabilities", "" ) with gdaltest.error_handler(): ds = ogr.Open("CSW:/vsimem/csw_endpoint") assert ds is None assert gdal.GetLastErrorMsg().find("Cannot find Capabilities.version") >= 0 ############################################################################### def test_ogr_csw_vsimem_fail_because_exception(): gdal.FileFromMemBuffer( "/vsimem/csw_endpoint?SERVICE=CSW&REQUEST=GetCapabilities", "", ) with gdaltest.error_handler(): ds = ogr.Open("CSW:/vsimem/csw_endpoint") assert ds is None assert ( gdal.GetLastErrorMsg().find( "Error returned by server : " ) >= 0 ) ############################################################################### def test_ogr_csw_vsimem_fail_because_invalid_xml_capabilities(): gdal.FileFromMemBuffer( "/vsimem/csw_endpoint?SERVICE=CSW&REQUEST=GetCapabilities", "= 0 ############################################################################### def test_ogr_csw_vsimem_fail_because_missing_version(): gdal.FileFromMemBuffer( "/vsimem/csw_endpoint?SERVICE=CSW&REQUEST=GetCapabilities", """ """, ) with gdaltest.error_handler(): ds = ogr.Open("CSW:/vsimem/csw_endpoint") assert ds is None assert gdal.GetLastErrorMsg().find("Cannot find Capabilities.version") >= 0 ############################################################################### def test_ogr_csw_vsimem_csw_minimal_instance(): # Invalid response, but enough for use gdal.FileFromMemBuffer( "/vsimem/csw_endpoint?SERVICE=CSW&REQUEST=GetCapabilities", """ """, ) ds = ogr.Open("CSW:/vsimem/csw_endpoint") assert ds is not None ds.TestCapability("foo") assert ds.GetLayerCount() == 1 assert ds.GetLayer(-1) is None and ds.GetLayer(1) is None lyr = ds.GetLayer(0) lyr.TestCapability("foo") with gdaltest.error_handler(): f = lyr.GetNextFeature() assert f is None with gdaltest.error_handler(): fc = lyr.GetFeatureCount() assert fc == 0 gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """""", ) lyr.ResetReading() with gdaltest.error_handler(): f = lyr.GetNextFeature() assert ( f is None and gdal.GetLastErrorMsg().find("Empty content returned by server") >= 0 ) gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """= 0 gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """""", ) lyr.ResetReading() with gdaltest.error_handler(): f = lyr.GetNextFeature() assert f is None and gdal.GetLastErrorMsg().find("Error: cannot parse") >= 0 gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """""", ) lyr.ResetReading() with gdaltest.error_handler(): f = lyr.GetNextFeature() assert f is None and gdal.GetLastErrorMsg().find("Error returned by server") >= 0 gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """ an_identifier another_identifier a_title dataset a_subject another_subject http://foo/ http://bar/ 2015-04-27 an_abstract 2015-04-27 eng a_format another_format -90 -180 90 180 """, ) lyr.ResetReading() f = lyr.GetNextFeature() assert f is not None if ( f["identifier"] != "an_identifier" or f["other_identifiers"] != ["another_identifier"] or f["subject"] != "a_subject" or f["other_subjects"] != ["another_subject"] or f["references"] != "http://foo/" or f["other_references"] != ["http://bar/"] or f["format"] != "a_format" or f["other_formats"] != ["another_format"] or f["boundingbox"].ExportToWkt() != "POLYGON ((-180 -90,-180 90,180 90,180 -90,-180 -90))" ): f.DumpReadable() pytest.fail() f = lyr.GetNextFeature() assert f is not None with gdaltest.error_handler(): f = lyr.GetNextFeature() assert f is None with gdaltest.error_handler(): fc = lyr.GetFeatureCount() assert fc == 2 lyr.ResetReading() f = lyr.GetNextFeature() assert f is not None gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """ """, ) f = lyr.GetNextFeature() assert f is not None f = lyr.GetNextFeature() assert f is not None gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """ """, ) f = lyr.GetNextFeature() assert f is None gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """""", ) with gdaltest.error_handler(): fc = lyr.GetFeatureCount() assert fc == 3, gdal.GetLastErrorMsg() gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """""", ) with gdaltest.error_handler(): fc = lyr.GetFeatureCount() assert fc == 3, gdal.GetLastErrorMsg() gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """""", ) with gdaltest.error_handler(): fc = lyr.GetFeatureCount() assert fc == 3, gdal.GetLastErrorMsg() gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """""", ) with gdaltest.error_handler(): fc = lyr.GetFeatureCount() assert fc == 3, gdal.GetLastErrorMsg() gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """ """, ) fc = lyr.GetFeatureCount() assert fc == 200 lyr.SetAttributeFilter("identifier = 'an_identifier'") lyr.SetSpatialFilterRect(-180, -90, 180, 90) gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=fullows:BoundingBox-90 -18090 180dc:identifieran_identifier""", """ """, ) f = lyr.GetNextFeature() assert f is not None gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=fullows:BoundingBox-90 -18090 180dc:identifieran_identifier""", """ """, ) fc = lyr.GetFeatureCount() assert fc == 300 lyr.SetAttributeFilter( "identifier = 'an_identifier' AND " + "references = 'http://foo/' AND " + "anytext LIKE '%%foo%%' AND " + "other_identifiers = '' AND " + "other_subjects = '' AND " + "other_formats = '' AND " + "other_references = '' AND " + "ST_Intersects(boundingbox, ST_MakeEnvelope(2,49,2,49,4326))" ) lyr.SetAttributeFilter(None) lyr.SetSpatialFilter(None) ############################################################################### def test_ogr_csw_vsimem_csw_output_schema_csw(): ds = gdal.OpenEx("CSW:/vsimem/csw_endpoint", open_options=["OUTPUT_SCHEMA=CSW"]) lyr = ds.GetLayer(0) gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """full""", """""", ) lyr.ResetReading() with gdaltest.error_handler(): f = lyr.GetNextFeature() assert f is None gdal.FileFromMemBuffer( """/vsimem/csw_endpoint&POSTFIELDS=full""", """ -90 -180 90 180 """, ) lyr.ResetReading() f = lyr.GetNextFeature() if ( f["raw_xml"].find("full""", """ -180 180 -90 90 """, ) f = lyr.GetNextFeature() if ( f["raw_xml"].find("full""", """ -180 180 90 -90 """, ) f = lyr.GetNextFeature() if ( f["raw_xml"].find("