""" *************************************************************************** GdalAlgorithmTests.py --------------------- Date : January 2016 Copyright : (C) 2016 by Matthias Kuhn Email : matthias@opengis.ch *************************************************************************** * * * This program is free software; you can redistribute it and/or modify * * it under the terms of the GNU General Public License as published by * * the Free Software Foundation; either version 2 of the License, or * * (at your option) any later version. * * * *************************************************************************** """ __author__ = "Matthias Kuhn" __date__ = "January 2016" __copyright__ = "(C) 2016, Matthias Kuhn" import nose2 import os import shutil import tempfile from qgis.core import ( QgsProcessingContext, QgsProcessingFeedback, QgsCoordinateReferenceSystem, QgsApplication, QgsFeature, QgsGeometry, QgsPointXY, QgsProject, QgsVectorLayer, QgsRectangle, QgsProjUtils, QgsProcessingException, QgsProcessingFeatureSourceDefinition, ) from qgis.testing import QgisTestCase, start_app from processing.algs.gdal.GdalUtils import GdalUtils from processing.algs.gdal.ogr2ogr import ogr2ogr from processing.algs.gdal.OgrToPostGis import OgrToPostGis testDataPath = os.path.join(os.path.dirname(__file__), "testdata") class TestGdalAlgorithms(QgisTestCase): @classmethod def setUpClass(cls): start_app() from processing.core.Processing import Processing Processing.initialize() cls.cleanup_paths = [] @classmethod def tearDownClass(cls): for path in cls.cleanup_paths: shutil.rmtree(path) def testCommandName(self): # Test that algorithms report a valid commandName p = QgsApplication.processingRegistry().providerById("gdal") for a in p.algorithms(): if a.id() in ("gdal:buildvirtualvector"): # build virtual vector is an exception continue self.assertTrue(a.commandName(), f"Algorithm {a.id()} has no commandName!") def testCommandNameInTags(self): # Test that algorithms commandName is present in provided tags p = QgsApplication.processingRegistry().providerById("gdal") for a in p.algorithms(): if not a.commandName(): continue self.assertTrue( a.commandName() in a.tags(), f"Algorithm {a.id()} commandName not found in tags!", ) def testNoParameters(self): # Test that algorithms throw QgsProcessingException and not base Python # exceptions when no parameters specified p = QgsApplication.processingRegistry().providerById("gdal") context = QgsProcessingContext() feedback = QgsProcessingFeedback() for a in p.algorithms(): try: a.getConsoleCommands({}, context, feedback) except QgsProcessingException: pass def testGetOgrCompatibleSourceFromMemoryLayer(self): # create a memory layer and add to project and context layer = QgsVectorLayer( "Point?field=fldtxt:string&field=fldint:integer", "testmem", "memory" ) self.assertTrue(layer.isValid()) pr = layer.dataProvider() f = QgsFeature() f.setAttributes(["test", 123]) f.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(100, 200))) f2 = QgsFeature() f2.setAttributes(["test2", 457]) f2.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(100, 200))) self.assertTrue(pr.addFeatures([f, f2])) self.assertEqual(layer.featureCount(), 2) QgsProject.instance().addMapLayer(layer) context = QgsProcessingContext() context.setProject(QgsProject.instance()) alg = QgsApplication.processingRegistry().createAlgorithmById( "gdal:buffervectors" ) self.assertIsNotNone(alg) parameters = {"INPUT": "testmem"} feedback = QgsProcessingFeedback() # check that memory layer is automatically saved out to geopackage when required by GDAL algorithms input_details = alg.getOgrCompatibleSource( "INPUT", parameters, context, feedback, executing=True ) self.assertTrue(input_details.connection_string) self.assertTrue(input_details.connection_string.endswith(".gpkg")) self.assertTrue(os.path.exists(input_details.connection_string)) self.assertTrue(input_details.connection_string) # make sure that layer has correct features res = QgsVectorLayer(input_details.connection_string, "res") self.assertTrue(res.isValid()) self.assertEqual(res.featureCount(), 2) # with memory layers - if not executing layer source should be ignored and replaced # with a dummy path, because: # - it has no meaning for the gdal command outside of QGIS, memory layers don't exist! # - we don't want to force an export of the whole memory layer to a temp file just to show the command preview # this might be very slow! input_details = alg.getOgrCompatibleSource( "INPUT", parameters, context, feedback, executing=False ) self.assertEqual(input_details.connection_string, "path_to_data_file") self.assertEqual(input_details.layer_name, "layer_name") QgsProject.instance().removeMapLayer(layer) def testGetOgrCompatibleSourceFromOgrLayer(self): p = QgsProject() source = os.path.join(testDataPath, "points.gml") vl = QgsVectorLayer(source) self.assertTrue(vl.isValid()) p.addMapLayer(vl) context = QgsProcessingContext() context.setProject(p) feedback = QgsProcessingFeedback() alg = ogr2ogr() alg.initAlgorithm() input_details = alg.getOgrCompatibleSource( "INPUT", {"INPUT": vl.id()}, context, feedback, True ) self.assertEqual(input_details.connection_string, source) input_details = alg.getOgrCompatibleSource( "INPUT", {"INPUT": vl.id()}, context, feedback, False ) self.assertEqual(input_details.connection_string, source) # with selected features only - if not executing, the 'selected features only' setting # should be ignored (because it has no meaning for the gdal command outside of QGIS!) parameters = {"INPUT": QgsProcessingFeatureSourceDefinition(vl.id(), True)} input_details = alg.getOgrCompatibleSource( "INPUT", parameters, context, feedback, False ) self.assertEqual(input_details.connection_string, source) # with subset string vl.setSubsetString("x") input_details = alg.getOgrCompatibleSource( "INPUT", parameters, context, feedback, False ) self.assertEqual(input_details.connection_string, source) # subset of layer must be exported input_details = alg.getOgrCompatibleSource( "INPUT", parameters, context, feedback, True ) self.assertNotEqual(input_details.connection_string, source) self.assertTrue(input_details.connection_string) self.assertTrue(input_details.connection_string.endswith(".gpkg")) self.assertTrue(os.path.exists(input_details.connection_string)) self.assertTrue(input_details.layer_name) # geopackage with layer source = os.path.join(testDataPath, "custom", "circular_strings.gpkg") vl2 = QgsVectorLayer(source + "|layername=circular_strings") self.assertTrue(vl2.isValid()) p.addMapLayer(vl2) input_details = alg.getOgrCompatibleSource( "INPUT", {"INPUT": vl2.id()}, context, feedback, True ) self.assertEqual(input_details.connection_string, source) self.assertEqual(input_details.layer_name, "circular_strings") vl3 = QgsVectorLayer(source + "|layername=circular_strings_with_line") self.assertTrue(vl3.isValid()) p.addMapLayer(vl3) input_details = alg.getOgrCompatibleSource( "INPUT", {"INPUT": vl3.id()}, context, feedback, True ) self.assertEqual(input_details.connection_string, source) self.assertEqual(input_details.layer_name, "circular_strings_with_line") def testGetOgrCompatibleSourceFromFeatureSource(self): # create a memory layer and add to project and context layer = QgsVectorLayer( "Point?field=fldtxt:string&field=fldint:integer", "testmem", "memory" ) self.assertTrue(layer.isValid()) pr = layer.dataProvider() f = QgsFeature() f.setAttributes(["test", 123]) f.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(100, 200))) f2 = QgsFeature() f2.setAttributes(["test2", 457]) f2.setGeometry(QgsGeometry.fromPointXY(QgsPointXY(100, 200))) self.assertTrue(pr.addFeatures([f, f2])) self.assertEqual(layer.featureCount(), 2) # select first feature layer.selectByIds([next(layer.getFeatures()).id()]) self.assertEqual(len(layer.selectedFeatureIds()), 1) QgsProject.instance().addMapLayer(layer) context = QgsProcessingContext() context.setProject(QgsProject.instance()) alg = QgsApplication.processingRegistry().createAlgorithmById( "gdal:buffervectors" ) self.assertIsNotNone(alg) parameters = {"INPUT": QgsProcessingFeatureSourceDefinition("testmem", True)} feedback = QgsProcessingFeedback() # check that memory layer is automatically saved out to geopackage when required by GDAL algorithms input_details = alg.getOgrCompatibleSource( "INPUT", parameters, context, feedback, executing=True ) self.assertTrue(input_details.connection_string) self.assertTrue(input_details.connection_string.endswith(".gpkg")) self.assertTrue(os.path.exists(input_details.connection_string)) self.assertTrue(input_details.layer_name) # make sure that layer has only selected feature res = QgsVectorLayer(input_details.connection_string, "res") self.assertTrue(res.isValid()) self.assertEqual(res.featureCount(), 1) QgsProject.instance().removeMapLayer(layer) def testOgrLayerNameExtraction(self): with tempfile.TemporaryDirectory() as outdir: def _copyFile(dst): shutil.copyfile( os.path.join(testDataPath, "custom", "weighted.csv"), dst ) # OGR provider - single layer _copyFile(os.path.join(outdir, "a.csv")) name = GdalUtils.ogrLayerName(outdir) self.assertEqual(name, "a") # OGR provider - multiple layers _copyFile(os.path.join(outdir, "b.csv")) name1 = GdalUtils.ogrLayerName(outdir + "|layerid=0") name2 = GdalUtils.ogrLayerName(outdir + "|layerid=1") self.assertEqual(sorted([name1, name2]), ["a", "b"]) name = GdalUtils.ogrLayerName(outdir + "|layerid=2") self.assertIsNone(name) # OGR provider - layername takes precedence name = GdalUtils.ogrLayerName(outdir + "|layername=f") self.assertEqual(name, "f") name = GdalUtils.ogrLayerName(outdir + "|layerid=0|layername=f") self.assertEqual(name, "f") name = GdalUtils.ogrLayerName(outdir + "|layername=f|layerid=0") self.assertEqual(name, "f") # SQLite provider name = GdalUtils.ogrLayerName( "dbname='/tmp/x.sqlite' table=\"t\" (geometry) sql=" ) self.assertEqual(name, "t") # PostgreSQL provider name = GdalUtils.ogrLayerName( 'port=5493 sslmode=disable key=\'edge_id\' srid=0 type=LineString table="city_data"."edge" (geom) sql=' ) self.assertEqual(name, "city_data.edge") def test_gdal_connection_details_from_uri(self): context = QgsProcessingContext() output_details = GdalUtils.gdal_connection_details_from_uri( "d:/test/test.shp", context ) self.assertEqual(output_details.connection_string, "d:/test/test.shp") self.assertEqual(output_details.format, '"ESRI Shapefile"') output_details = GdalUtils.gdal_connection_details_from_uri( "d:/test/test.mif", context ) self.assertEqual(output_details.connection_string, "d:/test/test.mif") self.assertEqual(output_details.format, '"MapInfo File"') def testConnectionString(self): alg = OgrToPostGis() alg.initAlgorithm() parameters = {} feedback = QgsProcessingFeedback() context = QgsProcessingContext() # NOTE: defaults are debatable, see # https://github.com/qgis/QGIS/pull/3607#issuecomment-253971020 self.assertEqual( alg.getConnectionString(parameters, context), "host=localhost port=5432 active_schema=public", ) parameters["HOST"] = "remote" self.assertEqual( alg.getConnectionString(parameters, context), "host=remote port=5432 active_schema=public", ) parameters["HOST"] = "" self.assertEqual( alg.getConnectionString(parameters, context), "port=5432 active_schema=public", ) parameters["PORT"] = "5555" self.assertEqual( alg.getConnectionString(parameters, context), "port=5555 active_schema=public", ) parameters["PORT"] = "" self.assertEqual( alg.getConnectionString(parameters, context), "active_schema=public" ) parameters["USER"] = "usr" self.assertEqual( alg.getConnectionString(parameters, context), "active_schema=public user=usr", ) parameters["PASSWORD"] = "pwd" self.assertEqual( alg.getConnectionString(parameters, context), "password=pwd active_schema=public user=usr", ) def testCrsConversion(self): self.assertFalse(GdalUtils.gdal_crs_string(QgsCoordinateReferenceSystem())) self.assertEqual( GdalUtils.gdal_crs_string(QgsCoordinateReferenceSystem("EPSG:3111")), "EPSG:3111", ) self.assertEqual( GdalUtils.gdal_crs_string(QgsCoordinateReferenceSystem("POSTGIS:3111")), "EPSG:3111", ) self.assertEqual( GdalUtils.gdal_crs_string( QgsCoordinateReferenceSystem( "proj4: +proj=utm +zone=36 +south +a=6378249.145 +b=6356514.966398753 +towgs84=-143,-90,-294,0,0,0,0 +units=m +no_defs" ) ), "EPSG:20936", ) crs = QgsCoordinateReferenceSystem() crs.createFromProj( "+proj=utm +zone=36 +south +a=600000 +b=70000 +towgs84=-143,-90,-294,0,0,0,0 +units=m +no_defs" ) self.assertTrue(crs.isValid()) # proj 6, WKT should be used self.assertEqual( GdalUtils.gdal_crs_string(crs)[:40], 'BOUNDCRS[SOURCECRS[PROJCRS["unknown",BAS', ) self.assertEqual( GdalUtils.gdal_crs_string(QgsCoordinateReferenceSystem("ESRI:102003")), "ESRI:102003", ) def testEscapeAndJoin(self): self.assertEqual( GdalUtils.escapeAndJoin([1, "a", "a b", "a&b", "a(b)", ";"]), '1 a "a b" "a&b" "a(b)" ";"', ) self.assertEqual( GdalUtils.escapeAndJoin([1, "-srcnodata", "--srcnodata", "-9999 9999"]), '1 -srcnodata --srcnodata "-9999 9999"', ) if __name__ == "__main__": nose2.main()