[Pywps-commits] r1183 - in branches/pywps-3.2-soap: pywps/Parser tests tests/requests
scm-commit@wald.intevation.org
scm-commit at wald.intevation.org
Mon Aug 8 12:05:50 CEST 2011
Author: jesus
Date: 2011-08-08 12:05:49 +0200 (Mon, 08 Aug 2011)
New Revision: 1183
Added:
branches/pywps-3.2-soap/tests/parser.py
Removed:
branches/pywps-3.2-soap/tests/parser.py
Modified:
branches/pywps-3.2-soap/pywps/Parser/Execute.py
branches/pywps-3.2-soap/pywps/Parser/Get.py
branches/pywps-3.2-soap/pywps/Parser/Post.py
branches/pywps-3.2-soap/tests/perform_requests.py
branches/pywps-3.2-soap/tests/requests/wps_execute_request-literalinput.xml
Log:
FEATURE: There wasn't a way to input "@" as incoded "%40". Path by Andrea Francia (andrea__AT__andreafrancia.it)
Modified: branches/pywps-3.2-soap/pywps/Parser/Execute.py
===================================================================
--- branches/pywps-3.2-soap/pywps/Parser/Execute.py 2011-08-01 10:16:57 UTC (rev 1182)
+++ branches/pywps-3.2-soap/pywps/Parser/Execute.py 2011-08-08 10:05:49 UTC (rev 1183)
@@ -28,7 +28,7 @@
from pywps.Parser.Post import Post as PostParser
from pywps.Parser.Get import Get as GetParser
-import string,re
+import string,re,urllib
class Post(PostParser):
""" HTTP POST XML request encoding parser. """
@@ -552,35 +552,40 @@
for dataInput in dataInputs.split(";"):
try:
# key is separated by "=" from value
- key,value = string.split(dataInput,"=",maxsplit=1)
+ key,valueAndAttrs = string.split(dataInput,"=",maxsplit=1)
except ValueError,e:
key = dataInput
- value = ""
+ valueAndAttrs = ""
- if not key and not value:
+ if not key and not valueAndAttrs:
continue
-
+
# initial value
- parsedDataInputs.append({"identifier":key, "value":None})
+ parsed={"identifier":key, "value":None}
# additional input attributes are separated by "@"
attributes = []
- if value.find("@") > 0:
- parsedDataInputs[-1]["value"]=value.split("@")[0]
- attributes=value.split("@")[1:]
- elif value.find("@") == 0:
- parsedDataInputs[-1]["value"]=None
- attributes=value.split("@")[1:]
+ if valueAndAttrs.find("@") > 0:
+
+ encodedValue=valueAndAttrs.split("@")[0]
+ parsed["value"]=urllib.unquote(encodedValue)
+ attributes=valueAndAttrs.split("@")[1:]
+
+ elif valueAndAttrs.find("@") == 0:
+ parsed["value"]=None
+ attributes=valueAndAttrs.split("@")[1:]
else:
#needs to be checked for trueOrFalse
- parsedDataInputs[-1]["value"]=self._trueOrFalse(value)
+ encodedValue=valueAndAttrs
+ parsed["value"]=self._trueOrFalse(urllib.unquote(valueAndAttrs))
attributes = []
-
+
# additional attribute key is separated by "=" from it's value
for attribute in attributes:
attributeKey, attributeValue = attribute.split("=")
- parsedDataInputs[-1][attributeKey.lower()] =self._trueOrFalse(attributeValue)
+
+ parsed[attributeKey.lower()]=self._trueOrFalse(urllib.unquote(attributeValue))
-
+ parsedDataInputs.append(parsed)
return parsedDataInputs
#Moved to Parser class
Modified: branches/pywps-3.2-soap/pywps/Parser/Get.py
===================================================================
--- branches/pywps-3.2-soap/pywps/Parser/Get.py 2011-08-01 10:16:57 UTC (rev 1182)
+++ branches/pywps-3.2-soap/pywps/Parser/Get.py 2011-08-08 10:05:49 UTC (rev 1183)
@@ -41,8 +41,8 @@
import pywps.config
from pywps.Parser import Parser
from pywps.Process.Lang import Lang
-import urllib
+
class Get(Parser):
""" Main Class for parsing HTTP GET request types """
unparsedInputs = None # temporary store for later validation
@@ -83,7 +83,6 @@
else:
try:
key,value = split(feature,"=",maxsplit=1)
- value = urllib.unquote(value)
except:
raise NoApplicableCode(\
'Invalid Key-Value-Pair: "' + \
Modified: branches/pywps-3.2-soap/pywps/Parser/Post.py
===================================================================
--- branches/pywps-3.2-soap/pywps/Parser/Post.py 2011-08-01 10:16:57 UTC (rev 1182)
+++ branches/pywps-3.2-soap/pywps/Parser/Post.py 2011-08-08 10:05:49 UTC (rev 1183)
@@ -101,14 +101,12 @@
raise pywps.FileSizeExceeded()
else:
inputXml = file.read()
-
# make DOM from XML
try:
self.document = parseString(inputXml)
except xml.parsers.expat.ExpatError,e:
raise pywps.NoApplicableCode(e.message)
-
# get first child
firstChild = self.isSoapFirstChild(self.document)
Deleted: branches/pywps-3.2-soap/tests/parser.py
===================================================================
--- branches/pywps-3.2-soap/tests/parser.py 2011-08-01 10:16:57 UTC (rev 1182)
+++ branches/pywps-3.2-soap/tests/parser.py 2011-08-08 10:05:49 UTC (rev 1183)
@@ -1,213 +0,0 @@
-import os
-import sys
-
-pywpsPath = os.path.abspath(os.path.join(os.path.split(os.path.abspath(__file__))[0],".."))
-sys.path.append(pywpsPath)
-
-import pywps
-import pywps.Process
-import unittest
-import os
-from xml.dom import minidom
-import urllib
-import base64
-import tempfile
-from osgeo import ogr
-
-class RequestParseTestCase(unittest.TestCase):
- """Test case for input parsing"""
- wfsurl = "http://www2.dmsolutions.ca/cgi-bin/mswfs_gmap?version=1.0.0&request=getfeature&service=wfs&typename=park"
- wcsurl = "http://www.bnhelp.cz/cgi-bin/crtopo?service=WMS&request=GetMap&LAYERS=sitwgs&TRANSPARENT=true&FORMAT=image%2Ftiff&EXCEPTIONS=application%2Fvnd.ogc.se_xml&VERSION=1.1.1&STYLES=default&SRS=EPSG%3A4326&BBOX=-10,-10,10,10&WIDTH=50&HEIGHT=50"
- wpsns = "http://www.opengis.net/wps/1.0.0"
- getpywps = None
- postpywps = None
-
- def setUp(self):
- pass
-
- def testParseGetCapabilities(self):
- """Test if GetCapabilities request is parsed and if POST and GET methods do get the same result"""
-
- getpywps = pywps.Pywps(pywps.METHOD_GET)
- postpywps = pywps.Pywps(pywps.METHOD_POST)
- getinputs = getpywps.parseRequest("service=wps&request=getcapabilities")
- getCapabilitiesRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_getcapabilities_request.xml"))
- postinputs = postpywps.parseRequest(getCapabilitiesRequestFile)
-
- self.assertEquals(getpywps.inputs["version"], "1.0.0")
- self.assertEquals(getpywps.inputs["request"], "getcapabilities")
- self.assertEquals(getpywps.inputs["service"], "wps")
-
- self.assertEquals(postpywps.inputs["version"], "1.0.0")
- self.assertEquals(postpywps.inputs["request"], "getcapabilities")
- self.assertEquals(postpywps.inputs["service"], "wps")
-
- self.assertEquals(getinputs, postinputs)
-
- def testParseDescribeProcess(self):
- """Test if DescribeProcess request is parsed and if POST and GET
- methods are producing the same result"""
-
- getpywps = pywps.Pywps(pywps.METHOD_GET)
- postpywps = pywps.Pywps(pywps.METHOD_POST)
- getinputs = getpywps.parseRequest("service=wps&request=describeprocess&version=1.0.0&identifier=dummyprocess")
- describeProcessFile = open(os.path.join(pywpsPath,"tests","requests","wps_describeprocess_request_dummyprocess.xml"))
- postinputs = postpywps.parseRequest(describeProcessFile)
-
- self.assertEquals(getpywps.inputs["request"], "describeprocess")
- self.assertTrue("dummyprocess" in getpywps.inputs["identifier"])
- self.assertFalse("returner" in getpywps.inputs["identifier"])
-
- self.assertEquals(postpywps.inputs["request"], "describeprocess")
- self.assertTrue("dummyprocess" in postpywps.inputs["identifier"])
- self.assertFalse("returner" in postpywps.inputs["identifier"])
-
- self.assertEquals(getinputs, postinputs)
-
- ######################################################################################
- def testParseExecuteNoInput(self):
- """Test if Execute request is parsed, no input given"""
-
- getpywps = pywps.Pywps(pywps.METHOD_GET)
- postpywps = pywps.Pywps(pywps.METHOD_POST)
- getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=noinputprocess")
- executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-noinputs.xml"))
- postinputs = postpywps.parseRequest(executeRequestFile)
-
- self.assertEquals(getinputs["request"], "execute")
- self.assertTrue("noinputprocess" in getinputs["identifier"],"noinputsprocess not found in %s"%getinputs)
-
- self.assertEquals(postinputs["request"], "execute")
- self.assertTrue("noinputprocess" in postinputs["identifier"],"noinputsprocess not found in %s"%postinputs)
-
- self.assertEquals(getinputs, postinputs,"Get and Post inputs are not same:\n%s\n%s" % (getinputs,postinputs))
-
- def testParseExecuteLiteralInput(self):
- """Test if Execute request is parsed, literal data inputs"""
-
- getpywps = pywps.Pywps(pywps.METHOD_GET)
- postpywps = pywps.Pywps(pywps.METHOD_POST)
- executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-literalinput.xml"))
- getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=literalprocess&datainputs=[int=1;string=spam;float=1.1]")
- postinputs = postpywps.parseRequest(executeRequestFile)
-
- self.assertEquals(getinputs["request"], "execute")
- self.assertTrue("literalprocess" in getinputs["identifier"])
-
- self.assertEquals(postinputs["request"], "execute")
- self.assertTrue("literalprocess" in postinputs["identifier"])
-
- #self.assertEquals(getinputs, postinputs)
- self.assertEquals(getinputs["datainputs"][0]["value"],postinputs["datainputs"][0]["value"])
- self.assertEquals(getinputs["datainputs"][1]["value"],postinputs["datainputs"][1]["value"])
- self.assertEquals(getinputs["datainputs"][2]["value"],postinputs["datainputs"][2]["value"])
- self.assertTrue(getinputs["datainputs"][0]["value"],1)
- self.assertTrue(getinputs["datainputs"][1]["value"],"spam")
- self.assertTrue(getinputs["datainputs"][2]["value"],"1.1")
-
- def testParseExecuteComplexInputAsReference(self):
- """Test if Execute request is parsed, complex data inputs, given as reference"""
-
- getpywps = pywps.Pywps(pywps.METHOD_GET)
- postpywps = pywps.Pywps(pywps.METHOD_POST)
- executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-complexinput-as-reference.xml"))
- getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=complexprocess&datainputs=[rasterin=%s;vectorin=%s]" %\
- (urllib.quote(self.wfsurl), urllib.quote(self.wcsurl)))
- #print "service=wps&version=1.0.0&request=execute&identifier=complexprocess&datainputs=[rasterin=%s;vectorin=%s]" % (urllib.quote(self.wfsurl), urllib.quote(self.wcsurl))
- postinputs = postpywps.parseRequest(executeRequestFile)
-
- self.assertEquals(getinputs["request"], "execute")
- self.assertEquals(postinputs["request"], "execute")
- self.assertTrue("complexprocess" in getinputs["identifier"])
- self.assertTrue("complexprocess" in postinputs["identifier"])
-
- #self.assertEquals(getinputs, postinputs)
- self.assertEquals(getinputs["datainputs"][0]["value"],postinputs["datainputs"][0]["value"])
- self.assertEquals(getinputs["datainputs"][1]["value"],postinputs["datainputs"][1]["value"])
-
- def testParseBBoxInput(self):
- """Parsing Bounding Box Input"""
- getpywps = pywps.Pywps(pywps.METHOD_GET)
- postpywps = pywps.Pywps(pywps.METHOD_POST)
- executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-bbox.xml"))
- getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=bboxprocess&datainputs=[bboxin=%s]" %\
- ("-11,-12,13,14"))
- postinputs = postpywps.parseRequest(executeRequestFile)
-
- self.assertTrue("bboxprocess" in getinputs["identifier"])
- self.assertTrue("bboxprocess" in postinputs["identifier"])
-
- self.assertEquals(getinputs["datainputs"][0]["value"],"-11,-12,13,14")
- self.assertEquals(postinputs["datainputs"][0]["value"],[-11,-12,13,14])
-
- def testParseRawDataOutput(self):
- """Test, if PyWPS parsers RawData output request correctly"""
- postpywps = pywps.Pywps(pywps.METHOD_POST)
- getpywps = pywps.Pywps(pywps.METHOD_GET)
- executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-complexinput-direct-rawdata-output.xml"))
- postinputs = postpywps.parseRequest(executeRequestFile)
- getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=literalprocess&datainputs=[int=1;string=spam;float=1.1]&rawdataoutput=string")
-
- self.assertFalse(postinputs["responseform"]["responsedocument"])
- self.assertTrue(postinputs["responseform"]["rawdataoutput"]["rasterout"])
-
- self.assertFalse(getinputs["responseform"]["responsedocument"])
- self.assertTrue(getinputs["responseform"]["rawdataoutput"]["string"])
-
- def testParseExecuteComplexInputDirectly(self):
- """Test if Execute request is parsed, complex data inputs, given as """
-
- postpywps = pywps.Pywps(pywps.METHOD_POST)
- executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-complexinput-direct.xml"))
- postinputs = postpywps.parseRequest(executeRequestFile)
-
- self.assertEquals(postinputs["request"], "execute")
- self.assertTrue("complexprocess" in postinputs["identifier"])
- rasterOrig = open(os.path.join(pywpsPath,"tests","datainputs","dem.tiff"))
- rasterOrigData = rasterOrig.read()
- rasterWpsData = base64.decodestring(postinputs["datainputs"][0]["value"])
-
- self.assertEquals(rasterOrigData, rasterWpsData)
-
-
- gmlDriver = ogr.GetDriverByName("GML")
- origDs = gmlDriver.Open(os.path.join(pywpsPath,"tests","datainputs","lakes.gml"))
-
- wpsFile = tempfile.mktemp(prefix="pywps-test")
- wpsFile = open(wpsFile,"w")
- wpsFile.write(postinputs["datainputs"][1]["value"])
- wpsFile.close()
- wpsDs = gmlDriver.Open(wpsFile.name)
-
- wpslayer = wpsDs.GetLayerByIndex(0)
- origlayer = origDs.GetLayerByIndex(0)
-
- self.assertTrue(wpslayer.GetFeatureCount(), origlayer.GetFeatureCount())
-
- # enough here
- # for f in range(wpslayer.GetFeatureCount()):
- # origFeature = origlayer.GetFeature(f)
- # wpsFeature = wpslayer.GetFeature(f)
- # self.assertTrue(origFeature.Equal(wpsFeature))
-
- def testParseExecuteComplexAsReferenceOut(self):
- """Test if Execute request is parsed, we want data outputs as reference"""
-
- postpywps = pywps.Pywps(pywps.METHOD_POST)
- getpywps = pywps.Pywps(pywps.METHOD_GET)
- executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-complexinput-output-as-reference.xml"))
- postinputs = postpywps.parseRequest(executeRequestFile)
- getinputs = getpywps.parseRequest("service=wps&request=execute&version=1.0.0&identifier=complexprocess&datainputs=[rasterin=http://foo/bar/raster.tif;vectorin=http://foo/bar/vector.gml]&responsedocument=[rasterout=@asreference=true;vectorout=@asreference=true]")
-
- self.assertTrue(postinputs["responseform"]["responsedocument"]["outputs"][0]["asreference"] == \
- postinputs["responseform"]["responsedocument"]["outputs"][0]["asreference"] == \
- True)
-
- self.assertTrue(postinputs["responseform"]["responsedocument"]["outputs"][1]["asreference"] == \
- postinputs["responseform"]["responsedocument"]["outputs"][1]["asreference"] == \
- True)
-
-if __name__ == "__main__":
- # unittest.main()
- suite = unittest.TestLoader().loadTestsFromTestCase(RequestParseTestCase)
- unittest.TextTestRunner(verbosity=2).run(suite)
Added: branches/pywps-3.2-soap/tests/parser.py
===================================================================
--- branches/pywps-3.2-soap/tests/parser.py 2011-08-01 10:16:57 UTC (rev 1182)
+++ branches/pywps-3.2-soap/tests/parser.py 2011-08-08 10:05:49 UTC (rev 1183)
@@ -0,0 +1,221 @@
+import os
+import sys
+
+pywpsPath = os.path.abspath(os.path.join(os.path.split(os.path.abspath(__file__))[0],".."))
+sys.path.insert(0,pywpsPath)
+sys.path.append(pywpsPath)
+
+import pywps
+import pywps.Process
+import unittest
+import os
+from xml.dom import minidom
+import urllib
+import base64
+import tempfile
+from osgeo import ogr
+
+
+class RequestParseTestCase(unittest.TestCase):
+ """Test case for input parsing"""
+ wfsurl = "http://www2.dmsolutions.ca/cgi-bin/mswfs_gmap?version=1.0.0&request=getfeature&service=wfs&typename=park"
+ wcsurl = "http://www.bnhelp.cz/cgi-bin/crtopo?service=WMS&request=GetMap&LAYERS=sitwgs&TRANSPARENT=true&FORMAT=image%2Ftiff&EXCEPTIONS=application%2Fvnd.ogc.se_xml&VERSION=1.1.1&STYLES=default&SRS=EPSG%3A4326&BBOX=-10,-10,10,10&WIDTH=50&HEIGHT=50"
+ wpsns = "http://www.opengis.net/wps/1.0.0"
+ getpywps = None
+ postpywps = None
+
+ def testParseGetCapabilities(self):
+ """Test if GetCapabilities request is parsed and if POST and GET methods do get the same result"""
+
+ getpywps = pywps.Pywps(pywps.METHOD_GET)
+ postpywps = pywps.Pywps(pywps.METHOD_POST)
+ getinputs = getpywps.parseRequest("service=wps&request=getcapabilities")
+ getCapabilitiesRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_getcapabilities_request.xml"))
+ postinputs = postpywps.parseRequest(getCapabilitiesRequestFile)
+
+ self.assertEquals(getpywps.inputs["version"], "1.0.0")
+ self.assertEquals(getpywps.inputs["request"], "getcapabilities")
+ self.assertEquals(getpywps.inputs["service"], "wps")
+
+ self.assertEquals(postpywps.inputs["version"], "1.0.0")
+ self.assertEquals(postpywps.inputs["request"], "getcapabilities")
+ self.assertEquals(postpywps.inputs["service"], "wps")
+
+ self.assertEquals(getinputs, postinputs)
+
+ def testParseDescribeProcess(self):
+ """Test if DescribeProcess request is parsed and if POST and GET
+ methods are producing the same result"""
+
+ getpywps = pywps.Pywps(pywps.METHOD_GET)
+ postpywps = pywps.Pywps(pywps.METHOD_POST)
+ getinputs = getpywps.parseRequest("service=wps&request=describeprocess&version=1.0.0&identifier=dummyprocess")
+ describeProcessFile = open(os.path.join(pywpsPath,"tests","requests","wps_describeprocess_request_dummyprocess.xml"))
+ postinputs = postpywps.parseRequest(describeProcessFile)
+
+ self.assertEquals(getpywps.inputs["request"], "describeprocess")
+ self.assertTrue("dummyprocess" in getpywps.inputs["identifier"])
+ self.assertFalse("returner" in getpywps.inputs["identifier"])
+
+ self.assertEquals(postpywps.inputs["request"], "describeprocess")
+ self.assertTrue("dummyprocess" in postpywps.inputs["identifier"])
+ self.assertFalse("returner" in postpywps.inputs["identifier"])
+
+ self.assertEquals(getinputs, postinputs)
+
+ ######################################################################################
+ def testParseExecuteNoInput(self):
+ """Test if Execute request is parsed, no input given"""
+
+ getpywps = pywps.Pywps(pywps.METHOD_GET)
+ postpywps = pywps.Pywps(pywps.METHOD_POST)
+ getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=noinputprocess")
+ executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-noinputs.xml"))
+ postinputs = postpywps.parseRequest(executeRequestFile)
+
+ self.assertEquals(getinputs["request"], "execute")
+ self.assertTrue("noinputprocess" in getinputs["identifier"],"noinputsprocess not found in %s"%getinputs)
+
+ self.assertEquals(postinputs["request"], "execute")
+ self.assertTrue("noinputprocess" in postinputs["identifier"],"noinputsprocess not found in %s"%postinputs)
+
+ self.assertEquals(getinputs, postinputs,"Get and Post inputs are not same:\n%s\n%s" % (getinputs,postinputs))
+
+ def testParseExecuteLiteralInput(self):
+ """Test if Execute request is parsed, literal data inputs, including '@' in GET """
+
+ #NOTE: Unittest changed after SVN: 1146 to check for the parsing of "@"
+
+ getpywps = pywps.Pywps(pywps.METHOD_GET)
+ postpywps = pywps.Pywps(pywps.METHOD_POST)
+ executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-literalinput.xml"))
+ getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=literalprocess&datainputs=[int=1;string=spam%40foo.com at mimetype=text/plain at xlink:href=http%3A//www.w3.org/TR/xmlschema-2/%23string;float=1.1]")
+ postinputs = postpywps.parseRequest(executeRequestFile)
+
+ self.assertEquals(getinputs["request"], "execute")
+ self.assertTrue("literalprocess" in getinputs["identifier"])
+
+ self.assertEquals(postinputs["request"], "execute")
+ self.assertTrue("literalprocess" in postinputs["identifier"])
+
+ #self.assertEquals(getinputs, postinputs)
+ self.assertEquals(getinputs["datainputs"][0]["value"],postinputs["datainputs"][0]["value"])
+ self.assertEquals(getinputs["datainputs"][1]["value"],postinputs["datainputs"][1]["value"])
+ self.assertEquals(getinputs["datainputs"][2]["value"],postinputs["datainputs"][2]["value"])
+ self.assertTrue(getinputs["datainputs"][0]["value"],1)
+ self.assertTrue(getinputs["datainputs"][1]["value"],"spam%40foo.com")
+ self.assertTrue(getinputs["datainputs"][2]["value"],"1.1")
+
+
+ def testParseExecuteComplexInputAsReference(self):
+ """Test if Execute request is parsed, complex data inputs, given as reference"""
+
+ getpywps = pywps.Pywps(pywps.METHOD_GET)
+ postpywps = pywps.Pywps(pywps.METHOD_POST)
+ executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-complexinput-as-reference.xml"))
+ getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=complexprocess&datainputs=[rasterin=%s;vectorin=%s]" %\
+ (urllib.quote(self.wfsurl), urllib.quote(self.wcsurl)))
+ #print "service=wps&version=1.0.0&request=execute&identifier=complexprocess&datainputs=[rasterin=%s;vectorin=%s]" % (urllib.quote(self.wfsurl), urllib.quote(self.wcsurl))
+ postinputs = postpywps.parseRequest(executeRequestFile)
+
+ self.assertEquals(getinputs["request"], "execute")
+ self.assertEquals(postinputs["request"], "execute")
+ self.assertTrue("complexprocess" in getinputs["identifier"])
+ self.assertTrue("complexprocess" in postinputs["identifier"])
+
+ #self.assertEquals(getinputs, postinputs)
+ self.assertEquals(getinputs["datainputs"][0]["value"],postinputs["datainputs"][0]["value"])
+ self.assertEquals(getinputs["datainputs"][1]["value"],postinputs["datainputs"][1]["value"])
+
+ def testParseBBoxInput(self):
+ """Parsing Bounding Box Input"""
+ getpywps = pywps.Pywps(pywps.METHOD_GET)
+ postpywps = pywps.Pywps(pywps.METHOD_POST)
+ executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-bbox.xml"))
+ getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=bboxprocess&datainputs=[bboxin=%s]" %\
+ ("-11,-12,13,14"))
+ postinputs = postpywps.parseRequest(executeRequestFile)
+
+ self.assertTrue("bboxprocess" in getinputs["identifier"])
+ self.assertTrue("bboxprocess" in postinputs["identifier"])
+
+ self.assertEquals(getinputs["datainputs"][0]["value"],"-11,-12,13,14")
+ self.assertEquals(postinputs["datainputs"][0]["value"],[-11,-12,13,14])
+
+ def testParseRawDataOutput(self):
+ """Test, if PyWPS parsers RawData output request correctly"""
+ postpywps = pywps.Pywps(pywps.METHOD_POST)
+ getpywps = pywps.Pywps(pywps.METHOD_GET)
+ executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-complexinput-direct-rawdata-output.xml"))
+ postinputs = postpywps.parseRequest(executeRequestFile)
+ getinputs = getpywps.parseRequest("service=wps&version=1.0.0&request=execute&identifier=literalprocess&datainputs=[int=1;string=spam;float=1.1]&rawdataoutput=string")
+
+ self.assertFalse(postinputs["responseform"]["responsedocument"])
+ self.assertTrue(postinputs["responseform"]["rawdataoutput"]["rasterout"])
+
+ self.assertFalse(getinputs["responseform"]["responsedocument"])
+ self.assertTrue(getinputs["responseform"]["rawdataoutput"]["string"])
+
+ def testParseExecuteComplexInputDirectly(self):
+ """Test if Execute request is parsed, complex data inputs, given as """
+
+ postpywps = pywps.Pywps(pywps.METHOD_POST)
+ executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-complexinput-direct.xml"))
+ postinputs = postpywps.parseRequest(executeRequestFile)
+
+ self.assertEquals(postinputs["request"], "execute")
+ self.assertTrue("complexprocess" in postinputs["identifier"])
+ rasterOrig = open(os.path.join(pywpsPath,"tests","datainputs","dem.tiff"))
+ rasterOrigData = rasterOrig.read()
+ rasterWpsData = base64.decodestring(postinputs["datainputs"][0]["value"])
+
+ self.assertEquals(rasterOrigData, rasterWpsData)
+
+
+ gmlDriver = ogr.GetDriverByName("GML")
+ origDs = gmlDriver.Open(os.path.join(pywpsPath,"tests","datainputs","lakes.gml"))
+
+ wpsFile = tempfile.mktemp(prefix="pywps-test")
+ wpsFile = open(wpsFile,"w")
+ wpsFile.write(postinputs["datainputs"][1]["value"])
+ wpsFile.close()
+ wpsDs = gmlDriver.Open(wpsFile.name)
+
+ wpslayer = wpsDs.GetLayerByIndex(0)
+ origlayer = origDs.GetLayerByIndex(0)
+
+ self.assertTrue(wpslayer.GetFeatureCount(), origlayer.GetFeatureCount())
+
+ # enough here
+ # for f in range(wpslayer.GetFeatureCount()):
+ # origFeature = origlayer.GetFeature(f)
+ # wpsFeature = wpslayer.GetFeature(f)
+ # self.assertTrue(origFeature.Equal(wpsFeature))
+
+ def testParseExecuteComplexAsReferenceOut(self):
+ """Test if Execute request is parsed, we want data outputs as reference"""
+ self._setFromEnv()
+ postpywps = pywps.Pywps(pywps.METHOD_POST)
+ getpywps = pywps.Pywps(pywps.METHOD_GET)
+ executeRequestFile = open(os.path.join(pywpsPath,"tests","requests","wps_execute_request-complexinput-output-as-reference.xml"))
+ postinputs = postpywps.parseRequest(executeRequestFile)
+ getinputs = getpywps.parseRequest("service=wps&request=execute&version=1.0.0&identifier=complexprocess&datainputs=[rasterin=http://foo/bar/raster.tif;vectorin=http://foo/bar/vector.gml]&responsedocument=[rasterout=@asreference=true;vectorout=@asreference=true]")
+
+ self.assertTrue(postinputs["responseform"]["responsedocument"]["outputs"][0]["asreference"] == \
+ postinputs["responseform"]["responsedocument"]["outputs"][0]["asreference"] == \
+ True)
+
+ self.assertTrue(postinputs["responseform"]["responsedocument"]["outputs"][1]["asreference"] == \
+ postinputs["responseform"]["responsedocument"]["outputs"][1]["asreference"] == \
+ True)
+
+
+ def _setFromEnv(self):
+ os.putenv("PYWPS_PROCESSES", os.path.join(pywpsPath,"tests","processes"))
+ os.environ["PYWPS_PROCESSES"] = os.path.join(pywpsPath,"tests","processes")
+
+
+
+if __name__ == "__main__":
+ suite = unittest.TestLoader().loadTestsFromTestCase(RequestParseTestCase)
+ unittest.TextTestRunner(verbosity=2).run(suite)
Modified: branches/pywps-3.2-soap/tests/perform_requests.py
===================================================================
--- branches/pywps-3.2-soap/tests/perform_requests.py 2011-08-01 10:16:57 UTC (rev 1182)
+++ branches/pywps-3.2-soap/tests/perform_requests.py 2011-08-08 10:05:49 UTC (rev 1183)
@@ -437,7 +437,8 @@
upperSet=set([len(coord.split(" ")) for coord in [item.getElementsByTagNameNS(self.owsns,"UpperCorner")[0].childNodes[0].nodeValue for item in bboxDom]])
self.assertEquals(len(dimSet.difference(lowerSet)),0) #0
self.assertEquals(len(dimSet.difference(upperSet)),0) #0
-
+
+
def _setFromEnv(self):
os.putenv("PYWPS_PROCESSES", os.path.join(pywpsPath,"tests","processes"))
os.environ["PYWPS_PROCESSES"] = os.path.join(pywpsPath,"tests","processes")
Modified: branches/pywps-3.2-soap/tests/requests/wps_execute_request-literalinput.xml
===================================================================
--- branches/pywps-3.2-soap/tests/requests/wps_execute_request-literalinput.xml 2011-08-01 10:16:57 UTC (rev 1182)
+++ branches/pywps-3.2-soap/tests/requests/wps_execute_request-literalinput.xml 2011-08-08 10:05:49 UTC (rev 1183)
@@ -19,7 +19,7 @@
<wps:Input>
<ows:Identifier>string</ows:Identifier>
<wps:Data>
- <wps:LiteralData>spam</wps:LiteralData>
+ <wps:LiteralData>spam at foo.com</wps:LiteralData>
</wps:Data>
</wps:Input>
<wps:Input>
More information about the Pywps-commits
mailing list