1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
"""
***************************************************************************
QgisAlgorithmTests.py
---------------------
Date : January 2019
Copyright : (C) 2019 by Nyall Dawson
Email : nyall dot dawson 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__ = "Nyall Dawson"
__date__ = "January 2019"
__copyright__ = "(C) 2019, Nyall Dawson"
import nose2
import shutil
import gc
from qgis.core import (
QgsApplication,
QgsProcessing,
QgsProcessingContext,
QgsVectorLayer,
QgsProject,
)
from qgis.PyQt import sip
from qgis.analysis import QgsNativeAlgorithms
import unittest
from qgis.testing import start_app, QgisTestCase
import processing
from processing.tests.TestData import points
class TestProcessingGeneral(QgisTestCase):
@classmethod
def setUpClass(cls):
start_app()
from processing.core.Processing import Processing
Processing.initialize()
cls.cleanup_paths = []
cls.in_place_layers = {}
cls.vector_layer_params = {}
@classmethod
def tearDownClass(cls):
from processing.core.Processing import Processing
Processing.deinitialize()
for path in cls.cleanup_paths:
shutil.rmtree(path)
def testRun(self):
context = QgsProcessingContext()
# try running an alg using processing.run - ownership of result layer should be transferred back to the caller
res = processing.run(
"qgis:buffer",
{
"DISTANCE": 1,
"INPUT": points(),
"OUTPUT": QgsProcessing.TEMPORARY_OUTPUT,
},
context=context,
)
self.assertIn("OUTPUT", res)
# output should be the layer instance itself
self.assertIsInstance(res["OUTPUT"], QgsVectorLayer)
# Python should have ownership
self.assertTrue(sip.ispyowned(res["OUTPUT"]))
del context
gc.collect()
self.assertFalse(sip.isdeleted(res["OUTPUT"]))
# now try using processing.run with is_child_algorithm = True. Ownership should remain with the context
context = QgsProcessingContext()
res = processing.run(
"qgis:buffer",
{
"DISTANCE": 1,
"INPUT": points(),
"OUTPUT": QgsProcessing.TEMPORARY_OUTPUT,
},
context=context,
is_child_algorithm=True,
)
self.assertIn("OUTPUT", res)
# output should be a layer string reference, NOT the layer itself
self.assertIsInstance(res["OUTPUT"], str)
layer = context.temporaryLayerStore().mapLayer(res["OUTPUT"])
self.assertIsInstance(layer, QgsVectorLayer)
# context should have ownership
self.assertFalse(sip.ispyowned(layer))
del context
gc.collect()
self.assertTrue(sip.isdeleted(layer))
def testRunAndLoadResults(self):
QgsProject.instance().removeAllMapLayers()
context = QgsProcessingContext()
# try running an alg using processing.runAndLoadResults - ownership of result layer should be transferred to
# project, and layer should be present in project
res = processing.runAndLoadResults(
"qgis:buffer",
{
"DISTANCE": 1,
"INPUT": points(),
"OUTPUT": QgsProcessing.TEMPORARY_OUTPUT,
},
context=context,
)
self.assertIn("OUTPUT", res)
# output should be the layer path
self.assertIsInstance(res["OUTPUT"], str)
self.assertEqual(
context.layersToLoadOnCompletion()[res["OUTPUT"]].project,
QgsProject.instance(),
)
layer = QgsProject.instance().mapLayer(res["OUTPUT"])
self.assertIsInstance(layer, QgsVectorLayer)
# Python should NOT have ownership
self.assertFalse(sip.ispyowned(layer))
def testProviders(self):
"""
When run from a standalone script (like this test), ensure that the providers from separate plugins are available
"""
providers = [p.id() for p in QgsApplication.processingRegistry().providers()]
self.assertIn("qgis", providers)
self.assertIn("native", providers)
self.assertIn("gdal", providers)
self.assertIn("project", providers)
self.assertIn("script", providers)
self.assertIn("model", providers)
self.assertIn("grass", providers)
if __name__ == "__main__":
nose2.main()
|
