diff --git a/modelbaker/iliwrapper/iliexecutable.py b/modelbaker/iliwrapper/iliexecutable.py index 34f61804..0839dc55 100644 --- a/modelbaker/iliwrapper/iliexecutable.py +++ b/modelbaker/iliwrapper/iliexecutable.py @@ -104,7 +104,6 @@ def command(self, hide_password): get_java_path(self.configuration.base_configuration) ) command_args = ili2db_jar_arg + args - valid_args = [] for command_arg in command_args: valid_args.append(self._escaped_arg(command_arg)) diff --git a/modelbaker/processing/__init__.py b/modelbaker/processing/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/modelbaker/processing/ili2db_algorithm.py b/modelbaker/processing/ili2db_algorithm.py new file mode 100644 index 00000000..3e1dd07a --- /dev/null +++ b/modelbaker/processing/ili2db_algorithm.py @@ -0,0 +1,324 @@ +import os +from abc import abstractmethod + +from qgis.core import ( + QgsProcessingAlgorithm, + QgsProcessingOutputBoolean, + QgsProcessingOutputFile, + QgsProcessingOutputNumber, + QgsProcessingOutputString, + QgsProcessingParameterAuthConfig, + QgsProcessingParameterBoolean, + QgsProcessingParameterEnum, + QgsProcessingParameterFile, + QgsProcessingParameterNumber, + QgsProcessingParameterString, +) +from qgis.PyQt.QtCore import QCoreApplication, QSettings +from qgis.PyQt.QtGui import QIcon + +from ..iliwrapper.ili2dbconfig import BaseConfiguration +from ..utils.db_utils import get_authconfig_map, get_service_config + + +class Ili2dbAlgorithm(QgsProcessingAlgorithm): + def __init__(self): + super().__init__() + + def group(self): + return self.tr("ili2db") + + def groupId(self): + return "ili2db" + + def icon(self): + return QIcon(os.path.join(os.path.dirname(__file__), "images/interlis.png")) + + def tr(self, string): + return QCoreApplication.translate("Processing", string) + + def createInstance(self): + return self.__class__() + + @abstractmethod + def connection_input_params(self): + return [] + + @abstractmethod + def connection_output_params(self): + return [] + + @abstractmethod + def get_db_configuration_from_input(self, parameters, context, configuration): + return False + + @abstractmethod + def get_output_from_db_configuration(self, configuration): + return {} + + @abstractmethod + def get_settings_configuration_from_input(self, parameters, context, configuration): + return + + def current_baseconfig(self): + baseconfig = BaseConfiguration() + settings = QSettings() + settings.beginGroup("QgisModelBaker/ili2db") + baseconfig.restore(settings) + return baseconfig + + +class Ili2pgAlgorithm(Ili2dbAlgorithm): + SERVICE = "SERVICE" + HOST = "HOST" + DBNAME = "DBNAME" + PORT = "PORT" + USER = "USER" + PASSWORD = "PASSWORD" + SCHEMA = "SCHEMA" + SSLMODE = "SSLMODE" + USESUPERUSER = "USESUPERUSER" + AUTHCFG = "AUTHCFG" + + def __init__(self): + super().__init__() + + def connection_input_params(self): + params = [] + + service_param = QgsProcessingParameterString( + self.SERVICE, + self.tr("Service"), + None, + optional=True, + ) + service_param.setHelp(self.tr("The PostgreSQL service config file.")) + params.append(service_param) + + host_param = QgsProcessingParameterString( + self.HOST, + self.tr("Host"), + defaultValue="localhost", + optional=True, + ) + host_param.setHelp( + self.tr("The host of the database server. By default is localhost.") + ) + params.append(host_param) + + port_param = QgsProcessingParameterNumber( + self.PORT, + self.tr("Port"), + type=QgsProcessingParameterNumber.Type.Integer, + defaultValue=5432, + optional=True, + ) + port_param.setHelp( + self.tr("The port of the database server. By default is 5432.") + ) + params.append(port_param) + + dbname_param = QgsProcessingParameterString( + self.DBNAME, + self.tr("Database"), + defaultValue=None, + optional=True, + ) + dbname_param.setHelp(self.tr("The database name. The database should exist.")) + params.append(dbname_param) + + schema_param = QgsProcessingParameterString( + self.SCHEMA, + self.tr("Schema"), + defaultValue=None, + optional=True, + ) + schema_param.setHelp(self.tr("The database schema.")) + params.append(schema_param) + + sslmode_param = QgsProcessingParameterEnum( + self.SSLMODE, + self.tr("SSL Mode"), + ["disable", "allow", "prefer", "require", "verify-ca", "verify-full"], + defaultValue=None, + optional=True, + usesStaticStrings=True, + ) + sslmode_param.setHelp(self.tr("The SSL mode if needed.")) + params.append(sslmode_param) + user_param = QgsProcessingParameterString( + self.USER, + self.tr("User"), + defaultValue=None, + optional=True, + ) + user_param.setHelp(self.tr("The user to access the database.")) + params.append(user_param) + + password_param = QgsProcessingParameterString( + self.PASSWORD, + self.tr("Password"), + defaultValue=None, + optional=True, + ) + password_param.setHelp(self.tr("The password of the user.")) + params.append(password_param) + + usesuperuser_param = QgsProcessingParameterBoolean( + self.USESUPERUSER, + self.tr("Use superuser login from the settings"), + defaultValue=False, + ) + usesuperuser_param.setHelp( + self.tr("Excecute the task with the super user login from the settings") + ) + params.append(usesuperuser_param) + + authcfg_param = QgsProcessingParameterAuthConfig( + self.AUTHCFG, + self.tr("Authentication"), + defaultValue=None, + optional=True, + ) + authcfg_param.setHelp( + self.tr( + "When choosing a QGIS Authentication you don't need user and password." + ) + ) + params.append(authcfg_param) + + return params + + def connection_output_params(self): + params = [] + + # outputs for pass through + params.append(QgsProcessingOutputString(self.SERVICE, self.tr("Service"))) + params.append(QgsProcessingOutputString(self.HOST, self.tr("Host"))) + params.append(QgsProcessingOutputString(self.DBNAME, self.tr("Database"))) + params.append(QgsProcessingOutputNumber(self.PORT, self.tr("Port"))) + params.append(QgsProcessingOutputString(self.USER, self.tr("User"))) + params.append(QgsProcessingOutputString(self.PASSWORD, self.tr("Password"))) + params.append(QgsProcessingOutputString(self.SCHEMA, self.tr("Schema"))) + params.append(QgsProcessingOutputString(self.SSLMODE, self.tr("SSL Mode"))) + params.append( + QgsProcessingOutputBoolean(self.USESUPERUSER, self.tr("Use Superuser")) + ) + params.append( + QgsProcessingOutputString(self.AUTHCFG, self.tr("Authentication")) + ) + return params + + def get_db_configuration_from_input(self, parameters, context, configuration): + """ + Returns true if mandatory parameters are given + """ + + configuration.dbservice = self.parameterAsString( + parameters, self.SERVICE, context + ) + service_map, _ = get_service_config(configuration.dbservice) + + if self.parameterAsString(parameters, self.AUTHCFG, context): + configuration.dbauthid = self.parameterAsString( + parameters, self.AUTHCFG, context + ) # needed for passthroug + authconfig_map = get_authconfig_map(configuration.dbauthid) + configuration.dbusr = authconfig_map.get("username") + configuration.dbpwd = authconfig_map.get("password") + else: + configuration.dbusr = self.parameterAsString( + parameters, self.USER, context + ) or service_map.get("user") + configuration.dbpwd = self.parameterAsString( + parameters, self.PASSWORD, context + ) or service_map.get("password") + configuration.dbhost = self.parameterAsString( + parameters, self.HOST, context + ) or service_map.get("host") + configuration.dbport = str( + self.parameterAsInt(parameters, self.PORT, context) + ) or service_map.get("port") + configuration.database = self.parameterAsString( + parameters, self.DBNAME, context + ) or service_map.get("dbname") + configuration.dbschema = self.parameterAsString( + parameters, self.SCHEMA, context + ) + configuration.sslmode = self.parameterAsEnum(parameters, self.SSLMODE, context) + configuration.db_use_super_login = self.parameterAsBoolean( + parameters, self.USESUPERUSER, context + ) + valid = bool( + configuration.dbhost and configuration.database and configuration.dbschema + ) + return valid + + def get_output_from_db_configuration(self, configuration): + """ + Returns an output map + """ + + output_map = { + self.SERVICE: configuration.dbservice, + self.HOST: configuration.dbhost, + self.DBNAME: configuration.database, + self.PORT: configuration.dbport, + self.USER: configuration.dbusr, + self.PASSWORD: configuration.dbpwd, + self.SCHEMA: configuration.dbschema, + self.SSLMODE: configuration.sslmode, + self.USESUPERUSER: configuration.db_use_super_login, + self.AUTHCFG: configuration.dbauthid, + } + return output_map + + +class Ili2gpkgAlgorithm(Ili2dbAlgorithm): + + DBPATH = "DBPATH" + + def __init__(self): + super().__init__() + + def connection_input_params(self): + params = [] + + dbpath_param = QgsProcessingParameterFile( + self.DBPATH, + self.tr("Database File Path"), + defaultValue=None, + optional=True, + ) + dbpath_param.setHelp(self.tr("The database file path (*.gpkg).")) + params.append(dbpath_param) + + return params + + def connection_output_params(self): + params = [] + + params.append( + QgsProcessingOutputFile(self.DBPATH, self.tr("Database File Path")) + ) + + return params + + def get_db_configuration_from_input(self, parameters, context, configuration): + """ + Returns true if mandatory parameters are given + """ + valid = False + dbpath = self.parameterAsFile(parameters, self.DBPATH, context) + if dbpath and dbpath.endswith(".gpkg"): + configuration.dbfile = dbpath + valid = True + return valid + + def get_output_from_db_configuration(self, configuration): + """ + Returns an output map + """ + + output_map = {self.DBPATH: configuration.dbfile} + return output_map diff --git a/modelbaker/processing/ili2db_exporting.py b/modelbaker/processing/ili2db_exporting.py new file mode 100644 index 00000000..5ef17d33 --- /dev/null +++ b/modelbaker/processing/ili2db_exporting.py @@ -0,0 +1,410 @@ +""" +*************************************************************************** +* * +* 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. * +* * +*************************************************************************** +""" + +from typing import Any, Optional + +from qgis.core import ( + QgsProcessingContext, + QgsProcessingException, + QgsProcessingFeedback, + QgsProcessingOutputBoolean, + QgsProcessingOutputFile, + QgsProcessingParameterBoolean, + QgsProcessingParameterEnum, + QgsProcessingParameterFileDestination, + QgsProcessingParameterString, +) + +from ..iliwrapper import iliexporter +from ..iliwrapper.globals import DbIliMode +from ..iliwrapper.ili2dbconfig import ExportConfiguration +from ..iliwrapper.ili2dbutils import JavaNotFoundError +from .ili2db_algorithm import Ili2gpkgAlgorithm, Ili2pgAlgorithm +from .ili2db_operating import ProcessOperatorBase + + +class ProcessExporter(ProcessOperatorBase): + + # Filters + FILTERMODE = "FILTERMODE" # none, models, baskets or datasets + FILTER = "FILTER" # model, basket or dataset names + + # Settings + EXPORTMODEL = "EXPORTMODEL" + DISABLEVALIDATION = "DISABLEVALIDATION" + + XTFFILEPATH = "XTFFILEPATH" + + # Result + ISVALID = "ISVALID" + + def __init__(self, parent): + super().__init__(parent) + self.parent = parent + + def export_input_params(self): + params = [] + + xtffile_param = QgsProcessingParameterFileDestination( + self.XTFFILEPATH, + self.tr("Target Transfer File (XTF)"), + self.tr("INTERLIS Transfer File (*.xtf *.xml)"), + defaultValue=None, + optional=False, + ) + xtffile_param.setHelp( + self.tr("The XTF File where the data should be exported to.") + ) + params.append(xtffile_param) + + filtermode_param = QgsProcessingParameterEnum( + self.FILTERMODE, + self.tr("Filter Mode"), + ["Models", "Baskets", "Datasets"], + optional=True, + usesStaticStrings=True, + ) + filtermode_param.setHelp( + self.tr( + "Whether data should be filtered according to the models, baskets or datasets in which they are stored." + ) + ) + params.append(filtermode_param) + + filter_param = QgsProcessingParameterString( + self.FILTER, self.tr("Filter (semicolon-separated)"), optional=True + ) + filter_param.setHelp( + self.tr( + "A semicolon-separated list of the relevant models, baskets or datasets" + ) + ) + params.append(filter_param) + + exportmodel_param = QgsProcessingParameterString( + self.EXPORTMODEL, + self.tr("Export Models"), + optional=True, + ) + exportmodel_param.setHelp( + self.tr( + """ +

If your data is in the format of the cantonal model, but you want to export it in the format of the national model you need to define this here.

+

Usually, this is one single model. However, it is also possible to pass multiple models, which makes sense if there are multiple base models in the schema you want to export.

+ """ + ) + ) + params.append(exportmodel_param) + + disablevalidation_param = QgsProcessingParameterBoolean( + self.DISABLEVALIDATION, + self.tr("Disable validaton of the data"), + defaultValue=False, + ) + disablevalidation_param.setHelp( + self.tr( + "Ignores geometry errors (--skipGeometryErrors) and constraint validation (--disableValidation)" + ) + ) + params.append(disablevalidation_param) + + return params + + def export_output_params(self): + params = [] + + params.append( + QgsProcessingOutputBoolean(self.ISVALID, self.tr("Export Result")) + ) + params.append( + QgsProcessingOutputFile(self.XTFFILEPATH, self.tr("Transfer File Path")) + ) + + return params + + def initParameters(self): + for connection_input_param in self.parent.connection_input_params(): + self.parent.addParameter(connection_input_param) + for connection_output_param in self.parent.connection_output_params(): + self.parent.addOutput(connection_output_param) + + for export_input_param in self.export_input_params(): + self.parent.addParameter(export_input_param) + for export_output_param in self.export_output_params(): + self.parent.addOutput(export_output_param) + + def run(self, configuration, feedback): + # run + exporter = iliexporter.Exporter(self) + exporter.tool = configuration.tool + + # to do superuser finden? und auch dpparams? + exporter.configuration = configuration + exporter.stdout.connect(feedback.pushInfo) + exporter.stderr.connect(feedback.pushInfo) + + if feedback.isCanceled(): + return {} + + isvalid = False + try: + feedback.pushInfo(f"Run: {exporter.command(True)}") + result = exporter.run(None) + if result == iliexporter.Exporter.SUCCESS: + feedback.pushInfo(self.tr("... export succeeded")) + else: + feedback.pushWarning(self.tr("... export failed")) + isvalid = bool(result == iliexporter.Exporter.SUCCESS) + except JavaNotFoundError as e: + raise QgsProcessingException( + self.tr("Java not found error:").format(e.error_string) + ) + + return {self.ISVALID: isvalid, self.XTFFILEPATH: configuration.xtffile} + + def get_configuration_from_input(self, parameters, context, tool): + + configuration = ExportConfiguration() + configuration.base_configuration = self.parent.current_baseconfig() + configuration.tool = tool + + # get database settings form the parent + if not self.parent.get_db_configuration_from_input( + parameters, context, configuration + ): + return None + + # get settings according to the db + configuration.with_exporttid = self._get_tid_handling(configuration) + + # get settings from the input + filtermode = self.parent.parameterAsString(parameters, self.FILTERMODE, context) + filters = self.parent.parameterAsString(parameters, self.FILTER, context) + if filtermode == "Models" and filters: + configuration.ilimodels = filters + elif filtermode == "Datasets" and filters: + configuration.dataset = filters + elif filtermode == "Baskets" and filters: + configuration.baskets = filters + else: + configuration.ilimodels = ";".join(self._get_model_names(configuration)) + + exportmodels = self.parent.parameterAsString( + parameters, self.EXPORTMODEL, context + ) + if exportmodels: + configuration.iliexportmodels = exportmodels + + configuration.disable_validation = self.parent.parameterAsBool( + parameters, self.DISABLEVALIDATION, context + ) + + configuration.xtffile = self.parent.parameterAsFile( + parameters, self.XTFFILEPATH, context + ) + + return configuration + + +class ExportingPGAlgorithm(Ili2pgAlgorithm): + """ + This is an algorithm from Model Baker. + It is meant for the data export from a PostgreSQL database. + """ + + def __init__(self): + super().__init__() + + # initialize the exporter with self as parent + self.exporter = ProcessExporter(self) + + def name(self) -> str: + """ + Returns the algorithm name, used for identifying the algorithm. + """ + return "modelbaker_ili2pg_exporter" + + def displayName(self) -> str: + """ + Returns the translated algorithm name, which should be used for any + user-visible display of the algorithm name. + """ + return self.tr("Export with ili2pg (PostGIS)") + + def tags(self) -> list[str]: + + return [ + "modelbaker", + "interlis", + "model", + "baker", + "export", + "transferfile", + "xtf", + "ili2db", + "ili2pg", + "Postgres", + "PostGIS", + ] + + def shortDescription(self) -> str: + """ + Returns the tooltip text when hovering the algorithm + """ + return self.tr( + """ +

Exports data from PostgreSQL schema with ili2pg.

+

The ili2pg parameters are set in the same way as in the Model Baker Plugin.

+

General Model Baker settings like custom model directories or db parameters are concerned.

+

The parameter passed to ili2db by default is --exportTid.

+

In a database where you have created basket columns, you will have the possibility to filter by baskets or datasets.

+

Additionally, you can define an export model to specify the format in which you want to export the data (e.g. the base model of your extended model).

+ + """ + ) + + def shortHelpString(self) -> str: + """ + Returns the help text on the right. + """ + return self.tr( + """ +

Exports data from PostgreSQL schema with ili2pg.

+

The ili2pg parameters are set in the same way as in the Model Baker Plugin.

+

General Model Baker settings like custom model directories or db parameters are concerned.

+

The parameter passed to ili2db by default is --exportTid.

+

In a database where you have created basket columns, you will have the possibility to filter by baskets or datasets.

+

Additionally, you can define an export model to specify the format in which you want to export the data (e.g. the base model of your extended model).

+ + """ + ) + + def initAlgorithm(self, config: Optional[dict[str, Any]] = None): + self.exporter.initParameters() + + def processAlgorithm( + self, + parameters: dict[str, Any], + context: QgsProcessingContext, + feedback: QgsProcessingFeedback, + ) -> dict[str, Any]: + """ + Here is where the processing itself takes place. + """ + output_map = {} + configuration = self.exporter.get_configuration_from_input( + parameters, context, DbIliMode.pg + ) + if not configuration: + raise QgsProcessingException( + self.tr("Invalid input parameters. Cannot start export") + ) + else: + output_map.update(self.exporter.run(configuration, feedback)) + output_map.update(self.get_output_from_db_configuration(configuration)) + return output_map + + +class ExportingGPKGAlgorithm(Ili2gpkgAlgorithm): + """ + This is an algorithm from Model Baker. + It is meant for the data export from a GeoPackage file. + """ + + def __init__(self): + super().__init__() + + # initialize the exporter with self as parent + self.exporter = ProcessExporter(self) + + def name(self) -> str: + """ + Returns the algorithm name, used for identifying the algorithm. + """ + return "modelbaker_ili2gpkg_exporter" + + def displayName(self) -> str: + """ + Returns the translated algorithm name, which should be used for any + user-visible display of the algorithm name. + """ + return self.tr("Export with ili2gpkg (GeoPackage)") + + def tags(self) -> list[str]: + + return [ + "modelbaker", + "interlis", + "model", + "baker", + "export", + "transferfile", + "xtf", + "ili2db", + "ili2gpkg", + "GeoPackage", + "GPKG", + ] + + def shortDescription(self) -> str: + """ + Returns the tooltip text when hovering the algorithm + """ + return self.tr( + """ +

Exports data from GeoPackage file with ili2gpkg.

+

The ili2gpkg parameters are set in the same way as in the Model Baker Plugin.

+

The parameter passed to ili2db by default is --exportTid.

+

In a database where you have created basket columns, you will have the possibility to filter by baskets or datasets.

+

Additionally, you can define an export model to specify the format in which you want to export the data (e.g. the base model of your extended model).

+ + """ + ) + + def shortHelpString(self) -> str: + """ + Returns the help text on the right. + """ + return self.tr( + """ +

Exports data from GeoPackage file with ili2gpkg.

+

The ili2gpkg parameters are set in the same way as in the Model Baker Plugin.

+

The parameter passed to ili2db by default is --exportTid.

+

In a database where you have created basket columns, you will have the possibility to filter by baskets or datasets.

+

Additionally, you can define an export model to specify the format in which you want to export the data (e.g. the base model of your extended model).

+ + """ + ) + + def initAlgorithm(self, config: Optional[dict[str, Any]] = None): + self.exporter.initParameters() + + def processAlgorithm( + self, + parameters: dict[str, Any], + context: QgsProcessingContext, + feedback: QgsProcessingFeedback, + ) -> dict[str, Any]: + """ + Here is where the processing itself takes place. + """ + output_map = {} + configuration = self.exporter.get_configuration_from_input( + parameters, context, DbIliMode.gpkg + ) + if not configuration: + raise QgsProcessingException( + self.tr("Invalid input parameters. Cannot start export") + ) + else: + output_map.update(self.exporter.run(configuration, feedback)) + output_map.update(self.get_output_from_db_configuration(configuration)) + return output_map diff --git a/modelbaker/processing/ili2db_importing.py b/modelbaker/processing/ili2db_importing.py new file mode 100644 index 00000000..4af48826 --- /dev/null +++ b/modelbaker/processing/ili2db_importing.py @@ -0,0 +1,389 @@ +""" +*************************************************************************** +* * +* 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. * +* * +*************************************************************************** +""" + +from typing import Any, Optional + +from qgis.core import ( + QgsProcessingContext, + QgsProcessingException, + QgsProcessingFeedback, + QgsProcessingOutputBoolean, + QgsProcessingParameterBoolean, + QgsProcessingParameterFile, + QgsProcessingParameterString, +) + +from ..iliwrapper import iliimporter +from ..iliwrapper.globals import DbIliMode +from ..iliwrapper.ili2dbconfig import ( + Ili2DbCommandConfiguration, + ImportDataConfiguration, + UpdateDataConfiguration, +) +from ..iliwrapper.ili2dbutils import JavaNotFoundError +from .ili2db_algorithm import Ili2gpkgAlgorithm, Ili2pgAlgorithm +from .ili2db_operating import ProcessOperatorBase + + +class ProcessImporter(ProcessOperatorBase): + + # Settings + DISABLEVALIDATION = "DISABLEVALIDATION" + DATASET = "DATASET" + DELETEDATA = "DELETEDATA" + + XTFFILEPATH = "XTFFILEPATH" + + # Result + ISVALID = "ISVALID" + + def __init__(self, parent): + super().__init__(parent) + self.parent = parent + + def import_input_params(self): + params = [] + + xtffile_param = QgsProcessingParameterFile( + self.XTFFILEPATH, + self.tr("Source Transfer File (XTF)"), + defaultValue=None, + optional=False, + ) + xtffile_param.setHelp( + self.tr("The XTF File where the data should be imported from.") + ) + params.append(xtffile_param) + + dataset_param = QgsProcessingParameterString( + self.DATASET, self.tr("Dataset name"), optional=True + ) + dataset_param.setHelp( + self.tr( + "The dataset the data should be connected to (works only when basket handling is active in the existing database)." + ) + ) + params.append(dataset_param) + + deletedata_param = QgsProcessingParameterBoolean( + self.DELETEDATA, + self.tr("Delete data first"), + defaultValue=False, + ) + deletedata_param.setHelp( + self.tr( + "Deletes previously existing data from the target database (when basket handling active it performs a --replace instead of an --update)" + ) + ) + params.append(deletedata_param) + + disablevalidation_param = QgsProcessingParameterBoolean( + self.DISABLEVALIDATION, + self.tr("Disable validaton of the data"), + defaultValue=False, + ) + disablevalidation_param.setHelp( + self.tr( + "Ignores geometry errors (--skipGeometryErrors) and constraint validation (--disableValidation)" + ) + ) + params.append(disablevalidation_param) + + return params + + def import_output_params(self): + params = [] + + params.append( + QgsProcessingOutputBoolean(self.ISVALID, self.tr("Import Result")) + ) + + return params + + def initParameters(self): + for connection_input_param in self.parent.connection_input_params(): + self.parent.addParameter(connection_input_param) + for connection_output_param in self.parent.connection_output_params(): + self.parent.addOutput(connection_output_param) + + for import_input_param in self.import_input_params(): + self.parent.addParameter(import_input_param) + for import_output_param in self.import_output_params(): + self.parent.addOutput(import_output_param) + + def run(self, configuration, feedback): + + # run + importer = iliimporter.Importer(self) + importer.tool = configuration.tool + + # to do superuser finden? und auch dpparams? + importer.configuration = configuration + importer.stdout.connect(feedback.pushInfo) + importer.stderr.connect(feedback.pushInfo) + + if feedback.isCanceled(): + return {} + + isvalid = False + try: + feedback.pushInfo(f"Run: {importer.command(True)}") + result = importer.run(None) + if result == iliimporter.Importer.SUCCESS: + feedback.pushInfo(self.tr("... import succeeded")) + else: + feedback.pushWarning(self.tr("... import failed")) + isvalid = bool(result == iliimporter.Importer.SUCCESS) + except JavaNotFoundError as e: + raise QgsProcessingException( + self.tr("Java not found error:").format(e.error_string) + ) + + return {self.ISVALID: isvalid, self.XTFFILEPATH: configuration.xtffile} + + def get_configuration_from_input(self, parameters, context, tool): + + configuration = Ili2DbCommandConfiguration() + configuration.base_configuration = self.parent.current_baseconfig() + configuration.tool = tool + + # get database settings form the parent + if not self.parent.get_db_configuration_from_input( + parameters, context, configuration + ): + return None + + # get settings according to the db + if not self._basket_handling(configuration): + configuration = ImportDataConfiguration(configuration) + else: + configuration = UpdateDataConfiguration(configuration) + configuration.with_importbid = True + configuration.with_importtid = self._get_tid_handling(configuration) + + # get settings from the input + configuration.dataset = self.parent.parameterAsString( + parameters, self.DATASET, context + ) + configuration.disable_validation = self.parent.parameterAsBool( + parameters, self.DISABLEVALIDATION, context + ) + configuration.delete_data = self.parent.parameterAsBool( + parameters, self.DELETEDATA, context + ) + + configuration.xtffile = self.parent.parameterAsFile( + parameters, self.XTFFILEPATH, context + ) + + return configuration + + +class ImportingPGAlgorithm(Ili2pgAlgorithm): + """ + This is an algorithm from Model Baker. + It is meant for the data import to a PostgreSQL database. + """ + + def __init__(self): + super().__init__() + + # initialize the importer with self as parent + self.importer = ProcessImporter(self) + + def name(self) -> str: + """ + Returns the algorithm name, used for identifying the algorithm. + """ + return "modelbaker_ili2pg_importer" + + def displayName(self) -> str: + """ + Returns the translated algorithm name, which should be used for any + user-visible display of the algorithm name. + """ + return self.tr("Import with ili2pg (PostGIS)") + + def tags(self) -> list[str]: + + return [ + "modelbaker", + "interlis", + "model", + "baker", + "import", + "transferfile", + "xtf", + "ili2db", + "ili2pg", + "Postgres", + "PostGIS", + ] + + def shortDescription(self) -> str: + """ + Returns the tooltip text when hovering the algorithm + """ + return self.tr( + """ +

Imports data to PostgreSQL schema with ili2pg.

+

The ili2pg parameters are set in the same way as in the Model Baker Plugin.

+

General Model Baker settings like custom model directories or db parameters are concerned.

+

The parameters passed to ili2db by default are --importTid and, on databases where you have created basket columns, --importBid as well.

+

On a database where you have created basket columns, the command is --update (or --replace when you choose to delete data previously).

+

Additionally, you need to define a dataset name for the import on such databases.

+

On a database where no basket column has been created, the command will be --import (and --deleteData when you choose to delete data previously).

+ + """ + ) + + def shortHelpString(self) -> str: + """ + Returns the help text on the right. + """ + return self.tr( + """ +

Imports data to PostgreSQL schema with ili2pg.

+

The ili2pg parameters are set in the same way as in the Model Baker Plugin.

+

General Model Baker settings like custom model directories or db parameters are concerned.

+

The parameters passed to ili2db by default are --importTid and, on databases where you have created basket columns, --importBid as well.

+

On a database where you have created basket columns, the command is --update (or --replace when you choose to delete data previously).

+

Additionally, you need to define a dataset name for the import on such databases.

+

On a database where no basket column has been created, the command will be --import (and --deleteData when you choose to delete data previously).

+ + """ + ) + + def initAlgorithm(self, config: Optional[dict[str, Any]] = None): + self.importer.initParameters() + + def processAlgorithm( + self, + parameters: dict[str, Any], + context: QgsProcessingContext, + feedback: QgsProcessingFeedback, + ) -> dict[str, Any]: + """ + Here is where the processing itself takes place. + """ + output_map = {} + configuration = self.importer.get_configuration_from_input( + parameters, context, DbIliMode.pg + ) + if not configuration: + raise QgsProcessingException( + self.tr("Invalid input parameters. Cannot start import") + ) + else: + output_map.update(self.importer.run(configuration, feedback)) + output_map.update(self.get_output_from_db_configuration(configuration)) + return output_map + + +class ImportingGPKGAlgorithm(Ili2gpkgAlgorithm): + """ + This is an algorithm from Model Baker. + It is meant for the data import to a GeoPackage file. + """ + + def __init__(self): + super().__init__() + + # initialize the importer with self as parent + self.importer = ProcessImporter(self) + + def name(self) -> str: + """ + Returns the algorithm name, used for identifying the algorithm. + """ + return "modelbaker_ili2gpkg_importer" + + def displayName(self) -> str: + """ + Returns the translated algorithm name, which should be used for any + user-visible display of the algorithm name. + """ + return self.tr("Import with ili2gpkg (GeoPackage)") + + def tags(self) -> list[str]: + + return [ + "modelbaker", + "interlis", + "model", + "baker", + "import", + "transferfile", + "xtf", + "ili2db", + "ili2gpkg", + "GeoPackage", + "GPKG", + ] + + def shortDescription(self) -> str: + """ + Returns the tooltip text when hovering the algorithm + """ + return self.tr( + """ +

Imports data to GeoPackage file with ili2gpkg.

+

The ili2gpkg parameters are set in the same way as in the Model Baker Plugin.

+

General Model Baker settings like custom model directories concerned.

+

The parameters passed to ili2db by default are --importTid and, on databases where you have created basket columns, --importBid as well.

+

On a database where you have created basket columns, the command is --update (or --replace when you choose to delete data previously).

+

Additionally, you need to define a dataset name for the import on such databases.

+

On a database where no basket column has been created, the command will be --import (and --deleteData when you choose to delete data previously).

+ + """ + ) + + def shortHelpString(self) -> str: + """ + Returns the help text on the right. + """ + return self.tr( + """ +

Imports data to GeoPackage file with ili2gpkg.

+

The ili2gpkg parameters are set in the same way as in the Model Baker Plugin.

+

General Model Baker settings like custom model directories concerned.

+

The parameters passed to ili2db by default are --importTid and, on databases where you have created basket columns, --importBid as well.

+

On a database where you have created basket columns, the command is --update (or --replace when you choose to delete data previously).

+

Additionally, you need to define a dataset name for the import on such databases.

+

On a database where no basket column has been created, the command will be --import (and --deleteData when you choose to delete data previously).

+ + """ + ) + + def initAlgorithm(self, config: Optional[dict[str, Any]] = None): + self.importer.initParameters() + + def processAlgorithm( + self, + parameters: dict[str, Any], + context: QgsProcessingContext, + feedback: QgsProcessingFeedback, + ) -> dict[str, Any]: + """ + Here is where the processing itself takes place. + """ + output_map = {} + configuration = self.importer.get_configuration_from_input( + parameters, context, DbIliMode.gpkg + ) + if not configuration: + raise QgsProcessingException( + self.tr("Invalid input parameters. Cannot start import") + ) + else: + output_map.update(self.importer.run(configuration, feedback)) + output_map.update(self.get_output_from_db_configuration(configuration)) + return output_map diff --git a/modelbaker/processing/ili2db_operating.py b/modelbaker/processing/ili2db_operating.py new file mode 100644 index 00000000..bb1ec0a4 --- /dev/null +++ b/modelbaker/processing/ili2db_operating.py @@ -0,0 +1,53 @@ +""" +*************************************************************************** +* * +* 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. * +* * +*************************************************************************** +""" + + +from qgis.PyQt.QtCore import QCoreApplication, QObject + +from ..utils.db_utils import get_db_connector +from ..utils.globals import MODELS_BLACKLIST + + +class ProcessOperatorBase(QObject): + def __init__(self, parent): + super().__init__() + self.parent = parent + + def _get_tid_handling(self, configuration): + db_connector = get_db_connector(configuration) + if db_connector: + return db_connector.get_tid_handling() + return False + + def _basket_handling(self, configuration): + db_connector = get_db_connector(configuration) + if db_connector: + return db_connector.get_basket_handling() + return False + + def _get_model_names(self, configuration): + modelnames = [] + + db_connector = get_db_connector(configuration) + if ( + db_connector + and db_connector.db_or_schema_exists() + and db_connector.metadata_exists() + ): + db_models = db_connector.get_models() + for db_model in db_models: + name = db_model["modelname"] + if name and name not in modelnames and name not in MODELS_BLACKLIST: + modelnames.append(name) + return modelnames + + def tr(self, string): + return QCoreApplication.translate("Processing", string) diff --git a/modelbaker/processing/ili2db_validating.py b/modelbaker/processing/ili2db_validating.py new file mode 100644 index 00000000..2977547e --- /dev/null +++ b/modelbaker/processing/ili2db_validating.py @@ -0,0 +1,445 @@ +""" +*************************************************************************** +* * +* 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. * +* * +*************************************************************************** +""" + +import os +from datetime import datetime +from typing import Any, Optional + +from qgis.core import ( + QgsProcessingContext, + QgsProcessingException, + QgsProcessingFeedback, + QgsProcessingOutputBoolean, + QgsProcessingOutputString, + QgsProcessingParameterBoolean, + QgsProcessingParameterEnum, + QgsProcessingParameterFile, + QgsProcessingParameterString, +) +from qgis.PyQt.QtCore import QStandardPaths + +from ..iliwrapper import ilivalidator +from ..iliwrapper.globals import DbIliMode +from ..iliwrapper.ili2dbconfig import ValidateConfiguration +from ..iliwrapper.ili2dbutils import JavaNotFoundError +from .ili2db_algorithm import Ili2gpkgAlgorithm, Ili2pgAlgorithm +from .ili2db_operating import ProcessOperatorBase + + +class ProcessValidator(ProcessOperatorBase): + + # Filters + FILTERMODE = "FILTERMODE" # none, models, baskets or datasets + FILTER = "FILTER" # model, basket or dataset names + + # Settings + EXPORTMODEL = "EXPORTMODEL" + SKIPGEOMETRYERRORS = "SKIPGEOMETRYERRORS" + VERBOSE = "VERBOSE" + VALIDATORCONFIGFILEPATH = "VALIDATORCONFIGFILEPATH" + + # Result + ISVALID = "ISVALID" + XTFLOGPATH = "XTFLOGPATH" + + def __init__(self, parent): + super().__init__(parent) + self.parent = parent + + def validation_input_params(self): + params = [] + filtermode_param = QgsProcessingParameterEnum( + self.FILTERMODE, + self.tr("Filter Mode"), + ["Models", "Baskets", "Datasets"], + optional=True, + usesStaticStrings=True, + ) + filtermode_param.setHelp( + self.tr( + "Whether data should be filtered according to the models, baskets or datasets in which they are stored." + ) + ) + params.append(filtermode_param) + + filter_param = QgsProcessingParameterString( + self.FILTER, self.tr("Filter (semicolon-separated)"), optional=True + ) + filter_param.setHelp( + self.tr( + "A semicolon-separated list of the relevant models, baskets or datasets" + ) + ) + params.append(filter_param) + + exportmodel_param = QgsProcessingParameterString( + self.EXPORTMODEL, + self.tr("Export Models"), + optional=True, + ) + exportmodel_param.setHelp( + self.tr( + """ +

If your data is in the format of the cantonal model, but you want to validate it in the format of the national model you need to define this here.

+

Usually, this is one single model. However, it is also possible to pass multiple models, which makes sense if there are multiple base models in the schema you want to validate.

+ """ + ) + ) + params.append(exportmodel_param) + + skipgeom_param = QgsProcessingParameterBoolean( + self.SKIPGEOMETRYERRORS, + self.tr("Skip Geometry Errors"), + defaultValue=False, + ) + skipgeom_param.setHelp( + self.tr( + "Ignores geometry errors (--skipGeometryErrors) and AREA topology validation (--disableAreaValidation)" + ) + ) + params.append(skipgeom_param) + + verbose_param = QgsProcessingParameterBoolean( + self.VERBOSE, + self.tr("Activate Verbose Mode"), + defaultValue=False, + ) + verbose_param.setHelp( + self.tr("Verbose Mode provides you more information in the log output.") + ) + params.append(verbose_param) + + validatorconfig_param = QgsProcessingParameterFile( + self.VALIDATORCONFIGFILEPATH, + self.tr("Validator config file"), + optional=True, + ) + validatorconfig_param.setHelp( + self.tr("You can add a validator config file to control the validation.") + ) + params.append(validatorconfig_param) + + return params + + def validation_output_params(self): + params = [] + + params.append( + QgsProcessingOutputBoolean(self.ISVALID, self.tr("Validation Result")) + ) + params.append( + QgsProcessingOutputString(self.XTFLOGPATH, self.tr("XTF Log File")) + ) + + return params + + def initParameters(self): + for connection_input_param in self.parent.connection_input_params(): + self.parent.addParameter(connection_input_param) + for connection_output_param in self.parent.connection_output_params(): + self.parent.addOutput(connection_output_param) + + for validation_input_param in self.validation_input_params(): + self.parent.addParameter(validation_input_param) + for validation_output_param in self.validation_output_params(): + self.parent.addOutput(validation_output_param) + + def run(self, configuration, feedback): + # run + validator = ilivalidator.Validator(self) + validator.tool = configuration.tool + + # to do superuser finden? und auch dpparams? + validator.configuration = configuration + validator.stdout.connect(feedback.pushInfo) + validator.stderr.connect(feedback.pushInfo) + + if feedback.isCanceled(): + return {} + + isvalid = False + try: + feedback.pushInfo(f"Run: {validator.command(True)}") + result = validator.run(None) + if result == ilivalidator.Validator.SUCCESS: + feedback.pushInfo(self.tr("... validation succeeded")) + else: + feedback.pushWarning(self.tr("... validation failed")) + isvalid = bool(result == ilivalidator.Validator.SUCCESS) + except JavaNotFoundError as e: + raise QgsProcessingException( + self.tr("Java not found error:").format(e.error_string) + ) + + return {self.ISVALID: isvalid, self.XTFLOGPATH: configuration.xtflog} + + def get_configuration_from_input(self, parameters, context, tool): + + configuration = ValidateConfiguration() + configuration.base_configuration = self.parent.current_baseconfig() + configuration.tool = tool + + # get database settings form the parent + if not self.parent.get_db_configuration_from_input( + parameters, context, configuration + ): + return None + + # get static + output_file_name = ( + "modelbakerili2dbvalidatingalgorithm_xtflog_{:%Y%m%d%H%M%S%f}.xtf".format( + datetime.now() + ) + ) + configuration.xtflog = os.path.join( + QStandardPaths.writableLocation( + QStandardPaths.StandardLocation.TempLocation + ), + output_file_name, + ) + + # get settings according to the db + configuration.with_exporttid = self._get_tid_handling(configuration) + + # get settings from the input + filtermode = self.parent.parameterAsString(parameters, self.FILTERMODE, context) + filters = self.parent.parameterAsString(parameters, self.FILTER, context) + if filtermode == "Models" and filters: + configuration.ilimodels = filters + elif filtermode == "Datasets" and filters: + configuration.dataset = filters + elif filtermode == "Baskets" and filters: + configuration.baskets = filters + else: + configuration.ilimodels = ";".join(self._get_model_names(configuration)) + + exportmodels = self.parent.parameterAsString( + parameters, self.EXPORTMODEL, context + ) + if exportmodels: + configuration.iliexportmodels = exportmodels + + configuration.skip_geometry_errors = self.parent.parameterAsBool( + parameters, self.SKIPGEOMETRYERRORS, context + ) + + configuration.verbose = self.parent.parameterAsBool( + parameters, self.VERBOSE, context + ) + + validatorconfigfile = self.parent.parameterAsFile( + parameters, self.VALIDATORCONFIGFILEPATH, context + ) + + if validatorconfigfile: + configuration.valid_config = validatorconfigfile + + return configuration + + +class ValidatingPGAlgorithm(Ili2pgAlgorithm): + """ + This is an algorithm from Model Baker. + It is meant for the data validation stored in a PostgreSQL database. + """ + + def __init__(self): + super().__init__() + + # initialize the validator with self as parent + self.validator = ProcessValidator(self) + + def name(self) -> str: + """ + Returns the algorithm name, used for identifying the algorithm. + """ + return "modelbaker_ili2pg_validator" + + def displayName(self) -> str: + """ + Returns the translated algorithm name, which should be used for any + user-visible display of the algorithm name. + """ + return self.tr("Validate with ili2pg (PostGIS)") + + def tags(self) -> list[str]: + + return [ + "modelbaker", + "interlis", + "model", + "baker", + "validate", + "validation", + "ili2db", + "ili2pg", + "Postgres", + "PostGIS", + ] + + def shortDescription(self) -> str: + """ + Returns the tooltip text when hovering the algorithm + """ + return self.tr( + """ +

Validates data in PostgreSQL schema with ili2pg.

+

The ili2pg parameters are set in the same way as in the Model Baker Plugin Validator.

+

General Model Baker settings like custom model directories and db parameters are concerned.

+

The parameter passed to ili2db by default is --exportTid.

+

In a database where you have created basket columns, you will have the possibility to filter by baskets or datasets.

+

Skip Geometry Errors ignores geometry errors (--skipGeometryErrors) and AREA topology validation (--disableAreaValidation) and the verbose mode provides you more information in the log output.

+

Additionally, you can define an export model to specify the format in which you want to validate the data (e.g. the base model of your extended model). Also You can add a validator config file to control the validation.

+ + """ + ) + + def shortHelpString(self) -> str: + """ + Returns the help text on the right. + """ + return self.tr( + """ +

Validates data in PostgreSQL schema with ili2pg.

+

The ili2pg parameters are set in the same way as in the Model Baker Plugin Validator.

+

General Model Baker settings like custom model directories and db parameters are concerned.

+

The parameter passed to ili2db by default is --exportTid.

+

In a database where you have created basket columns, you will have the possibility to filter by baskets or datasets.

+

Skip Geometry Errors ignores geometry errors (--skipGeometryErrors) and AREA topology validation (--disableAreaValidation) and the verbose mode provides you more information in the log output.

+

Additionally, you can define an export model to specify the format in which you want to validate the data (e.g. the base model of your extended model). Also You can add a validator config file to control the validation.

+ + """ + ) + + def initAlgorithm(self, config: Optional[dict[str, Any]] = None): + self.validator.initParameters() + + def processAlgorithm( + self, + parameters: dict[str, Any], + context: QgsProcessingContext, + feedback: QgsProcessingFeedback, + ) -> dict[str, Any]: + """ + Here is where the processing itself takes place. + """ + output_map = {} + configuration = self.validator.get_configuration_from_input( + parameters, context, DbIliMode.pg + ) + if not configuration: + raise QgsProcessingException( + self.tr("Invalid input parameters. Cannot start validation") + ) + else: + output_map.update(self.validator.run(configuration, feedback)) + output_map.update(self.get_output_from_db_configuration(configuration)) + return output_map + + +class ValidatingGPKGAlgorithm(Ili2gpkgAlgorithm): + """ + This is an algorithm from Model Baker. + It is meant for the data validation stored in a GeoPackage file. + """ + + def __init__(self): + super().__init__() + + # initialize the validator with self as parent + self.validator = ProcessValidator(self) + + def name(self) -> str: + """ + Returns the algorithm name, used for identifying the algorithm. + """ + return "modelbaker_ili2gpkg_validator" + + def displayName(self) -> str: + """ + Returns the translated algorithm name, which should be used for any + user-visible display of the algorithm name. + """ + return self.tr("Validate with ili2gpkg (GeoPackage)") + + def tags(self) -> list[str]: + + return [ + "modelbaker", + "interlis", + "model", + "baker", + "validate", + "validation", + "ili2db", + "ili2gpkg", + "GeoPackage", + "GPKG", + ] + + def shortDescription(self) -> str: + """ + Returns the tooltip text when hovering the algorithm + """ + return self.tr( + """ +

Validates data in GeoPackage file with ili2gpkg.

+

The ili2gpkg parameters are set in the same way as in the Model Baker Plugin Validator.

+

General Model Baker settings like custom model directories and db parameters are concerned.

+

The parameter passed to ili2db by default is --exportTid.

+

In a database where you have created basket columns, you will have the possibility to filter by baskets or datasets.

+

Skip Geometry Errors ignores geometry errors (--skipGeometryErrors) and AREA topology validation (--disableAreaValidation) and the verbose mode provides you more information in the log output.

+

Additionally, you can define an export model to specify the format in which you want to validate the data (e.g. the base model of your extended model). Also You can add a validator config file to control the validation.

+ + """ + ) + + def shortHelpString(self) -> str: + """ + Returns the help text on the right. + """ + return self.tr( + """ +

Validates data in GeoPackage file with ili2gpkg.

+

The ili2gpkg parameters are set in the same way as in the Model Baker Plugin Validator.

+

General Model Baker settings like custom model directories and db parameters are concerned.

+

The parameter passed to ili2db by default is --exportTid.

+

In a database where you have created basket columns, you will have the possibility to filter by baskets or datasets.

+

Skip Geometry Errors ignores geometry errors (--skipGeometryErrors) and AREA topology validation (--disableAreaValidation) and the verbose mode provides you more information in the log output.

+

Additionally, you can define an export model to specify the format in which you want to validate the data (e.g. the base model of your extended model). Also You can add a validator config file to control the validation.

+ + """ + ) + + def initAlgorithm(self, config: Optional[dict[str, Any]] = None): + self.validator.initParameters() + + def processAlgorithm( + self, + parameters: dict[str, Any], + context: QgsProcessingContext, + feedback: QgsProcessingFeedback, + ) -> dict[str, Any]: + """ + Here is where the processing itself takes place. + """ + output_map = {} + configuration = self.validator.get_configuration_from_input( + parameters, context, DbIliMode.gpkg + ) + if not configuration: + raise QgsProcessingException( + self.tr("Invalid input parameters. Cannot start validation") + ) + else: + output_map.update(self.validator.run(configuration, feedback)) + output_map.update(self.get_output_from_db_configuration(configuration)) + return output_map diff --git a/modelbaker/processing/images/interlis.png b/modelbaker/processing/images/interlis.png new file mode 100644 index 00000000..0dfa4a59 Binary files /dev/null and b/modelbaker/processing/images/interlis.png differ diff --git a/modelbaker/processing/util_algorithm.py b/modelbaker/processing/util_algorithm.py new file mode 100644 index 00000000..4b23727d --- /dev/null +++ b/modelbaker/processing/util_algorithm.py @@ -0,0 +1,12 @@ +from qgis.core import QgsProcessingAlgorithm + + +class UtilAlgorithm(QgsProcessingAlgorithm): + def __init__(self): + super().__init__() + + def group(self): + return self.tr("Utils") + + def groupId(self): + return "utils" diff --git a/modelbaker/processing/util_layersourceparsing.py b/modelbaker/processing/util_layersourceparsing.py new file mode 100644 index 00000000..25790843 --- /dev/null +++ b/modelbaker/processing/util_layersourceparsing.py @@ -0,0 +1,180 @@ +""" +*************************************************************************** +* * +* 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. * +* * +*************************************************************************** +""" + +from typing import Any, Optional + +from qgis.core import ( + QgsProcessing, + QgsProcessingContext, + QgsProcessingException, + QgsProcessingFeedback, + QgsProcessingOutputNumber, + QgsProcessingOutputString, + QgsProcessingParameterVectorLayer, +) +from qgis.PyQt.QtCore import QCoreApplication + +from ..iliwrapper.ili2dbconfig import Ili2DbCommandConfiguration +from ..utils.db_utils import get_configuration_from_sourceprovider +from .util_algorithm import UtilAlgorithm + + +class LayerSourceParsingAlgorithm(UtilAlgorithm): + """ + This is an algorithm from Model Baker. + + It is meant for reading the data source parameters from a layer + """ + + # Connection + LAYER = "LAYER" + + # Output + PROVIDER = "PROVIDER" + ## PG + SERVICE = "SERVICE" + HOST = "HOST" + DBNAME = "DBNAME" + PORT = "PORT" + USER = "USER" + PASSWORD = "PASSWORD" + SCHEMA = "SCHEMA" + SSLMODE = "SSLMODE" + AUTHCFG = "AUTHCFG" + ## GPKG + DBPATH = "DBPATH" + + def __init__(self): + super().__init__() + + def name(self) -> str: + """ + Returns the algorithm name, used for identifying the algorithm. + """ + return "modelbaker_util_layerconnection" + + def displayName(self) -> str: + """ + Returns the translated algorithm name, which should be used for any + user-visible display of the algorithm name. + """ + return self.tr("Get connection from layersource") + + def tags(self) -> list[str]: + + return ["layer", "source", "database", "modelbaker", "ili2db", "interlis"] + + def shortDescription(self) -> str: + """ + Returns a short description string for the algorithm. + """ + return self.tr( + """ +

Receives connection parameters from the layer source.

+

Parses the GeoPackage or PostgreSQL source of a layer.

+

Provides the parameters to be used in the Model Baker ili2db algorithms.

+ + """ + ) + + def shortHelpString(self) -> str: + """ + Returns a short helper string for the algorithm. + """ + return self.tr( + """ +

Receives connection parameters from the layer source.

+

Parses the GeoPackage or PostgreSQL source of a layer.

+

Provides the parameters to be used in the Model Baker ili2db algorithms.

+ + """ + ) + + def initAlgorithm(self, config: Optional[dict[str, Any]] = None): + + sourcelayer_param = QgsProcessingParameterVectorLayer( + self.LAYER, + self.tr("Layer"), + [QgsProcessing.SourceType.TypeVector], + self.tr("No source layer selected"), + ) + sourcelayer_param.setHelp( + self.tr( + "Source layer to get database connection from. If set, it will be prefered over the other connection settings." + ) + ) + self.addParameter(sourcelayer_param) + + self.addOutput(QgsProcessingOutputString(self.PROVIDER, self.tr("Provider"))) + self.addOutput(QgsProcessingOutputString(self.SERVICE, self.tr("Service"))) + self.addOutput(QgsProcessingOutputString(self.HOST, self.tr("Host"))) + self.addOutput(QgsProcessingOutputString(self.DBNAME, self.tr("Database"))) + self.addOutput(QgsProcessingOutputNumber(self.PORT, self.tr("Port"))) + self.addOutput(QgsProcessingOutputString(self.USER, self.tr("User"))) + self.addOutput(QgsProcessingOutputString(self.PASSWORD, self.tr("Password"))) + self.addOutput(QgsProcessingOutputString(self.SCHEMA, self.tr("Schema"))) + self.addOutput(QgsProcessingOutputString(self.SSLMODE, self.tr("SSL Mode"))) + self.addOutput( + QgsProcessingOutputString(self.AUTHCFG, self.tr("Authentication")) + ) + + self.addOutput( + QgsProcessingOutputString(self.DBPATH, self.tr("Database File Path")) + ) + + def processAlgorithm( + self, + parameters: dict[str, Any], + context: QgsProcessingContext, + feedback: QgsProcessingFeedback, + ) -> dict[str, Any]: + """ + Here is where the processing itself takes place. + """ + configuration = Ili2DbCommandConfiguration() + + sourcelayer = self.parameterAsVectorLayer(parameters, self.LAYER, context) + if not sourcelayer: + raise QgsProcessingException( + self.invalidSourceError(parameters, self.LAYER) + ) + return {} + + source_provider = sourcelayer.dataProvider() + valid, mode = get_configuration_from_sourceprovider( + source_provider, configuration + ) + if not (valid and mode): + # error + return {} + + if feedback.isCanceled(): + return {} + + return { + self.PROVIDER: source_provider.name(), + self.SERVICE: configuration.dbservice, + self.HOST: configuration.dbhost, + self.DBNAME: configuration.database, + self.PORT: configuration.dbport, + self.USER: configuration.dbusr, + self.PASSWORD: configuration.dbpwd, + self.SCHEMA: configuration.dbschema, + self.SSLMODE: configuration.sslmode, + self.AUTHCFG: configuration.dbauthid, + self.DBPATH: configuration.dbfile, + } + + def tr(self, string): + return QCoreApplication.translate("Processing", string) + + def createInstance(self): + return self.__class__() diff --git a/modelbaker/utils/db_utils.py b/modelbaker/utils/db_utils.py index be1158e9..871b92ac 100644 --- a/modelbaker/utils/db_utils.py +++ b/modelbaker/utils/db_utils.py @@ -86,6 +86,7 @@ def get_configuration_from_sourceprovider(provider, configuration): configuration.dbhost = layer_source.host() or service_map.get("host") configuration.dbport = layer_source.port() or service_map.get("port") configuration.database = layer_source.database() or service_map.get("dbname") + configuration.sslmode = QgsDataSourceUri.encodeSslMode(layer_source.sslMode()) configuration.dbschema = layer_source.schema() valid = bool( configuration.dbhost and configuration.database and configuration.dbschema diff --git a/modelbaker/utils/globals.py b/modelbaker/utils/globals.py index 64967cde..ffcf8d33 100644 --- a/modelbaker/utils/globals.py +++ b/modelbaker/utils/globals.py @@ -18,6 +18,75 @@ """ from enum import Enum, IntEnum +MODELS_BLACKLIST = [ + "CHBaseEx_MapCatalogue_V1", + "CHBaseEx_WaterNet_V1", + "CHBaseEx_Sewage_V1", + "CHAdminCodes_V1", + "AdministrativeUnits_V1", + "AdministrativeUnitsCH_V1", + "WithOneState_V1", + "WithLatestModification_V1", + "WithModificationObjects_V1", + "GraphicCHLV03_V1", + "GraphicCHLV95_V1", + "NonVector_Base_V2", + "NonVector_Base_V3", + "NonVector_Base_LV03_V3_1", + "NonVector_Base_LV95_V3_1", + "GeometryCHLV03_V1", + "GeometryCHLV95_V1", + "Geometry_V1", + "InternationalCodes_V1", + "Localisation_V1", + "LocalisationCH_V1", + "Dictionaries_V1", + "DictionariesCH_V1", + "CatalogueObjects_V1", + "CatalogueObjectTrees_V1", + "AbstractSymbology", + "CodeISO", + "CoordSys", + "GM03_2_1Comprehensive", + "GM03_2_1Core", + "GM03_2Comprehensive", + "GM03_2Core", + "GM03Comprehensive", + "GM03Core", + "IliRepository09", + "IliSite09", + "IlisMeta07", + "IliVErrors", + "INTERLIS_ext", + "RoadsExdm2ben", + "RoadsExdm2ben_10", + "RoadsExgm2ien", + "RoadsExgm2ien_10", + "StandardSymbology", + "StandardSymbology", + "Time", + "Units", + "", + "CHAdminCodes_V2", + "AdministrativeUnits_V2", + "AdministrativeUnitsCH_V2", + "WithOneState_V2", + "WithLatestModification_V2", + "WithModificationObjects_V2", + "GraphicCHLV03_V2", + "GraphicCHLV95_V2", + "GeometryCHLV03_V2", + "GeometryCHLV95_V2", + "Geometry_V2", + "InternationalCodes_V2", + "Localisation_V2", + "LocalisationCH_V2", + "Dictionaries_V2", + "DictionariesCH_V2", + "CatalogueObjects_V2", + "CatalogueObjectTrees_V2", +] + class DbActionType(Enum): """Defines constants for generate, schema_import, import data, or export actions of modelbaker.""" diff --git a/tests/test_processing.py b/tests/test_processing.py new file mode 100644 index 00000000..3461f0c3 --- /dev/null +++ b/tests/test_processing.py @@ -0,0 +1,363 @@ +""" +/*************************************************************************** + begin : 27/09/2025 + git sha : :%H$ + copyright : (C) 2025 by Dave Signer + email : david@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. * + * * + ***************************************************************************/ +""" + +import datetime +import logging +import os +import tempfile + +from qgis.core import QgsProcessingContext, QgsProcessingFeedback +from qgis.testing import start_app, unittest + +from modelbaker.iliwrapper import iliimporter +from modelbaker.iliwrapper.globals import DbIliMode +from modelbaker.processing.ili2db_exporting import ( + ExportingGPKGAlgorithm, + ExportingPGAlgorithm, +) +from modelbaker.processing.ili2db_importing import ( + ImportingGPKGAlgorithm, + ImportingPGAlgorithm, +) +from modelbaker.processing.ili2db_validating import ( + ValidatingGPKGAlgorithm, + ValidatingPGAlgorithm, +) +from tests.utils import iliimporter_config, testdata_path + +start_app() + + +class TestProcessingAlgorithms(unittest.TestCase): + @classmethod + def setUpClass(cls): + """Run before all tests.""" + cls.basetestpath = tempfile.mkdtemp() + + def iliimporter_pg_config_params(self): + configuration = iliimporter_config(DbIliMode.ili2pg) + params = { + "HOST": configuration.dbhost, + "DBNAME": configuration.database, + "USER": configuration.dbusr, + "PASSWORD": configuration.dbpwd, + } + return params + + def gpkg_file(self, basket_col): + importer = iliimporter.Importer() + importer.tool = DbIliMode.ili2gpkg + importer.configuration.ilifile = testdata_path("ilimodels/RoadsSimple.ili") + importer.configuration.ilimodels = "RoadsSimple" + importer.configuration.dbfile = os.path.join( + self.basetestpath, + "tmp_roads_simple_{:%Y%m%d%H%M%S%f}.gpkg".format(datetime.datetime.now()), + ) + importer.configuration.create_basket_col = basket_col + importer.configuration.inheritance = "smart2" + importer.stdout.connect(self.print_info) + importer.stderr.connect(self.print_error) + if importer.run() == iliimporter.Importer.SUCCESS: + return importer.configuration.dbfile + else: + return None + + def pg_schema(self, basket_col): + importer = iliimporter.Importer() + importer.tool = DbIliMode.ili2pg + importer.configuration = iliimporter_config(importer.tool) + importer.configuration.ilifile = testdata_path("ilimodels/RoadsSimple.ili") + importer.configuration.ilimodels = "RoadsSimple" + importer.configuration.dbschema = "roads_simple_{:%Y%m%d%H%M%S%f}".format( + datetime.datetime.now() + ) + importer.configuration.srs_code = 2056 + importer.configuration.inheritance = "smart2" + importer.configuration.create_basket_col = basket_col + importer.stdout.connect(self.print_info) + importer.stderr.connect(self.print_error) + if importer.run() == iliimporter.Importer.SUCCESS: + return importer.configuration.dbschema + else: + return None + + def test_algs_gpkg(self): + conn_parameters_baskets = {} + conn_parameters_baskets["DBPATH"] = self.gpkg_file(True) + self._algs_with_baskets( + conn_parameters_baskets, + ImportingGPKGAlgorithm, + ValidatingGPKGAlgorithm, + ExportingGPKGAlgorithm, + ) + conn_parameters = {} + conn_parameters["DBPATH"] = self.gpkg_file(False) + self._algs_without_baskets( + conn_parameters, + ImportingGPKGAlgorithm, + ValidatingGPKGAlgorithm, + ExportingGPKGAlgorithm, + ) + + def test_algs_pg(self): + conn_parameters_baskets = self.iliimporter_pg_config_params() + conn_parameters_baskets["SCHEMA"] = self.pg_schema(True) + self._algs_with_baskets( + conn_parameters_baskets, + ImportingPGAlgorithm, + ValidatingPGAlgorithm, + ExportingPGAlgorithm, + ) + conn_parameters = self.iliimporter_pg_config_params() + conn_parameters["SCHEMA"] = self.pg_schema(False) + self._algs_without_baskets( + conn_parameters, + ImportingPGAlgorithm, + ValidatingPGAlgorithm, + ExportingPGAlgorithm, + ) + + def _algs_with_baskets( + self, + conn_parameters, + importing_algorithm, + validating_algorithm, + exporting_algorithm, + ): + # import valid data now to a dataset called 'validdata' + import_parameters = { + "XTFFILEPATH": testdata_path("xtf/test_roads_simple.xtf"), + "DATASET": "validdata", + } + import_parameters.update(conn_parameters) + alg = importing_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(import_parameters, context, feedback) + assert output["ISVALID"] + + # import invalid data now to a dataset called 'invaliddata' + import_parameters = { + "XTFFILEPATH": testdata_path("xtf/test_roads_simple_invalid.xtf"), + "DATASET": "invaliddata", + } + import_parameters.update(conn_parameters) + alg = importing_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(import_parameters, context, feedback) + # fails + assert not output["ISVALID"] + + # import invalid data now to a dataset called 'invaliddata' + # this time we disable the validation + import_parameters = { + "XTFFILEPATH": testdata_path("xtf/test_roads_simple_invalid.xtf"), + "DATASET": "invaliddata", + "DISABLEVALIDATION": True, + } + import_parameters.update(conn_parameters) + alg = importing_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(import_parameters, context, feedback) + assert output["ISVALID"] + + # validate without specific parameters + validation_parameters = {} + validation_parameters.update(conn_parameters) + alg = validating_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(validation_parameters, context, feedback) + # fails + assert not output["ISVALID"] + + # validate again only the dataset 'validdata' + validation_parameters = {"FILTERMODE": "Datasets", "FILTER": "validdata"} + validation_parameters.update(conn_parameters) + alg = validating_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(validation_parameters, context, feedback) + assert output["ISVALID"] + + valid_targetfile = os.path.join(self.basetestpath, "valid_export.xtf") + invalid_targetfile = os.path.join(self.basetestpath, "invalid_export.xtf") + + # let's export without specific parameters + export_parameters = {"XTFFILEPATH": valid_targetfile} + export_parameters.update(conn_parameters) + alg = exporting_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(export_parameters, context, feedback) + # fails + assert not output["ISVALID"] + + # let's export again only the dataset 'validdata' + export_parameters = { + "XTFFILEPATH": valid_targetfile, + "FILTERMODE": "Datasets", + "FILTER": "validdata", + } + export_parameters.update(conn_parameters) + alg = exporting_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(export_parameters, context, feedback) + assert output["ISVALID"] + + # let's export the invalid dataset 'invaliddata' + export_parameters = { + "XTFFILEPATH": invalid_targetfile, + "FILTERMODE": "Datasets", + "FILTER": "invaliddata", + } + export_parameters.update(conn_parameters) + alg = exporting_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(export_parameters, context, feedback) + # fails + assert not output["ISVALID"] + + # let's export the invalid dataset 'invaliddata' and disable validation + export_parameters = { + "XTFFILEPATH": invalid_targetfile, + "FILTERMODE": "Datasets", + "FILTER": "invaliddata", + "DISABLEVALIDATION": True, + } + export_parameters.update(conn_parameters) + alg = exporting_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(export_parameters, context, feedback) + assert output["ISVALID"] + + assert os.path.isfile(valid_targetfile) + assert os.path.isfile(invalid_targetfile) + + def _algs_without_baskets( + self, + conn_parameters, + importing_algorithm, + validating_algorithm, + exporting_algorithm, + ): + # import valid data now + import_parameters = {"XTFFILEPATH": testdata_path("xtf/test_roads_simple.xtf")} + import_parameters.update(conn_parameters) + alg = importing_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(import_parameters, context, feedback) + assert output["ISVALID"] + + # validate without specific parameters + validation_parameters = {} + validation_parameters.update(conn_parameters) + alg = validating_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(validation_parameters, context, feedback) + assert output["ISVALID"] + + # import invalid data now + import_parameters = { + "XTFFILEPATH": testdata_path("xtf/test_roads_simple_invalid.xtf") + } + import_parameters.update(conn_parameters) + alg = importing_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(import_parameters, context, feedback) + # fails + assert not output["ISVALID"] + + # import invalid data again + # this time we disable the validation + import_parameters = { + "XTFFILEPATH": testdata_path("xtf/test_roads_simple_invalid.xtf"), + "DISABLEVALIDATION": True, + } + import_parameters.update(conn_parameters) + alg = importing_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(import_parameters, context, feedback) + assert output["ISVALID"] + + # validate without specific parameters + validation_parameters = {} + validation_parameters.update(conn_parameters) + alg = validating_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(validation_parameters, context, feedback) + # fails + assert not output["ISVALID"] + + invalid_targetfile = os.path.join(self.basetestpath, "invalid_export.xtf") + + # let's export without specific parameters + export_parameters = {"XTFFILEPATH": invalid_targetfile} + export_parameters.update(conn_parameters) + alg = exporting_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(export_parameters, context, feedback) + # fails + assert not output["ISVALID"] + + # let's export and disable validation + export_parameters = { + "XTFFILEPATH": invalid_targetfile, + "DISABLEVALIDATION": True, + } + export_parameters.update(conn_parameters) + alg = exporting_algorithm() + alg.initAlgorithm() + context = QgsProcessingContext() + feedback = QgsProcessingFeedback() + output = alg.processAlgorithm(export_parameters, context, feedback) + assert output["ISVALID"] + + assert os.path.isfile(invalid_targetfile) + + def print_info(self, text): + logging.info(text) + + def print_error(self, text): + logging.error(text) diff --git a/tests/test_projectgen.py b/tests/test_projectgen.py index 26c362ca..0a6e02ea 100644 --- a/tests/test_projectgen.py +++ b/tests/test_projectgen.py @@ -3741,14 +3741,14 @@ def test_kbs_postgis_toppings(self): ) assert belasteter_standort_group is not None belasteter_standort_group_layer = belasteter_standort_group.findLayers() - assert [layer.name() for layer in belasteter_standort_group_layer] == [ + assert {layer.name() for layer in belasteter_standort_group_layer} == { "Belasteter_Standort (Geo_Lage_Punkt)", "Belasteter_Standort (Geo_Lage_Polygon)", - ] + } informationen_group = qgis_project.layerTreeRoot().findGroup("Informationen") assert informationen_group is not None informationen_group_layers = informationen_group.findLayers() - assert [layer.name() for layer in informationen_group_layers] == [ + assert {layer.name() for layer in informationen_group_layers} == { "EGRID_", "Deponietyp_", "ZustaendigkeitKataster", @@ -3767,27 +3767,27 @@ def test_kbs_postgis_toppings(self): "UntersMassn", "Deponietyp", "LanguageCode_ISO639_1", - ] + } text_infos_group = informationen_group.findGroup("Text Infos") assert text_infos_group is not None text_infos_group_layers = text_infos_group.findLayers() - assert [layer.name() for layer in text_infos_group_layers] == [ + assert {layer.name() for layer in text_infos_group_layers} == { "MultilingualMText", "LocalisedMText", "MultilingualText", "LocalisedText", - ] + } other_infos_group = informationen_group.findGroup("Other Infos") self.assertIsNotNone(other_infos_group) other_infos_group_layers = other_infos_group.findLayers() - assert [layer.name() for layer in other_infos_group_layers] == [ + assert {layer.name() for layer in other_infos_group_layers} == { "StatusAltlV", "Standorttyp", "UntersMassn", "Deponietyp", "LanguageCode_ISO639_1", - ] + } # check the node properties belasteter_standort_punkt_layer = None belasteter_standort_polygon_layer = None @@ -4133,16 +4133,16 @@ def test_kbs_geopackage_toppings(self): ) assert belasteter_standort_group is not None belasteter_standort_group_layer = belasteter_standort_group.findLayers() - assert [layer.name() for layer in belasteter_standort_group_layer] == [ + assert {layer.name() for layer in belasteter_standort_group_layer} == { "Belasteter_Standort (Geo_Lage_Punkt)", "Belasteter_Standort", - ] + } informationen_group = qgis_project.layerTreeRoot().findGroup("Informationen") assert informationen_group is not None informationen_group_layers = informationen_group.findLayers() - assert [layer.name() for layer in informationen_group_layers] == [ + assert {layer.name() for layer in informationen_group_layers} == { "EGRID_", "Deponietyp_", "ZustaendigkeitKataster", @@ -4161,27 +4161,27 @@ def test_kbs_geopackage_toppings(self): "UntersMassn", "Deponietyp", "LanguageCode_ISO639_1", - ] + } text_infos_group = informationen_group.findGroup("Text Infos") assert text_infos_group is not None text_infos_group_layers = text_infos_group.findLayers() - assert [layer.name() for layer in text_infos_group_layers] == [ + assert {layer.name() for layer in text_infos_group_layers} == { "MultilingualMText", "LocalisedMText", "MultilingualText", "LocalisedText", - ] + } other_infos_group = informationen_group.findGroup("Other Infos") assert other_infos_group is not None other_infos_group_layers = other_infos_group.findLayers() - assert [layer.name() for layer in other_infos_group_layers] == [ + assert {layer.name() for layer in other_infos_group_layers} == { "StatusAltlV", "Standorttyp", "UntersMassn", "Deponietyp", "LanguageCode_ISO639_1", - ] + } # check the node properties belasteter_standort_punkt_layer = None belasteter_standort_polygon_layer = None diff --git a/tests/test_projecttopping.py b/tests/test_projecttopping.py index 9860dc4f..290d0393 100644 --- a/tests/test_projecttopping.py +++ b/tests/test_projecttopping.py @@ -127,10 +127,10 @@ def test_kbs_postgis_qlr_layers(self): ) assert ili_layers_group is not None ili_layers_group_layers = ili_layers_group.findLayers() - assert [layer.name() for layer in ili_layers_group_layers] == [ + assert {layer.name() for layer in ili_layers_group_layers} == { "Belasteter_Standort (Geo_Lage_Punkt)", "Belasteter_Standort (Geo_Lage_Polygon)", - ] + } qlr_layers_group = qgis_project.layerTreeRoot().findGroup("Other Layers") assert qlr_layers_group is not None @@ -248,10 +248,10 @@ def test_kbs_postgis_source_layers(self): ) assert ili_layers_group is not None ili_layers_group_layers = ili_layers_group.findLayers() - assert [layer.name() for layer in ili_layers_group_layers] == [ + assert {layer.name() for layer in ili_layers_group_layers} == { "Belasteter_Standort (Geo_Lage_Punkt)", "Belasteter_Standort (Geo_Lage_Polygon)", - ] + } source_layers_group = qgis_project.layerTreeRoot().findGroup("Other Layers") assert source_layers_group is not None diff --git a/tests/test_validate.py b/tests/test_validate.py index 85f518af..cc703fe0 100644 --- a/tests/test_validate.py +++ b/tests/test_validate.py @@ -344,17 +344,17 @@ def test_validate_skips(self): "value 1314393.469 is out of range in attribute Geometry", "Attribute Name requires a value", "Object should associate 1 to 1 target objects (instead of 0)", - "Set Constraint Brutalism.Buildings.Resident.Constraint1 is not true." + "Set Constraint Brutalism.Buildings.Resident.Constraint1 is not true.", } - errors = set() + errors = set() for i in range(4): errors.add( result_model.index(i, 0).data( int(ilivalidator.ValidationResultModel.Roles.MESSAGE) ) ) - + assert errors == expected_errors # Skip geometry errors @@ -369,17 +369,17 @@ def test_validate_skips(self): expected_errors_no_geom = { "Attribute Name requires a value", "Object should associate 1 to 1 target objects (instead of 0)", - "Set Constraint Brutalism.Buildings.Resident.Constraint1 is not true." + "Set Constraint Brutalism.Buildings.Resident.Constraint1 is not true.", } - errors = set() + errors = set() for i in range(3): errors.add( result_model.index(i, 0).data( int(ilivalidator.ValidationResultModel.Roles.MESSAGE) ) ) - + assert errors == expected_errors_no_geom # Multiplicity off @@ -395,17 +395,17 @@ def test_validate_skips(self): assert result_model.rowCount() == 2 expected_errors_no_multi = { "value 1314393.469 is out of range in attribute Geometry", - "Set Constraint Brutalism.Buildings.Resident.Constraint1 is not true." + "Set Constraint Brutalism.Buildings.Resident.Constraint1 is not true.", } - errors = set() + errors = set() for i in range(2): errors.add( result_model.index(i, 0).data( int(ilivalidator.ValidationResultModel.Roles.MESSAGE) ) ) - + assert errors == expected_errors_no_multi # Constraint off @@ -425,14 +425,14 @@ def test_validate_skips(self): "Object should associate 1 to 1 target objects (instead of 0)", } - errors = set() + errors = set() for i in range(3): errors.add( result_model.index(i, 0).data( int(ilivalidator.ValidationResultModel.Roles.MESSAGE) ) ) - + assert errors == expected_errors_no_const def test_validate_exportmodels_gpkg(self): @@ -496,21 +496,21 @@ def test_validate_exportmodels_gpkg(self): result_model.configuration = validator.configuration result_model.reload() assert result_model.rowCount() == 3 - + expected_errors = { "Needs an ethical evaluation (EthischeBeurteilung)", "Beschreibung and/or Referenzcode must be defined.", - "Beschreibung needed when top secret." + "Beschreibung needed when top secret.", } - errors = set() + errors = set() for i in range(3): errors.add( result_model.index(i, 0).data( int(ilivalidator.ValidationResultModel.Roles.MESSAGE) ) ) - + assert errors == expected_errors # Validate at cantonal level (setting --exportModels KantonaleOrtsplanung_V1_1)