# -*- coding: utf-8 -*- """ *************************************************************************** TestTools.py --------------------- Date : February 2013 Copyright : (C) 2013 by Victor Olaya Email : volayaf at gmail dot com *************************************************************************** * * * 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__ = 'Victor Olaya' __date__ = 'February 2013' __copyright__ = '(C) 2013, Victor Olaya' import os import posixpath import re import yaml import hashlib import ast from osgeo import gdal from osgeo.gdalconst import GA_ReadOnly from numpy import nan_to_num from qgis.core import (QgsApplication, QgsProcessing, QgsProcessingParameterDefinition, QgsProcessingParameterBoolean, QgsProcessingParameterNumber, QgsProcessingParameterDistance, QgsProcessingParameterDuration, QgsProcessingParameterFile, QgsProcessingParameterBand, QgsProcessingParameterString, QgsProcessingParameterVectorLayer, QgsProcessingParameterFeatureSource, QgsProcessingParameterRasterLayer, QgsProcessingParameterMultipleLayers, QgsProcessingParameterRasterDestination, QgsProcessingParameterFeatureSink, QgsProcessingParameterVectorDestination, QgsProcessingParameterFileDestination, QgsProcessingParameterEnum) from qgis.PyQt.QtCore import QCoreApplication, QMetaObject from qgis.PyQt.QtWidgets import QDialog, QVBoxLayout, QTextEdit, QMessageBox def extractSchemaPath(filepath): """ Tries to find where the file is relative to the QGIS source code directory. If it is already placed in the processing or QGIS testdata directory it will return an appropriate schema and relative filepath Args: filepath: The path of the file to examine Returns: A tuple (schema, relative_file_path) where the schema is 'qgs' or 'proc' if we can assume that the file is in this testdata directory. """ parts = [] schema = None localpath = '' path = filepath part = True while part and filepath: (path, part) = os.path.split(path) if part == 'testdata' and not localpath: localparts = parts localparts.reverse() # we always want posix style paths here localpath = posixpath.join(*localparts) parts.append(part) parts.reverse() try: testsindex = parts.index('tests') except ValueError: return '', filepath if parts[testsindex - 1] == 'processing': schema = 'proc' return schema, localpath def parseParameters(command): """ Parse alg string to grab parameters value. Can handle quotes and comma. """ pos = 0 exp = re.compile(r"""(['"]?)(.*?)\1(,|$)""") while True: m = exp.search(command, pos) result = m.group(2) separator = m.group(3) # Handle special values: if result == 'None': result = None elif result.lower() == str(True).lower(): result = True elif result.lower() == str(False).lower(): result = False yield result if not separator: break pos = m.end(0) def splitAlgIdAndParameters(command): """ Extracts the algorithm ID and input parameter list from a processing runalg command """ exp = re.compile(r"""['"](.*?)['"]\s*,\s*(.*)""") m = exp.search(command[len('processing.run('):-1]) alg_id = m.group(1) params = m.group(2) # replace QgsCoordinateReferenceSystem('EPSG:4325') with just string value exp = re.compile(r"""QgsCoordinateReferenceSystem\((['"].*?['"])\)""") params = exp.sub('\\1', params) return alg_id, ast.literal_eval(params) def createTest(text): definition = {} alg_id, parameters = splitAlgIdAndParameters(text) alg = QgsApplication.processingRegistry().createAlgorithmById(alg_id) definition['name'] = 'Test ({})'.format(alg_id) definition['algorithm'] = alg_id params = {} results = {} i = 0 for param in alg.parameterDefinitions(): if param.flags() & QgsProcessingParameterDefinition.FlagHidden or param.isDestination(): continue if not param.name() in parameters: continue i += 1 token = parameters[param.name()] # Handle empty parameters that are optionals if param.flags() & QgsProcessingParameterDefinition.FlagOptional and token is None: continue if isinstance(param, (QgsProcessingParameterVectorLayer, QgsProcessingParameterFeatureSource)): schema, filepath = extractSchemaPath(token) p = { 'type': 'vector', 'name': filepath } if not schema: p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]' params[param.name()] = p elif isinstance(param, QgsProcessingParameterRasterLayer): schema, filepath = extractSchemaPath(token) p = { 'type': 'raster', 'name': filepath } if not schema: p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]' params[param.name()] = p elif isinstance(param, QgsProcessingParameterMultipleLayers): multiparams = token newparam = [] # Handle datatype detection dataType = param.layerType() if dataType in [QgsProcessing.TypeVectorAnyGeometry, QgsProcessing.TypeVectorPoint, QgsProcessing.TypeVectorLine, QgsProcessing.TypeVectorPolygon, QgsProcessing.TypeVector]: dataType = 'vector' else: dataType = 'raster' schema = None for mp in multiparams: schema, filepath = extractSchemaPath(mp) newparam.append({ 'type': dataType, 'name': filepath }) p = { 'type': 'multi', 'params': newparam } if not schema: p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]' params[param.name()] = p elif isinstance(param, QgsProcessingParameterFile): schema, filepath = extractSchemaPath(token) p = { 'type': 'file', 'name': filepath } if not schema: p['location'] = '[The source data is not in the testdata directory. Please use data in the processing/tests/testdata folder.]' params[param.name()] = p elif isinstance(param, QgsProcessingParameterString): params[param.name()] = token elif isinstance(param, QgsProcessingParameterBoolean): params[param.name()] = token elif isinstance(param, (QgsProcessingParameterNumber, QgsProcessingParameterDistance)): if param.dataType() == QgsProcessingParameterNumber.Integer: params[param.name()] = int(token) else: params[param.name()] = float(token) elif isinstance(param, QgsProcessingParameterEnum): if isinstance(token, list): params[param.name()] = [int(t) for t in token] else: params[param.name()] = int(token) elif isinstance(param, QgsProcessingParameterBand): params[param.name()] = int(token) elif token: if token[0] == '"': token = token[1:] if token[-1] == '"': token = token[:-1] params[param.name()] = token definition['params'] = params for i, out in enumerate([out for out in alg.destinationParameterDefinitions() if not out.flags() & QgsProcessingParameterDefinition.FlagHidden]): if not out.name() in parameters: continue token = parameters[out.name()] if isinstance(out, QgsProcessingParameterRasterDestination): if token is None: QMessageBox.warning(None, tr('Error'), tr('Seems some outputs are temporary ' 'files. To create test you need to ' 'redirect all algorithm outputs to ' 'files')) return dataset = gdal.Open(token, GA_ReadOnly) if dataset is None: QMessageBox.warning(None, tr('Error'), tr('Seems some outputs are temporary ' 'files. To create test you need to ' 'redirect all algorithm outputs to ' 'files')) return dataArray = nan_to_num(dataset.ReadAsArray(0)) strhash = hashlib.sha224(dataArray.data).hexdigest() results[out.name()] = { 'type': 'rasterhash', 'hash': strhash } elif isinstance(out, (QgsProcessingParameterVectorDestination, QgsProcessingParameterFeatureSink)): schema, filepath = extractSchemaPath(token) results[out.name()] = { 'type': 'vector', 'name': filepath } if not schema: results[out.name()]['location'] = '[The expected result data is not in the testdata directory. Please write it to processing/tests/testdata/expected. Prefer gml files.]' elif isinstance(out, QgsProcessingParameterFileDestination): schema, filepath = extractSchemaPath(token) results[out.name()] = { 'type': 'file', 'name': filepath } if not schema: results[out.name()]['location'] = '[The expected result file is not in the testdata directory. Please redirect the output to processing/tests/testdata/expected.]' definition['results'] = results dlg = ShowTestDialog(yaml.dump([definition], default_flow_style=False)) dlg.exec_() def tr(string): return QCoreApplication.translate('TestTools', string) class ShowTestDialog(QDialog): def __init__(self, s): QDialog.__init__(self) self.setModal(True) self.resize(600, 400) self.setWindowTitle(self.tr('Unit Test')) layout = QVBoxLayout() self.text = QTextEdit() self.text.setFontFamily("monospace") self.text.setEnabled(True) # Add two spaces in front of each text for faster copy/paste self.text.setText(' {}'.format(s.replace('\n', '\n '))) layout.addWidget(self.text) self.setLayout(layout) QMetaObject.connectSlotsByName(self)