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