Make asynchronous process test run asynchronously

CONTRIBUTORS.md

@@ -10,6 +10,7 @@
 * @jonas-eberle Jonas Eberle
 * @cehbrecht Carsten Ehbrecht
 * @idanmiara Idan Miara
+* @huard David Huard
 
 # Contributor to older versions of PyWPS (< 4.x)
 

pywps/response/__init__.py

@@ -54,9 +54,9 @@ def _update_status(self, status, message, status_percentage):
         Update status report of currently running process instance
 
         :param str message: Message you need to share with the client
-        :param int status_percentage: Percent done (number betwen <0-100>)
+        :param int status_percentage: Percent done (number between <0-100>)
         :param pywps.response.status.WPS_STATUS status: process status - user should usually
-            ommit this parameter
+            omit this parameter
         """
         self.message = message
         self.status = status

tests/__init__.py

@@ -26,7 +26,7 @@
 from tests import test_service
 from tests import test_process
 from tests import test_processing
-from tests import test_assync
+from tests import test_z_async
 from tests import test_grass_location
 from tests import test_storage
 from tests import test_filestorage
@@ -90,7 +90,7 @@ def load_tests(loader=None, tests=None, pattern=None):
         test_service.load_tests(),
         test_process.load_tests(),
         test_processing.load_tests(),
-        test_assync.load_tests(),
+        test_z_async.load_tests(),
         test_grass_location.load_tests(),
         test_storage.load_tests(),
         test_filestorage.load_tests(),

tests/processes/__init__.py

@@ -99,3 +99,38 @@ def bbox(request, response):
         area = request.inputs['area'][0].data
         response.outputs['extent'].data = area
         return response
+
+
+class Sleep(Process):
+    """A long running process, just sleeping."""
+    def __init__(self):
+        inputs = [
+            LiteralInput('seconds', title='Seconds', data_type='float')
+        ]
+        outputs = [
+            LiteralOutput('finished', title='Finished', data_type='boolean')
+        ]
+
+        super(Sleep, self).__init__(
+            self._handler,
+            identifier='sleep',
+            title='Sleep',
+            abstract='Wait for specified number of seconds.',
+            inputs=inputs,
+            outputs=outputs,
+            store_supported=True,
+            status_supported=True
+        )
+
+    @staticmethod
+    def _handler(request, response):
+        import time
+
+        seconds = request.inputs['seconds'][0].data
+        step = seconds / 3
+        for i in range(3):
+            response.update_status('Sleep in progress...', i / 3 * 100)
+            time.sleep(step)
+
+        response.outputs['finished'].data = "True"
+        return response

tests/test_dblog.py

@@ -16,9 +16,8 @@
 class DBLogTest(unittest.TestCase):
     """DBGLog test cases"""
 
-    def setUp(self):
-
-        self.database = configuration.get_config_value('logging', 'database')
+    def tearDown(self) -> None:
+        configuration.load_configuration()
 
     def test_0_dblog(self):
         """Test pywps.formats.Format class
@@ -30,18 +29,18 @@ def test_db_content(self):
         session = get_session()
         null_time_end = session.query(ProcessInstance).filter(ProcessInstance.time_end == None)
         self.assertEqual(null_time_end.count(), 0,
-                         'There are no unfinished processes loged')
+                         'There are no unfinished processes logged')
 
         null_status = session.query(ProcessInstance).filter(ProcessInstance.status == None)
         self.assertEqual(null_status.count(), 0,
-                         'There are no processes without status loged')
+                         'There are no processes without status logged')
 
         null_percent = session.query(ProcessInstance).filter(ProcessInstance.percent_done == None)
         self.assertEqual(null_percent.count(), 0,
-                         'There are no processes without percent loged')
+                         'There are no processes without percent logged')
 
-        null_percent = session.query(ProcessInstance).filter(ProcessInstance.percent_done < 100)
-        self.assertEqual(null_percent.count(), 0,
+        unfinished = session.query(ProcessInstance).filter(ProcessInstance.percent_done < 100)
+        self.assertEqual(unfinished.count(), 0,
                          'There are no unfinished processes')
 
 def load_tests(loader=None, tests=None, pattern=None):

tests/test_storage.py

@@ -34,6 +34,9 @@ def _get_file(self):
 
 class TestStorageBuilder():
 
+    def teardown_class(cls):
+        configuration.load_configuration()
+
     def test_default_storage(self):
         storage = StorageBuilder.buildStorage()
         assert isinstance(storage, FileStorage)

tests/test_z_async.py

@@ -0,0 +1,59 @@
+##################################################################
+# Copyright 2018 Open Source Geospatial Foundation and others    #
+# licensed under MIT, Please consult LICENSE.txt for details     #
+##################################################################
+
+import unittest
+import time
+from pywps import Service, configuration
+from pywps import get_ElementMakerForVersion
+from pywps.tests import client_for, assert_response_accepted
+from .processes import Sleep
+from owslib.wps import WPSExecution
+from pathlib import Path
+
+VERSION = "1.0.0"
+
+WPS, OWS = get_ElementMakerForVersion(VERSION)
+
+
+class ExecuteTest(unittest.TestCase):
+    def setUp(self) -> None:
+        configuration.CONFIG.set('processing', 'mode', 'distributed')
+
+    def tearDown(self) -> None:
+        configuration.load_configuration()
+
+    def test_async(self):
+        client = client_for(Service(processes=[Sleep()]))
+        wps = WPSExecution()
+
+        # Build an asynchronous request (requires specifying outputs and setting the mode).
+        doc = wps.buildRequest('sleep',
+                               inputs=[('seconds', '.01')],
+                               output=[('finished', None, None)],
+                               mode='async')
+
+        resp = client.post_xml(doc=doc)
+        wps.parseResponse(resp.xml)
+        assert_response_accepted(resp)
+
+        # Wait for process to complete. The test will fail otherwise, which confirms the process is asynchronous.
+        time.sleep(.5)
+
+        # Parse response to extract the status file path
+        url = resp.xml.xpath("//@statusLocation")[0]
+
+        # OWSlib only reads from URLs, not local files. So we need to read the response manually.
+        p = Path(url[6:])
+        wps.checkStatus(response=p.read_bytes(), sleepSecs=0)
+        assert wps.status == 'ProcessSucceeded'
+
+
+def load_tests(loader=None, tests=None, pattern=None):
+    if not loader:
+        loader = unittest.TestLoader()
+    suite_list = [
+        loader.loadTestsFromTestCase(ExecuteTest),
+    ]
+    return unittest.TestSuite(suite_list)