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