-
Notifications
You must be signed in to change notification settings - Fork 120
/
tests.py
180 lines (145 loc) · 6.3 KB
/
tests.py
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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
##################################################################
# Copyright 2018 Open Source Geospatial Foundation and others #
# licensed under MIT, Please consult LICENSE.txt for details #
##################################################################
import tempfile
from pathlib import Path
import lxml
from pywps import xml_util as etree
import requests
from werkzeug.test import Client
from werkzeug.wrappers import Response
from pywps import __version__
from pywps import Process
from pywps.inout import LiteralInput, LiteralOutput, ComplexInput, ComplexOutput, BoundingBoxInput, BoundingBoxOutput
from pywps.inout import Format
from pywps.app.Common import Metadata, MetadataUrl
import re
import logging
logging.disable(logging.CRITICAL)
def service_ok(url, timeout=5):
try:
resp = requests.get(url, timeout=timeout)
if 'html' in resp.headers['content-type']:
ok = False
else:
ok = resp.ok
except requests.exceptions.ReadTimeout:
ok = False
except requests.exceptions.ConnectTimeout:
ok = False
except Exception:
ok = False
return ok
class DocExampleProcess(Process):
"""This first line is going to be skipped by the :skiplines:1 option.
Notes
-----
This is additional documentation that can be added following the Numpy docstring convention.
"""
def __init__(self):
inputs = [
LiteralInput(
'literal_input', "Literal input title", 'integer', abstract="Literal input value abstract.",
min_occurs=0, max_occurs=1, uoms=['meters', 'feet'], default=1
),
LiteralInput('date_input', 'The title is shown when no abstract is provided.', 'date',
allowed_values=['2000-01-01', '2018-01-01']),
ComplexInput('complex_input', 'Complex input title',
[Format('application/json'), Format('application/x-netcdf')],
abstract="Complex input abstract.", ),
BoundingBoxInput('bb_input', 'BoundingBox input title', ['EPSG:4326', ],
metadata=[Metadata('EPSG.io', 'http://epsg.io/'), ]),
]
outputs = [
LiteralOutput(
'literal_output', 'Literal output title', 'boolean', abstract='Boolean output abstract.'
),
ComplexOutput('complex_output', 'Complex output', [Format('text/plain'), ], ),
BoundingBoxOutput('bb_output', 'BoundingBox output title', ['EPSG:4326', ])
]
super(DocExampleProcess, self).__init__(
self._handler,
identifier='doc_example_process_identifier',
title="Process title",
abstract="Multiline process abstract.",
version="4.0",
metadata=[Metadata('PyWPS docs', 'https://pywps.org'),
Metadata('NumPy docstring conventions',
'https://github.com/numpy/numpy/blob/master/doc/HOWTO_DOCUMENT.rst.txt'),
MetadataUrl('Duplicate label', 'http://one.example.com', anonymous=True),
MetadataUrl('Duplicate label', 'http://two.example.com', anonymous=True),
],
inputs=inputs,
outputs=outputs,
)
def _handler(self, request, response):
pass
class WpsClient(Client):
def post_xml(self, *args, **kwargs):
doc = kwargs.pop('doc')
data = etree.tostring(doc, pretty_print=True)
kwargs['data'] = data
return self.post(*args, **kwargs)
class WpsTestResponse(Response):
def __init__(self, *args):
super(WpsTestResponse, self).__init__(*args)
if re.match(r'text/xml(;\s*charset=.*)?', self.headers.get('Content-Type')):
self.xml = etree.fromstring(self.get_data())
def xpath(self, path):
version = self.xml.attrib["version"]
if version == "2.0.0":
from pywps import namespaces200
namespaces = namespaces200
else:
from pywps import namespaces100
namespaces = namespaces100
return self.xml.xpath(path, namespaces=namespaces)
def xpath_text(self, path):
return ' '.join(e.text for e in self.xpath(path))
def client_for(service):
return WpsClient(service, WpsTestResponse)
def assert_response_accepted(resp):
assert resp.status_code == 200
assert re.match(r'text/xml(;\s*charset=.*)?', resp.headers['Content-Type'])
success = resp.xpath_text('/wps:ExecuteResponse'
'/wps:Status'
'/wps:ProcessAccepted')
assert success is not None
# TODO: assert status URL is present
def assert_process_started(resp):
assert resp.status_code == 200
assert re.match(r'text/xml(;\s*charset=.*)?', resp.headers['Content-Type'])
success = resp.xpath_text('/wps:ExecuteResponse'
'/wps:Status'
'ProcessStarted')
# Is it still like this in PyWPS-4 ?
assert success.split[0] == "processstarted"
def assert_response_success(resp):
assert resp.status_code == 200
assert re.match(r'text/xml(;\s*charset=.*)?', resp.headers['Content-Type'])
success = resp.xpath('/wps:ExecuteResponse/wps:Status/wps:ProcessSucceeded')
assert len(success) == 1
def assert_process_exception(resp, code=None):
assert resp.status_code == 400
assert re.match(r'text/xml(;\s*charset=.*)?', resp.headers['Content-Type'])
elem = resp.xpath('/ows:ExceptionReport'
'/ows:Exception')
assert elem[0].attrib['exceptionCode'] == code
def assert_pywps_version(resp):
# get first child of root element
root_firstchild = resp.xpath('/*')[0].getprevious()
assert isinstance(root_firstchild, lxml.etree._Comment)
tokens = root_firstchild.text.split()
assert len(tokens) == 2
assert tokens[0] == 'PyWPS'
assert tokens[1] == __version__
def assert_wps_version(response, version="1.0.0"):
elem = response.xpath('/wps:Capabilities'
'/ows:ServiceIdentification'
'/ows:ServiceTypeVersion')
found_version = elem[0].text
assert version == found_version
tmp = Path(tempfile.mkdtemp())
with open(tmp / "out.xml", "wb") as out:
out.writelines(response.response)