tests/common.py - use cache with get_schema tests/test_execute.py - save schema_wps_url as const string tests/test_describe.py - reformat tests/test_capabilities.py - remove unnecessary imports

Author: idanmiara

tests/common.py

@@ -7,6 +7,8 @@
 # server_wps_url, server_base_url are used in the tests
 from app_config import server_wps_url, server_base_url
 
+schema_wps_url = 'http://schemas.opengis.net/wps/1.0.0/wpsExecute_response.xsd'
+
 NAMESPACES = {
     'xlink': "http://www.w3.org/1999/xlink",
     'wps': "http://www.opengis.net/wps/1.0.0",
@@ -32,9 +34,14 @@ def get_response(url, post_data=None):
 
 
 def get_schema(url):
-    schema = get_response(url)
-    xmlschema_doc = etree.parse(schema)
-    return etree.XMLSchema(xmlschema_doc)
+    if not hasattr(get_schema, "cache"):
+        get_schema.cache = dict()
+    if url not in get_schema.cache:
+        schema_response = get_response(url)
+        xmlschema_doc = etree.parse(schema_response)
+        # etree.XMLSchema takes ages, so I cache the result
+        get_schema.cache[url] = etree.XMLSchema(xmlschema_doc)
+    return get_schema.cache[url]
 
 
 def validate_file(path, schema):
@@ -57,3 +64,11 @@ def validate(url, schema, post_data=None):
         print(info)
         print(body)
         return False
+
+
+if __name__ == '__main__':
+    schema = get_schema(schema_wps_url)
+    print(schema)
+    schema = get_schema(schema_wps_url)
+    print(schema)
+

tests/test_capabilities.py

@@ -1,20 +1,18 @@
 import unittest
-import lxml.etree as etree
-import urllib
-import subprocess
 
 from tests.common import validate, server_wps_url
 
+
 class CapabilitiesTest(unittest.TestCase):
 
     def setUp(self):
-
         self.url = server_wps_url + "?service=wps&request=getcapabilities"
         self.schema_url = 'http://schemas.opengis.net/wps/1.0.0/wpsGetCapabilities_response.xsd'
 
     def test_valid(self):
         assert validate(self.url, self.schema_url)
 
+
 def load_tests(loader=None, tests=None, pattern=None):
     if not loader:
         loader = unittest.TestLoader()

tests/test_describe.py

@@ -4,16 +4,17 @@
 
 from tests.common import validate, server_wps_url
 
+
 class DescribeTest(unittest.TestCase):
 
     def setUp(self):
-
         self.url = server_wps_url + '?service=wps&request=describeprocess&version=1.0.0&identifier=all'
         self.schema_url = 'http://schemas.opengis.net/wps/1.0.0/wpsDescribeProcess_response.xsd'
 
     def test_valid(self):
         assert validate(self.url, self.schema_url)
 
+
 def load_tests(loader=None, tests=None, pattern=None):
     if not loader:
         loader = unittest.TestLoader()

tests/test_execute.py

@@ -4,9 +4,7 @@
 import lxml.etree as etree
 import time
 
-from tests.common import validate, server_wps_url, get_response
-
-from tests.common import NAMESPACES, server_base_url
+from tests.common import NAMESPACES, validate, server_wps_url, server_base_url, get_response, schema_wps_url
 
 
 process_succeeded = '//wps:ExecuteResponse/wps:Status/wps:ProcessSucceeded'
@@ -22,7 +20,7 @@ class SayHello(unittest.TestCase):
     """
 
     def setUp(self):
-        self.schema_url = 'http://schemas.opengis.net/wps/1.0.0/wpsExecute_response.xsd'
+        self.schema_url = schema_wps_url
 
     def test_valid(self):
         """GET Execute request"""
@@ -37,7 +35,7 @@ class Buffer(unittest.TestCase):
     """
 
     def setUp(self):
-        self.schema_url = 'http://schemas.opengis.net/wps/1.0.0/wpsExecute_response.xsd'
+        self.schema_url = schema_wps_url
         self.url = server_wps_url
         resp = get_response(server_base_url + '/static/requests/buffer.xml')
         self.request_data = resp.read()
@@ -169,3 +167,7 @@ def load_tests(loader=None, tests=None, pattern=None):
         loader.loadTestsFromTestCase(Buffer)
     ]
     return unittest.TestSuite(suite_list)
+
+
+if __name__ == "__main__":
+    load_tests()