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
151
152
153
154
155
156
|
##################################################################
# Copyright 2018 Open Source Geospatial Foundation and others #
# licensed under MIT, Please consult LICENSE.txt for details #
##################################################################
__author__ = "Luis de Sousa"
__date__ = "10-03-2015"
from basic import TestBase
import os
import tempfile
from pywps import Service, Process, ComplexInput, ComplexOutput, Format, FORMATS, get_format
from pywps.exceptions import NoApplicableCode
from pywps import get_ElementMakerForVersion
import pywps.configuration as config
from pywps.tests import client_for, assert_response_success, service_ok
import pytest
wfsResource = 'https://demo.mapserver.org/cgi-bin/wfs?service=WFS&version=1.1.0&request=GetFeature&typename=continents&maxfeatures=10' # noqa
wcsResource = 'https://demo.mapserver.org/cgi-bin/wcs?service=WCS&version=1.0.0&request=GetCoverage&coverage=ndvi&crs=EPSG:4326&bbox=-92,42,-85,45&format=image/tiff&width=400&height=300' # noqa
WPS, OWS = get_ElementMakerForVersion("1.0.0")
class ExecuteTests(TestBase):
def create_feature(self):
def feature(request, response):
input = request.inputs['input'][0].file
response.outputs['output'].data_format = FORMATS.GML
response.outputs['output'].file = input
return response
return Process(handler=feature,
identifier='feature',
title='Process Feature',
inputs=[ComplexInput(
'input',
title='Input',
supported_formats=[get_format('GML')])],
outputs=[ComplexOutput(
'output',
title='Output',
supported_formats=[get_format('GML')])])
def create_sum_one(self):
def sum_one(request, response):
input = request.inputs['input'][0].file
# What do we need to assert a Complex input?
# assert type(input) is str
import grass.script as grass
# Import the raster and set the region
if grass.run_command("r.in.gdal", flags="o", out="input",
input=input, quiet=True) != 0:
raise NoApplicableCode("Could not import cost map. "
"Please check the WCS service.")
if grass.run_command("g.region", flags="a", rast="input") != 0:
raise NoApplicableCode("Could not set GRASS region.")
# Add 1
if grass.mapcalc("$output = $input + $value", output="output",
input="input", value=1.0, quiet=True):
raise NoApplicableCode("Could not use GRASS map calculator.")
# Export the result
_, out = tempfile.mkstemp(dir=self.tmpdir.name)
os.environ['GRASS_VERBOSE'] = '-1'
if grass.run_command("r.out.gdal", flags="f", input="output",
type="UInt16", output=out,
overwrite=True) != 0:
raise NoApplicableCode("Could not export result from GRASS.")
del os.environ['GRASS_VERBOSE']
response.outputs['output'].file = out
return response
return Process(handler=sum_one,
identifier='sum_one',
title='Process Sum One',
inputs=[ComplexInput(
'input',
title='Input',
supported_formats=[Format('image/img')])],
outputs=[ComplexOutput(
'output',
title='Output',
supported_formats=[get_format('GEOTIFF')])],
grass_location='epsg:4326')
@pytest.mark.online
def test_wfs(self):
if not service_ok('https://demo.mapserver.org'):
self.skipTest("mapserver is unreachable")
client = client_for(Service(processes=[create_feature()]))
request_doc = WPS.Execute(
OWS.Identifier('feature'),
WPS.DataInputs(
WPS.Input(
OWS.Identifier('input'),
WPS.Reference(
{'{http://www.w3.org/1999/xlink}href': wfsResource},
mimeType=FORMATS.GML.mime_type,
encoding='',
schema=''))),
WPS.ProcessOutputs(
WPS.Output(
OWS.Identifier('output'))),
version='1.0.0'
)
resp = client.post_xml(doc=request_doc)
assert_response_success(resp)
# Other things to assert:
# . the inclusion of output
# . the type of output
@pytest.mark.online
def test_wcs(self):
if not config.CONFIG.get('grass', 'gisbase'):
self.skipTest('GRASS lib not found')
if not service_ok('https://demo.mapserver.org'):
self.skipTest("mapserver is unreachable")
client = client_for(Service(processes=[self.create_sum_one()]))
request_doc = WPS.Execute(
OWS.Identifier('sum_one'),
WPS.DataInputs(
WPS.Input(
OWS.Identifier('input'),
WPS.Reference(
{'{http://www.w3.org/1999/xlink}href': wcsResource}))),
WPS.ProcessOutputs(
WPS.Output(
OWS.Identifier('output'))),
version='1.0.0')
resp = client.post_xml(doc=request_doc)
assert_response_success(resp)
# Other things to assert:
# . the inclusion of output
# . the type of output
def load_tests(loader=None, tests=None, pattern=None):
import unittest
if not loader:
loader = unittest.TestLoader()
suite_list = [
loader.loadTestsFromTestCase(ExecuteTests),
]
return unittest.TestSuite(suite_list)
|
