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('qt_version', res) 84 self.assertIn('qgis_version', res) 85 self.assertIn('plugins', res) 86 self.assertIn('processing', res['plugins']) 87 self.assertTrue(res['plugins']['processing']['loaded']) 88 self.assertEqual(rc, 0) 89 90 def testAlgorithmList(self): 91 rc, output, err = self.run_process(['list']) 92 self.assertIn('available algorithms', output.lower()) 93 self.assertIn('native:reprojectlayer', output.lower()) 94 if os.environ.get('TRAVIS', '') != 'true': 95 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 96 self.assertFalse(err) 97 self.assertEqual(rc, 0) 98 99 def testAlgorithmsListJson(self): 100 rc, output, err = self.run_process(['list', '--json']) 101 res = json.loads(output) 102 self.assertIn('gdal_version', res) 103 self.assertIn('geos_version', res) 104 self.assertIn('proj_version', res) 105 self.assertIn('qt_version', res) 106 self.assertIn('qgis_version', res) 107 108 self.assertIn('providers', res) 109 self.assertIn('native', res['providers']) 110 self.assertTrue(res['providers']['native']['is_active']) 111 self.assertIn('native:buffer', res['providers']['native']['algorithms']) 112 self.assertFalse(res['providers']['native']['algorithms']['native:buffer']['deprecated']) 113 114 self.assertEqual(rc, 0) 115 116 def testAlgorithmHelpNoAlg(self): 117 rc, output, err = self.run_process(['help']) 118 self.assertEqual(rc, 1) 119 self.assertIn('algorithm id or model file not specified', err.lower()) 120 self.assertFalse(output) 121 122 def testAlgorithmHelp(self): 123 rc, output, err = self.run_process(['help', 'native:centroids']) 124 self.assertIn('representing the centroid', output.lower()) 125 self.assertIn('argument type', output.lower()) 126 if os.environ.get('TRAVIS', '') != 'true': 127 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 128 self.assertFalse(err) 129 self.assertEqual(rc, 0) 130 131 def testAlgorithmHelpJson(self): 132 rc, output, err = self.run_process(['help', 'native:buffer', '--json']) 133 res = json.loads(output) 134 135 self.assertIn('gdal_version', res) 136 self.assertIn('geos_version', res) 137 self.assertIn('proj_version', res) 138 self.assertIn('qt_version', res) 139 self.assertIn('qgis_version', res) 140 141 self.assertFalse(res['algorithm_details']['deprecated']) 142 self.assertTrue(res['provider_details']['is_active']) 143 144 self.assertIn('OUTPUT', res['outputs']) 145 self.assertEqual(res['outputs']['OUTPUT']['description'], 'Buffered') 146 self.assertEqual(res['parameters']['DISSOLVE']['description'], 'Dissolve result') 147 self.assertFalse(res['parameters']['DISTANCE']['is_advanced']) 148 149 self.assertEqual(rc, 0) 150 151 def testAlgorithmRunNoAlg(self): 152 rc, output, err = self.run_process(['run']) 153 self.assertIn('algorithm id or model file not specified', err.lower()) 154 self.assertFalse(output) 155 self.assertEqual(rc, 1) 156 157 def testAlgorithmRunNoArgs(self): 158 rc, output, err = self.run_process(['run', 'native:centroids']) 159 self.assertIn('the following mandatory parameters were not specified', err.lower()) 160 self.assertIn('inputs', output.lower()) 161 self.assertEqual(rc, 1) 162 163 def testAlgorithmRunLegacy(self): 164 output_file = self.TMP_DIR + '/polygon_centroid.shp' 165 rc, output, err = self.run_process(['run', 'native:centroids', '--INPUT={}'.format(TEST_DATA_DIR + '/polys.shp'), '--OUTPUT={}'.format(output_file)]) 166 if os.environ.get('TRAVIS', '') != 'true': 167 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 168 self.assertFalse(err) 169 self.assertIn('0...10...20...30...40...50...60...70...80...90', output.lower()) 170 self.assertIn('results', output.lower()) 171 self.assertIn('OUTPUT:\t' + output_file, output) 172 self.assertTrue(os.path.exists(output_file)) 173 self.assertEqual(rc, 0) 174 175 def testAlgorithmRun(self): 176 output_file = self.TMP_DIR + '/polygon_centroid.shp' 177 rc, output, err = self.run_process(['run', 'native:centroids', '--', 'INPUT={}'.format(TEST_DATA_DIR + '/polys.shp'), 'OUTPUT={}'.format(output_file)]) 178 if os.environ.get('TRAVIS', '') != 'true': 179 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 180 self.assertFalse(err) 181 self.assertIn('0...10...20...30...40...50...60...70...80...90', output.lower()) 182 self.assertIn('results', output.lower()) 183 self.assertIn('OUTPUT:\t' + output_file, output) 184 self.assertTrue(os.path.exists(output_file)) 185 self.assertEqual(rc, 0) 186 187 def testAlgorithmRunJson(self): 188 output_file = self.TMP_DIR + '/polygon_centroid2.shp' 189 rc, output, err = self.run_process(['run', '--json', 'native:centroids', '--', 'INPUT={}'.format(TEST_DATA_DIR + '/polys.shp'), 'OUTPUT={}'.format(output_file)]) 190 res = json.loads(output) 191 192 self.assertIn('gdal_version', res) 193 self.assertIn('geos_version', res) 194 self.assertIn('proj_version', res) 195 self.assertIn('qt_version', res) 196 self.assertIn('qgis_version', res) 197 198 self.assertEqual(res['algorithm_details']['name'], 'Centroids') 199 self.assertEqual(res['inputs']['INPUT'], TEST_DATA_DIR + '/polys.shp') 200 self.assertEqual(res['inputs']['OUTPUT'], output_file) 201 self.assertEqual(res['results']['OUTPUT'], output_file) 202 203 self.assertTrue(os.path.exists(output_file)) 204 self.assertEqual(rc, 0) 205 206 def testAlgorithmRunListValue(self): 207 """ 208 Test an algorithm which requires a list of layers as a parameter value 209 """ 210 output_file = self.TMP_DIR + '/package.gpkg' 211 rc, output, err = self.run_process(['run', '--json', 'native:package', '--', 212 'LAYERS={}'.format(TEST_DATA_DIR + '/polys.shp'), 213 'LAYERS={}'.format(TEST_DATA_DIR + '/points.shp'), 214 'LAYERS={}'.format(TEST_DATA_DIR + '/lines.shp'), 215 'OUTPUT={}'.format(output_file)]) 216 res = json.loads(output) 217 218 self.assertIn('gdal_version', res) 219 self.assertIn('geos_version', res) 220 self.assertIn('proj_version', res) 221 self.assertIn('qt_version', res) 222 self.assertIn('qgis_version', res) 223 224 self.assertEqual(res['algorithm_details']['name'], 'Package layers') 225 self.assertEqual(len(res['inputs']['LAYERS']), 3) 226 self.assertEqual(res['inputs']['OUTPUT'], output_file) 227 self.assertEqual(res['results']['OUTPUT'], output_file) 228 self.assertEqual(len(res['results']['OUTPUT_LAYERS']), 3) 229 230 self.assertTrue(os.path.exists(output_file)) 231 self.assertEqual(rc, 0) 232 233 def testModelHelp(self): 234 rc, output, err = self.run_process(['help', TEST_DATA_DIR + '/test_model.model3']) 235 if os.environ.get('TRAVIS', '') != 'true': 236 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 237 self.assertFalse(err) 238 self.assertEqual(rc, 0) 239 self.assertIn('model description', output.lower()) 240 241 def testModelRun(self): 242 output_file = self.TMP_DIR + '/model_output.shp' 243 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)]) 244 if os.environ.get('TRAVIS', '') != 'true': 245 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 246 self.assertFalse(err) 247 self.assertEqual(rc, 0) 248 self.assertIn('0...10...20...30...40...50...60...70...80...90', output.lower()) 249 self.assertIn('results', output.lower()) 250 self.assertTrue(os.path.exists(output_file)) 251 252 def testModelRunJson(self): 253 output_file = self.TMP_DIR + '/model_output2.shp' 254 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)]) 255 if os.environ.get('TRAVIS', '') != 'true': 256 # Travis DOES have errors, due to QStandardPaths: XDG_RUNTIME_DIR not set warnings raised by Qt 257 self.assertFalse(err) 258 self.assertEqual(rc, 0) 259 260 res = json.loads(output) 261 self.assertIn('gdal_version', res) 262 self.assertIn('geos_version', res) 263 self.assertIn('proj_version', res) 264 self.assertIn('qt_version', res) 265 self.assertIn('qgis_version', res) 266 self.assertEqual(res['algorithm_details']['id'], 'Test model') 267 self.assertTrue(os.path.exists(output_file)) 268 269 def testModelRunWithLog(self): 270 output_file = self.TMP_DIR + '/model_log.log' 271 rc, output, err = self.run_process(['run', TEST_DATA_DIR + '/test_logging_model.model3', '--', 'logfile={}'.format(output_file)]) 272 self.assertIn('Test logged message', err) 273 self.assertEqual(rc, 0) 274 self.assertIn('0...10...20...30...40...50...60...70...80...90', output.lower()) 275 self.assertIn('results', output.lower()) 276 self.assertTrue(os.path.exists(output_file)) 277 278 with open(output_file, 'rt') as f: 279 lines = '\n'.join(f.readlines()) 280 281 self.assertIn('Test logged message', lines) 282 283 284if __name__ == '__main__': 285 # look for qgis bin path 286 QGIS_PROCESS_BIN = '' 287 prefixPath = os.environ['QGIS_PREFIX_PATH'] 288 # see qgsapplication.cpp:98 289 for f in ['', '..', 'bin']: 290 d = os.path.join(prefixPath, f) 291 b = os.path.abspath(os.path.join(d, 'qgis_process')) 292 if os.path.exists(b): 293 QGIS_PROCESS_BIN = b 294 break 295 b = os.path.abspath(os.path.join(d, 'qgis_process.exe')) 296 if os.path.exists(b): 297 QGIS_PROCESS_BIN = b 298 break 299 if sys.platform[:3] == 'dar': # Mac 300 # QGIS.app may be QGIS_x.x-dev.app for nightlies 301 # internal binary will match, minus the '.app' 302 found = False 303 for app_path in glob.glob(d + '/QGIS*.app'): 304 m = re.search(r'/(QGIS(_\d\.\d-dev)?)\.app', app_path) 305 if m: 306 QGIS_PROCESS_BIN = app_path + '/Contents/MacOS/' + m.group(1) 307 found = True 308 break 309 if found: 310 break 311 312 print(('\nQGIS_PROCESS_BIN: {}'.format(QGIS_PROCESS_BIN))) 313 assert QGIS_PROCESS_BIN, 'qgis_process binary not found, skipping test suite' 314 unittest.main() 315