Added durable functions stein tests (#1199)

* Added durable functions stein tests

* Fixed durable tests

* Updated durable task extension version

* Flake8 fix

* Flake8 fixes

---------

Co-authored-by: Gavin Aguiar <gavin@GavinPC>

Author: gavin-aguiar

setup.py

@@ -56,7 +56,7 @@
         Include="Microsoft.Azure.WebJobs.Script.ExtensionsMetadataGenerator"
         Version="1.1.3" />
       <PackageReference Include="Microsoft.Azure.WebJobs.Extensions.DurableTask"
-        Version="2.7.2" />
+        Version="2.9.4" />
    </ItemGroup>
 </Project>
 """

tests/endtoend/durable_functions/Hello/__init__.py

@@ -6,6 +6,5 @@
 # - add azure-functions-durable to requirements.txt
 # - run pip install -r requirements.txt
 
-def main(name: str, blob) -> str:
-    blob.set("test-durable")
+def main(name: str) -> str:
     return f"Hello {name}!"

tests/endtoend/durable_functions/Hello/function.json

@@ -5,14 +5,6 @@
       "name": "name",
       "type": "activityTrigger",
       "direction": "in"
-    },
-    {
-      "type": "blob",
-      "direction": "out",
-      "name": "blob",
-      "path": "python-worker-tests/test-return1.txt",
-      "connection": "AzureWebJobsStorage"
     }
-
   ]
 }

tests/endtoend/durable_functions/durable_functions_stein/function_app.py

@@ -0,0 +1,28 @@
+import azure.functions as func
+import azure.durable_functions as df
+import logging
+
+app = df.DFApp()
+
+
+@app.orchestration_trigger(context_name="context")
+def durablefunctionsorchestrator(context):
+    result1 = yield context.call_activity('Hello', "Tokyo")
+    result2 = yield context.call_activity('Hello', "Seattle")
+    result3 = yield context.call_activity('Hello', "London")
+    return [result1, result2, result3]
+
+
+@app.route(route="orchestrators/{functionName}",
+           auth_level=func.AuthLevel.ANONYMOUS)
+@app.durable_client_input(client_name="client")
+async def durable_client(req: func.HttpRequest, client) -> func.HttpResponse:
+    instance_id = await client.start_new(req.route_params["functionName"], None,
+                                         None)
+    logging.info(f"Started orchestration with ID = '{instance_id}'.")
+    return client.create_check_status_response(req, instance_id)
+
+
+@app.activity_trigger(input_name="name")
+def hello(name: str) -> str:
+    return f"Hello {name}!"

tests/endtoend/test_durable_functions.py

@@ -1,6 +1,12 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
+import json
+import os
+import time
 from unittest import skipIf
+from unittest.mock import patch
+
+import requests
 
 from azure_functions_worker.utils.common import is_envvar_true
 from tests.utils import testutils
@@ -9,16 +15,24 @@
 
 @skipIf(is_envvar_true(DEDICATED_DOCKER_TEST)
         or is_envvar_true(CONSUMPTION_DOCKER_TEST),
-        "TODO: This will be fixed in "
-        "https://github.com/Azure/azure-functions-python-worker/pull/1199")
+        "Docker tests cannot retrieve port needed for a webhook")
 class TestDurableFunctions(testutils.WebHostTestCase):
 
     @classmethod
     def setUpClass(cls):
-        # webhook for durable tests
         cls.env_variables['WEBSITE_HOSTNAME'] = "http:"
+        os_environ = os.environ.copy()
+        os_environ.update(cls.env_variables)
+
+        cls._patch_environ = patch.dict('os.environ', os_environ)
+        cls._patch_environ.start()
         super().setUpClass()
 
+    @classmethod
+    def tearDownClass(cls):
+        super().tearDownClass()
+        cls._patch_environ.stop()
+
     @classmethod
     def get_libraries_to_install(cls):
         return ['azure-functions-durable']
@@ -34,4 +48,22 @@ def get_script_dir(cls):
     def test_durable(self):
         r = self.webhost.request('GET',
                                  'orchestrators/DurableFunctionsOrchestrator')
+        time.sleep(4)  # wait for the activity to complete
         self.assertEqual(r.status_code, 202)
+        content = json.loads(r.content)
+
+        status = requests.get(content['statusQueryGetUri'])
+        self.assertEqual(status.status_code, 200)
+
+        status_content = json.loads(status.content)
+        self.assertEqual(status_content['runtimeStatus'], 'Completed')
+        self.assertEqual(status_content['output'],
+                         ['Hello Tokyo!', 'Hello Seattle!', 'Hello London!'])
+
+
+class TestDurableFunctionsStein(TestDurableFunctions):
+
+    @classmethod
+    def get_script_dir(cls):
+        return testutils.E2E_TESTS_FOLDER / 'durable_functions' / \
+                                            'durable_functions_stein'

tests/utils/testutils.py

@@ -810,9 +810,6 @@ def popen_webhost(*, stdout, stderr, script_root=FUNCS_PATH, port=None):
     hostexe_args = []
     os.environ['AzureWebJobsFeatureFlags'] = 'EnableWorkerIndexing'
 
-    # webhook for durable tests
-    os.environ['WEBSITE_HOSTNAME'] = f'http://*:{port}'
-
     # If we want to use core-tools
     coretools_exe = os.environ.get('CORE_TOOLS_EXE_PATH')
     if coretools_exe: