Skip to content

Commit

Permalink
Fixed tests with complexinput, update of #45. (#48)
Browse files Browse the repository at this point in the history
  • Loading branch information
cehbrecht authored Jul 31, 2018
1 parent 482a9fa commit 3502572
Show file tree
Hide file tree
Showing 10 changed files with 2,505 additions and 39 deletions.
4 changes: 2 additions & 2 deletions emu/processes/wps_poly_centroid.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ def __init__(self):
supported_formats=[FORMATS.GML, ]),
]
outputs = [
LiteralOutput('centroid', 'The centroid of the polygon geometry.',
LiteralOutput('output', 'The centroid of the polygon geometry.',
abstract="The coordinates of the polygon's approximate centroid.",)
]

Expand Down Expand Up @@ -56,5 +56,5 @@ def _handler(request, response):
cx = sum(x) / n
cy = sum(y) / n

response.outputs['centroid'].data = '{},{}'.format(cx, cy)
response.outputs['output'].data = '{:.5f},{:.5f}'.format(cx, cy)
return response
2 changes: 1 addition & 1 deletion emu/processes/wps_wordcounter.py
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ def _handler(request, response):

def words(f):
for line in f:
for word in wordre.findall(line):
for word in wordre.findall(line.decode('UTF-8')):
yield word

counts = Counter(words(request.inputs['text'][0].stream))
Expand Down
25 changes: 25 additions & 0 deletions tests/common.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,18 @@
import os
from pywps import get_ElementMakerForVersion
from pywps.app.basic import get_xpath_ns
from pywps.tests import WpsClient, WpsTestResponse

TESTS_HOME = os.path.abspath(os.path.dirname(__file__))

VERSION = "1.0.0"
WPS, OWS = get_ElementMakerForVersion(VERSION)
xpath_ns = get_xpath_ns(VERSION)


def resource_file(filepath):
return os.path.join(TESTS_HOME, 'testdata', filepath)


class WpsTestClient(WpsClient):

Expand All @@ -12,3 +25,15 @@ def get(self, *args, **kwargs):

def client_for(service):
return WpsTestClient(service, WpsTestResponse)


def get_output(doc):
"""Copied from pywps/tests/test_execute.py.
TODO: make this helper method public in pywps."""
output = {}
for output_el in xpath_ns(doc, '/wps:ExecuteResponse'
'/wps:ProcessOutputs/wps:Output'):
[identifier_el] = xpath_ns(output_el, './ows:Identifier')
[value_el] = xpath_ns(output_el, './wps:Data/wps:LiteralData')
output[identifier_el.text] = value_el.text
return output
3 changes: 2 additions & 1 deletion tests/test_wps_dummy.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
from pywps import Service
from pywps.tests import assert_response_success

from .common import client_for
from .common import client_for, get_output
from emu.processes.wps_dummy import Dummy


Expand All @@ -13,3 +13,4 @@ def test_wps_dummy():
identifier='dummyprocess',
datainputs=datainputs)
assert_response_success(resp)
assert get_output(resp.xml) == {'output1': '11', 'output2': '1'}
24 changes: 5 additions & 19 deletions tests/test_wps_hello.py
Original file line number Diff line number Diff line change
@@ -1,31 +1,17 @@
import pytest

from pywps import Service
from pywps.tests import assert_response_success
from pywps.tests import client_for, assert_response_success

from .common import client_for
from .common import get_output
from emu.processes.wps_say_hello import SayHello


def test_wps_hello():
client = client_for(Service(processes=[SayHello()]))
datainputs = "name=LovelySugarBird"
resp = client.get(
service='WPS', request='Execute', version='1.0.0', identifier='hello',
datainputs=datainputs)
"?service=WPS&request=Execute&version=1.0.0&identifier=hello&datainputs={}".format(
datainputs))
assert_response_success(resp)


@pytest.mark.skip(reason="build_request_response method not available anymore")
def test_wps_hello_again():
"""Example of how to debug this process, running outside a PyWPS instance.
"""
hello = SayHello()
(request, response) = hello.build_request_response()
literal_in = hello.inputs[0]
literal_in.data = 'Alice'
request.inputs["name"].append(literal_in)
hello._handler(request, response)

assert response.outputs["output"].data == 'Hello Alice'
print("All good!")
assert get_output(resp.xml) == {'output': "Hello LovelySugarBird"}
31 changes: 22 additions & 9 deletions tests/test_wps_poly_centroid.py
Original file line number Diff line number Diff line change
@@ -1,24 +1,37 @@
import pytest
import os
import requests
from pywps import Service
from pywps.tests import assert_response_success

from .common import client_for
import emu
from .common import client_for, resource_file, TESTS_HOME, WPS, OWS, get_output
from emu.processes.wps_poly_centroid import PolyCentroid
from eggshell.config import Paths

paths = Paths(emu)
TESTS_HOME = os.path.abspath(os.path.dirname(__file__))
cfgfiles = os.path.join(TESTS_HOME, 'test.cfg')


def test_wps_poly_centroid():
def test_wps_poly_centroid_get():
client = client_for(Service(processes=[PolyCentroid(), ], cfgfiles=cfgfiles))
fn = os.path.join(TESTS_HOME, 'testdata', 'poly.xml')
datainputs = "polygon=files@xlink:href=file://{0}".format(fn)
datainputs = "polygon=@xlink:href=file://{0}".format(resource_file('poly.xml'))
resp = client.get(
service='WPS', request='Execute', version='1.0.0',
identifier='poly_centroid',
datainputs=datainputs)
assert_response_success(resp)
assert get_output(resp.xml) == {'output': "119.59740,-13.57388"}


def test_wps_poly_centroid_post():
client = client_for(Service(processes=[PolyCentroid(), ], cfgfiles=cfgfiles))
request_doc = WPS.Execute(
OWS.Identifier('poly_centroid'),
WPS.DataInputs(
WPS.Input(
OWS.Identifier('polygon'),
WPS.Data(WPS.ComplexData(open(resource_file('poly.xml'), 'r').read()))
)
),
version='1.0.0'
)
resp = client.post_xml(doc=request_doc)
assert_response_success(resp)
assert get_output(resp.xml) == {'output': "119.59740,-13.57388"}
9 changes: 3 additions & 6 deletions tests/test_wps_wordcounter.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,14 @@
import pytest

from pywps import Service
from pywps.tests import assert_response_success

from .common import client_for
from .common import client_for, resource_file
from emu.processes.wps_wordcounter import WordCounter


@pytest.mark.online
def test_wps_wordcount():
client = client_for(Service(processes=[WordCounter()]))
datainputs = "text={0}".format(
"https://en.wikipedia.org/wiki/Web_Processing_Service")
datainputs = "text=@xlink:href=file://{0}".format(
resource_file('alice-in-wonderland.txt'))
resp = client.get(
service='wps', request='execute', version='1.0.0',
identifier='wordcounter',
Expand Down
Loading

0 comments on commit 3502572

Please sign in to comment.