1# -*- coding: utf-8 -*- 2"""QGIS Unit tests for qgis_process. 3 4.. note:: This program is free software; you can redistribute it and/or modify 5it under the terms of the GNU General Public License as published by 6the Free Software Foundation; either version 2 of the License, or 7(at your option) any later version. 8""" 9__author__ = '(C) 2020 by Nyall Dawson' 10__date__ = '05/04/2020' 11__copyright__ = 'Copyright 2020, The QGIS Project' 12 13import sys 14import os 15import glob 16import re 17import time 18import shutil 19import subprocess 20import tempfile 21import json 22import errno 23 24from qgis.testing import unittest 25from utilities import unitTestDataPath 26 27print('CTEST_FULL_OUTPUT') 28 29TEST_DATA_DIR = unitTestDataPath() 30 31 32class TestQgsProcessExecutable(unittest.TestCase): 33 34 TMP_DIR = '' 35 36 @classmethod 37 def setUpClass(cls): 38 cls.TMP_DIR = tempfile.mkdtemp() 39 # print('TMP_DIR: ' + cls.TMP_DIR) 40 # subprocess.call(['open', cls.TMP_DIR]) 41 42 @classmethod 43 def tearDownClass(cls): 44 shutil.rmtree(cls.TMP_DIR, ignore_errors=True) 45 46 def run_process(self, arguments): 47 call = [QGIS_PROCESS_BIN] + arguments 48 print(' '.join(call)) 49 50 myenv = os.environ.copy() 51 myenv["QGIS_DEBUG"] = '0' 52 53 p = subprocess.Popen(call, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=myenv) 54 output, err = p.communicate() 55 rc = p.returncode 56 57 return rc, output.decode(), err.decode() 58 59 def testNoArgs(self): 60 rc, output, err = self.run_process([]) 61 self.assertIn('Available commands', output) 62 if os.environ.get('TRAVIS', '') != 'true': 63 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 64 self.assertFalse(err) 65 self.assertEqual(rc, 0) 66 67 def testPlugins(self): 68 rc, output, err = self.run_process(['plugins']) 69 self.assertIn('available plugins', output.lower()) 70 self.assertIn('processing', output.lower()) 71 self.assertNotIn('metasearch', output.lower()) 72 if os.environ.get('TRAVIS', '') != 'true': 73 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 74 self.assertFalse(err) 75 self.assertEqual(rc, 0) 76 77 def testPluginsJson(self): 78 rc, output, err = self.run_process(['plugins', '--json']) 79 res = json.loads(output) 80 self.assertIn('gdal_version', res) 81 self.assertIn('geos_version', res) 82 self.assertIn('proj_version', res) 83 self.assertIn('python_version', res) 84 self.assertIn('qt_version', res) 85 self.assertIn('qgis_version', res) 86 self.assertIn('plugins', res) 87 self.assertIn('processing', res['plugins']) 88 self.assertTrue(res['plugins']['processing']['loaded']) 89 self.assertEqual(rc, 0) 90 91 def testAlgorithmList(self): 92 rc, output, err = self.run_process(['list']) 93 self.assertIn('available algorithms', output.lower()) 94 self.assertIn('native:reprojectlayer', output.lower()) 95 if os.environ.get('TRAVIS', '') != 'true': 96 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 97 self.assertFalse(err) 98 self.assertEqual(rc, 0) 99 100 def testAlgorithmsListJson(self): 101 rc, output, err = self.run_process(['list', '--json']) 102 res = json.loads(output) 103 self.assertIn('gdal_version', res) 104 self.assertIn('geos_version', res) 105 self.assertIn('proj_version', res) 106 self.assertIn('python_version', res) 107 self.assertIn('qt_version', res) 108 self.assertIn('qgis_version', res) 109 110 self.assertIn('providers', res) 111 self.assertIn('native', res['providers']) 112 self.assertTrue(res['providers']['native']['is_active']) 113 self.assertIn('native:buffer', res['providers']['native']['algorithms']) 114 self.assertFalse(res['providers']['native']['algorithms']['native:buffer']['deprecated']) 115 116 self.assertEqual(rc, 0) 117 118 def testAlgorithmHelpNoAlg(self): 119 rc, output, err = self.run_process(['help']) 120 self.assertEqual(rc, 1) 121 self.assertIn('algorithm id or model file not specified', err.lower()) 122 self.assertFalse(output) 123 124 def testAlgorithmHelp(self): 125 rc, output, err = self.run_process(['help', 'native:centroids']) 126 self.assertIn('representing the centroid', output.lower()) 127 self.assertIn('argument type', output.lower()) 128 if os.environ.get('TRAVIS', '') != 'true': 129 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 130 self.assertFalse(err) 131 self.assertEqual(rc, 0) 132 133 def testAlgorithmHelpJson(self): 134 rc, output, err = self.run_process(['help', 'native:buffer', '--json']) 135 res = json.loads(output) 136 137 self.assertIn('gdal_version', res) 138 self.assertIn('geos_version', res) 139 self.assertIn('proj_version', res) 140 self.assertIn('python_version', res) 141 self.assertIn('qt_version', res) 142 self.assertIn('qgis_version', res) 143 144 self.assertFalse(res['algorithm_details']['deprecated']) 145 self.assertTrue(res['provider_details']['is_active']) 146 147 self.assertIn('OUTPUT', res['outputs']) 148 self.assertEqual(res['outputs']['OUTPUT']['description'], 'Buffered') 149 self.assertEqual(res['parameters']['DISSOLVE']['description'], 'Dissolve result') 150 self.assertFalse(res['parameters']['DISTANCE']['is_advanced']) 151 152 self.assertEqual(rc, 0) 153 154 def testAlgorithmRunNoAlg(self): 155 rc, output, err = self.run_process(['run']) 156 self.assertIn('algorithm id or model file not specified', err.lower()) 157 self.assertFalse(output) 158 self.assertEqual(rc, 1) 159 160 def testAlgorithmRunNoArgs(self): 161 rc, output, err = self.run_process(['run', 'native:centroids']) 162 self.assertIn('the following mandatory parameters were not specified', err.lower()) 163 self.assertIn('inputs', output.lower()) 164 self.assertEqual(rc, 1) 165 166 def testAlgorithmRunLegacy(self): 167 output_file = self.TMP_DIR + '/polygon_centroid.shp' 168 rc, output, err = self.run_process(['run', 'native:centroids', '--INPUT={}'.format(TEST_DATA_DIR + '/polys.shp'), '--OUTPUT={}'.format(output_file)]) 169 if os.environ.get('TRAVIS', '') != 'true': 170 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 171 self.assertFalse(err) 172 self.assertIn('0...10...20...30...40...50...60...70...80...90', output.lower()) 173 self.assertIn('results', output.lower()) 174 self.assertIn('OUTPUT:\t' + output_file, output) 175 self.assertTrue(os.path.exists(output_file)) 176 self.assertEqual(rc, 0) 177 178 def testAlgorithmRun(self): 179 output_file = self.TMP_DIR + '/polygon_centroid.shp' 180 rc, output, err = self.run_process(['run', 'native:centroids', '--', 'INPUT={}'.format(TEST_DATA_DIR + '/polys.shp'), 'OUTPUT={}'.format(output_file)]) 181 if os.environ.get('TRAVIS', '') != 'true': 182 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 183 self.assertFalse(err) 184 self.assertIn('0...10...20...30...40...50...60...70...80...90', output.lower()) 185 self.assertIn('results', output.lower()) 186 self.assertIn('OUTPUT:\t' + output_file, output) 187 self.assertTrue(os.path.exists(output_file)) 188 self.assertEqual(rc, 0) 189 190 def testAlgorithmRunJson(self): 191 output_file = self.TMP_DIR + '/polygon_centroid2.shp' 192 rc, output, err = self.run_process(['run', '--json', 'native:centroids', '--', 'INPUT={}'.format(TEST_DATA_DIR + '/polys.shp'), 'OUTPUT={}'.format(output_file)]) 193 res = json.loads(output) 194 195 self.assertIn('gdal_version', res) 196 self.assertIn('geos_version', res) 197 self.assertIn('proj_version', res) 198 self.assertIn('python_version', res) 199 self.assertIn('qt_version', res) 200 self.assertIn('qgis_version', res) 201 202 self.assertEqual(res['algorithm_details']['name'], 'Centroids') 203 self.assertEqual(res['inputs']['INPUT'], TEST_DATA_DIR + '/polys.shp') 204 self.assertEqual(res['inputs']['OUTPUT'], output_file) 205 self.assertEqual(res['results']['OUTPUT'], output_file) 206 207 self.assertTrue(os.path.exists(output_file)) 208 self.assertEqual(rc, 0) 209 210 def testAlgorithmRunListValue(self): 211 """ 212 Test an algorithm which requires a list of layers as a parameter value 213 """ 214 output_file = self.TMP_DIR + '/package.gpkg' 215 rc, output, err = self.run_process(['run', '--json', 'native:package', '--', 216 'LAYERS={}'.format(TEST_DATA_DIR + '/polys.shp'), 217 'LAYERS={}'.format(TEST_DATA_DIR + '/points.shp'), 218 'LAYERS={}'.format(TEST_DATA_DIR + '/lines.shp'), 219 'OUTPUT={}'.format(output_file)]) 220 res = json.loads(output) 221 222 self.assertIn('gdal_version', res) 223 self.assertIn('geos_version', res) 224 self.assertIn('proj_version', res) 225 self.assertIn('python_version', res) 226 self.assertIn('qt_version', res) 227 self.assertIn('qgis_version', res) 228 229 self.assertEqual(res['algorithm_details']['name'], 'Package layers') 230 self.assertEqual(len(res['inputs']['LAYERS']), 3) 231 self.assertEqual(res['inputs']['OUTPUT'], output_file) 232 self.assertEqual(res['results']['OUTPUT'], output_file) 233 self.assertEqual(len(res['results']['OUTPUT_LAYERS']), 3) 234 235 self.assertTrue(os.path.exists(output_file)) 236 self.assertEqual(rc, 0) 237 238 def testModelHelp(self): 239 rc, output, err = self.run_process(['help', TEST_DATA_DIR + '/test_model.model3']) 240 if os.environ.get('TRAVIS', '') != 'true': 241 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 242 self.assertFalse(err) 243 self.assertEqual(rc, 0) 244 self.assertIn('model description', output.lower()) 245 246 def testModelRun(self): 247 output_file = self.TMP_DIR + '/model_output.shp' 248 rc, output, err = self.run_process(['run', TEST_DATA_DIR + '/test_model.model3', '--', 'FEATS={}'.format(TEST_DATA_DIR + '/polys.shp'), 'native:centroids_1:CENTROIDS={}'.format(output_file)]) 249 if os.environ.get('TRAVIS', '') != 'true': 250 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 251 self.assertFalse(err) 252 self.assertEqual(rc, 0) 253 self.assertIn('0...10...20...30...40...50...60...70...80...90', output.lower()) 254 self.assertIn('results', output.lower()) 255 self.assertTrue(os.path.exists(output_file)) 256 257 def testModelRunJson(self): 258 output_file = self.TMP_DIR + '/model_output2.shp' 259 rc, output, err = self.run_process(['run', TEST_DATA_DIR + '/test_model.model3', '--json', '--', 'FEATS={}'.format(TEST_DATA_DIR + '/polys.shp'), 'native:centroids_1:CENTROIDS={}'.format(output_file)]) 260 if os.environ.get('TRAVIS', '') != 'true': 261 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 262 self.assertFalse(err) 263 self.assertEqual(rc, 0) 264 265 res = json.loads(output) 266 self.assertIn('gdal_version', res) 267 self.assertIn('geos_version', res) 268 self.assertIn('proj_version', res) 269 self.assertIn('python_version', res) 270 self.assertIn('qt_version', res) 271 self.assertIn('qgis_version', res) 272 self.assertEqual(res['algorithm_details']['id'], 'Test model') 273 self.assertTrue(os.path.exists(output_file)) 274 275 def testModelRunWithLog(self): 276 output_file = self.TMP_DIR + '/model_log.log' 277 rc, output, err = self.run_process(['run', TEST_DATA_DIR + '/test_logging_model.model3', '--', 'logfile={}'.format(output_file)]) 278 self.assertIn('Test logged message', err) 279 self.assertEqual(rc, 0) 280 self.assertIn('0...10...20...30...40...50...60...70...80...90', output.lower()) 281 self.assertIn('results', output.lower()) 282 self.assertTrue(os.path.exists(output_file)) 283 284 with open(output_file, 'rt') as f: 285 lines = '\n'.join(f.readlines()) 286 287 self.assertIn('Test logged message', lines) 288 289 290if __name__ == '__main__': 291 # look for qgis bin path 292 QGIS_PROCESS_BIN = '' 293 prefixPath = os.environ['QGIS_PREFIX_PATH'] 294 # see qgsapplication.cpp:98 295 for f in ['', '..', 'bin']: 296 d = os.path.join(prefixPath, f) 297 b = os.path.abspath(os.path.join(d, 'qgis_process')) 298 if os.path.exists(b): 299 QGIS_PROCESS_BIN = b 300 break 301 b = os.path.abspath(os.path.join(d, 'qgis_process.exe')) 302 if os.path.exists(b): 303 QGIS_PROCESS_BIN = b 304 break 305 if sys.platform[:3] == 'dar': # Mac 306 # QGIS.app may be QGIS_x.x-dev.app for nightlies 307 # internal binary will match, minus the '.app' 308 found = False 309 for app_path in glob.glob(d + '/QGIS*.app'): 310 m = re.search(r'/(QGIS(_\d\.\d-dev)?)\.app', app_path) 311 if m: 312 QGIS_PROCESS_BIN = app_path + '/Contents/MacOS/' + m.group(1) 313 found = True 314 break 315 if found: 316 break 317 318 print(('\nQGIS_PROCESS_BIN: {}'.format(QGIS_PROCESS_BIN))) 319 assert QGIS_PROCESS_BIN, 'qgis_process binary not found, skipping test suite' 320 unittest.main() 321