Adding E2E tests for EventGrid output binding (#724)

* Adding E2E tests for EventGrid output binding
* Skipping the test due to 401 issues from the host.

Author: vrdmr

.ci/linux_devops_e2e_tests.sh

@@ -5,5 +5,7 @@ export AzureWebJobsStorage=$LINUXSTORAGECONNECTIONSTRING
 export AzureWebJobsCosmosDBConnectionString=$LINUXCOSMOSDBCONNECTIONSTRING
 export AzureWebJobsEventHubConnectionString=$LINUXEVENTHUBCONNECTIONSTRING
 export AzureWebJobsServiceBusConnectionString=$LINUXSERVICEBUSCONNECTIONSTRING
+export AzureWebJobsEventGridTopicUri=$LINUXEVENTGRIDTOPICURI
+export AzureWebJobsEventGridConnectionKey=$LINUXEVENTGRIDTOPICCONNECTIONKEY
 
-pytest  --instafail --cov=./azure_functions_worker --cov-report xml --cov-branch --cov-append tests/endtoend
+pytest  --instafail --cov=./azure_functions_worker --cov-report xml --cov-branch --cov-append tests/endtoend

.gitignore

@@ -119,3 +119,6 @@ py3env/
 # PyCharm
 .idea/
 .idea_modules/
+
+# Profiling info
+prof/

azure-pipelines.yml

@@ -31,18 +31,24 @@ jobs:
         linuxCosmosDB: $(LinuxCosmosDBConnectionString36)
         linuxEventHub: $(LinuxEventHubConnectionString36)
         linuxServiceBus: $(LinuxServiceBusConnectionString36)
+        linuxEventGridTopicUri: $(LinuxEventGridTopicUriString36)
+        linuxEventGridConnectionKey: $(LinuxEventGridConnectionKeyString36)
       Python37:
         pythonVersion: '3.7'
         linuxStorage: $(LinuxStorageConnectionString37)
         linuxCosmosDB: $(LinuxCosmosDBConnectionString37)
         linuxEventHub: $(LinuxEventHubConnectionString37)
         linuxServiceBus: $(LinuxServiceBusConnectionString37)
+        linuxEventGridTopicUri: $(LinuxEventGridTopicUriString37)
+        linuxEventGridConnectionKey: $(LinuxEventGridConnectionKeyString37)
       Python38:
         pythonVersion: '3.8'
         linuxStorage: $(LinuxStorageConnectionString38)
         linuxCosmosDB: $(LinuxCosmosDBConnectionString38)
         linuxEventHub: $(LinuxEventHubConnectionString38)
         linuxServiceBus: $(LinuxServiceBusConnectionString38)
+        linuxEventGridTopicUri: $(LinuxEventGridTopicUriString38)
+        linuxEventGridConnectionKey: $(LinuxEventGridConnectionKeyString38)
   steps:
   - task: UsePythonVersion@0
     inputs:
@@ -70,6 +76,8 @@ jobs:
       LINUXCOSMOSDBCONNECTIONSTRING: $(linuxCosmosDB)
       LINUXEVENTHUBCONNECTIONSTRING: $(linuxEventHub)
       LINUXSERVICEBUSCONNECTIONSTRING: $(linuxServiceBus)
+      LINUXEVENTGRIDTOPICURI: $(linuxEventGridTopicUri)
+      LINUXEVENTGRIDTOPICCONNECTIONKEY: $(linuxEventGridConnectionKey)
     displayName: 'E2E Tests'
   - task: PublishCodeCoverageResults@1
     inputs:

azure_functions_worker/testutils.py

@@ -44,13 +44,14 @@
 E2E_TESTS_ROOT = TESTS_ROOT / E2E_TESTS_FOLDER
 UNIT_TESTS_FOLDER = pathlib.Path('unittests')
 UNIT_TESTS_ROOT = TESTS_ROOT / UNIT_TESTS_FOLDER
-DEFAULT_WEBHOST_DLL_PATH = PROJECT_ROOT / 'build' / 'webhost' / \
-    'Microsoft.Azure.WebJobs.Script.WebHost.dll'
+WEBHOST_DLL = "Microsoft.Azure.WebJobs.Script.WebHost.dll"
+DEFAULT_WEBHOST_DLL_PATH = PROJECT_ROOT / 'build' / 'webhost' / WEBHOST_DLL
 EXTENSIONS_PATH = PROJECT_ROOT / 'build' / 'extensions' / 'bin'
 FUNCS_PATH = TESTS_ROOT / UNIT_TESTS_FOLDER / 'http_functions'
 WORKER_PATH = PROJECT_ROOT / 'python' / 'test'
 WORKER_CONFIG = PROJECT_ROOT / '.testconfig'
 ON_WINDOWS = platform.system() == 'Windows'
+LOCALHOST = "127.0.0.1"
 
 HOST_JSON_TEMPLATE = """\
 {
@@ -317,7 +318,7 @@ def __init__(self, loop, scripts_dir):
         self._server = grpc.server(self._threadpool)
         self._servicer = _MockWebHostServicer(self)
         protos.add_FunctionRpcServicer_to_server(self._servicer, self._server)
-        self._port = self._server.add_insecure_port('127.0.0.1:0')
+        self._port = self._server.add_insecure_port(f'{LOCALHOST}:0')
 
         self._worker_id = self.make_id()
         self._request_id = self.make_id()
@@ -459,7 +460,7 @@ async def __aenter__(self):
         await self._host.start()
 
         self._worker = await dispatcher. \
-            Dispatcher.connect('127.0.0.1', self._host._port,
+            Dispatcher.connect(LOCALHOST, self._host._port,
                                self._host.worker_id,
                                self._host.request_id, connect_timeout=5.0)
 
@@ -469,6 +470,7 @@ async def __aenter__(self):
             wait([self._host._connected_fut, self._worker_task],
                  return_when=asyncio.FIRST_COMPLETED)
 
+        # noinspection PyBroadException
         try:
             if self._worker_task in done:
                 self._worker_task.result()
@@ -502,7 +504,7 @@ async def __aexit__(self, *exc):
 def start_mockhost(*, script_root=FUNCS_PATH):
     tests_dir = TESTS_ROOT
     scripts_dir = tests_dir / script_root
-    if not scripts_dir.exists() or not scripts_dir.is_dir():
+    if not (scripts_dir.exists() and scripts_dir.is_dir()):
         raise RuntimeError(
             f'invalid script_root argument: '
             f'{scripts_dir} directory does not exist')
@@ -536,7 +538,7 @@ def close(self):
 
 def _find_open_port():
     with socket.socket() as s:
-        s.bind(('127.0.0.1', 0))
+        s.bind((LOCALHOST, 0))
         s.listen(1)
         return s.getsockname()[1]
 
@@ -572,10 +574,10 @@ def popen_webhost(*, stdout, stderr, script_root=FUNCS_PATH, port=None):
         if not dll:
             dll = DEFAULT_WEBHOST_DLL_PATH
 
-            secrets = SECRETS_TEMPLATE
-
             os.makedirs(dll.parent / 'Secrets', exist_ok=True)
             with open(dll.parent / 'Secrets' / 'host.json', 'w') as f:
+                secrets = SECRETS_TEMPLATE
+
                 f.write(secrets)
 
         if dll and pathlib.Path(dll).exists():
@@ -605,11 +607,7 @@ def popen_webhost(*, stdout, stderr, script_root=FUNCS_PATH, port=None):
         ]))
 
     worker_path = os.environ.get('PYAZURE_WORKER_DIR')
-    if not worker_path:
-        worker_path = WORKER_PATH
-    else:
-        worker_path = pathlib.Path(worker_path)
-
+    worker_path = WORKER_PATH if not worker_path else pathlib.Path(worker_path)
     if not worker_path.exists():
         raise RuntimeError(f'Worker path {worker_path} does not exist')
 
@@ -640,6 +638,15 @@ def popen_webhost(*, stdout, stderr, script_root=FUNCS_PATH, port=None):
         if servicebus:
             extra_env['AzureWebJobsServiceBusConnectionString'] = servicebus
 
+        eventgrid_topic_uri = testconfig['azure'].get('eventgrid_topic_uri')
+        if eventgrid_topic_uri:
+            extra_env['AzureWebJobsEventGridTopicUri'] = eventgrid_topic_uri
+
+        eventgrid_topic_key = testconfig['azure'].get('eventgrid_topic_key')
+        if eventgrid_topic_key:
+            extra_env['AzureWebJobsEventGridConnectionKey'] = \
+                eventgrid_topic_key
+
     if port is not None:
         extra_env['ASPNETCORE_URLS'] = f'http://*:{port}'
 
@@ -655,11 +662,7 @@ def popen_webhost(*, stdout, stderr, script_root=FUNCS_PATH, port=None):
 
 
 def start_webhost(*, script_dir=None, stdout=None):
-    if script_dir:
-        script_root = TESTS_ROOT / script_dir
-    else:
-        script_root = FUNCS_PATH
-
+    script_root = TESTS_ROOT / script_dir if script_dir else FUNCS_PATH
     if stdout is None:
         if is_envvar_true(PYAZURE_WEBHOST_DEBUG):
             stdout = sys.stdout
@@ -670,8 +673,8 @@ def start_webhost(*, script_dir=None, stdout=None):
     proc = popen_webhost(stdout=stdout, stderr=subprocess.STDOUT,
                          script_root=script_root, port=port)
 
-    addr = f'http://127.0.0.1:{port}'
-    for n in range(10):
+    addr = f'http://{LOCALHOST}:{port}'
+    for _ in range(10):
         try:
             r = requests.get(f'{addr}/api/ping',
                              params={'code': 'testFunctionKey'})
@@ -682,11 +685,11 @@ def start_webhost(*, script_dir=None, stdout=None):
         except requests.exceptions.ConnectionError:
             pass
 
-        time.sleep(0.5)
+        time.sleep(1)
     else:
         proc.terminate()
         try:
-            proc.wait(10)
+            proc.wait(20)
         except subprocess.TimeoutExpired:
             proc.kill()
         raise RuntimeError('could not start the webworker')
@@ -697,7 +700,7 @@ def start_webhost(*, script_dir=None, stdout=None):
 def create_dummy_dispatcher():
     dummy_event_loop = asyncio.new_event_loop()
     disp = dispatcher.Dispatcher(
-        dummy_event_loop, '127.0.0.1', 0,
+        dummy_event_loop, LOCALHOST, 0,
         'test_worker_id', 'test_request_id',
         1.0, 1000)
     dummy_event_loop.close()

setup.py

@@ -195,9 +195,8 @@ def _install_webhost(self):
                 print('Downloading Azure Functions Web Host...')
                 urllib.request.urlretrieve(self.webhost_url, zipf.name)
             except Exception as e:
-                print(
-                    f"could not download Azure Functions Web Host binaries "
-                    f"from {self.webhost_url}: {e!r}", file=sys.stderr)
+                print(f"could not download Azure Functions Web Host binaries "
+                      f"from {self.webhost_url}: {e!r}", file=sys.stderr)
                 sys.exit(1)
 
             if not self.webhost_dir.exists():

tests/endtoend/eventgrid_functions/eventgrid_output_binding/__init__.py

@@ -0,0 +1,27 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from datetime import datetime
+
+import azure.functions as func
+
+
+def main(req: func.HttpRequest,
+         outputEvent: func.Out[func.EventGridOutputEvent]) -> func.HttpResponse:
+    test_uuid = req.params.get('test_uuid')
+    data_to_event_grid = func.EventGridOutputEvent(id="test-id",
+                                                   data={
+                                                       "test_uuid": test_uuid
+                                                   },
+                                                   subject="test-subject",
+                                                   event_type="test-event-1",
+                                                   event_time=datetime.utcnow(),
+                                                   data_version="1.0")
+
+    outputEvent.set(data_to_event_grid)
+    r_value = "Sent event with subject: {}, id: {}, data: {}, event_type: {} " \
+              "to EventGrid!".format(data_to_event_grid.subject,
+                                     data_to_event_grid.id,
+                                     data_to_event_grid.get_json(),
+                                     data_to_event_grid.event_type)
+    return func.HttpResponse(r_value)

tests/endtoend/eventgrid_functions/eventgrid_output_binding/function.json

@@ -0,0 +1,23 @@
+{
+  "scriptFile": "__init__.py",
+
+  "bindings": [
+    {
+      "type": "httpTrigger",
+      "direction": "in",
+      "name": "req"
+    },
+    {
+      "type": "eventGrid",
+      "name": "outputEvent",
+      "topicEndpointUri": "AzureWebJobsEventGridTopicUri",
+      "topicKeySetting": "AzureWebJobsEventGridConnectionKey",
+      "direction": "out"
+    },
+    {
+      "type": "http",
+      "direction": "out",
+      "name": "$return"
+    }
+  ]
+}

tests/endtoend/eventgrid_functions/eventgrid_output_binding_message_to_blobstore/__init__.py

@@ -0,0 +1,8 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import azure.functions as func
+
+
+def main(msg: func.QueueMessage) -> bytes:
+    return msg.get_body()

tests/endtoend/eventgrid_functions/eventgrid_output_binding_message_to_blobstore/function.json

@@ -0,0 +1,19 @@
+{
+  "scriptFile": "__init__.py",
+  "bindings": [
+    {
+      "name": "msg",
+      "type": "queueTrigger",
+      "direction": "in",
+      "queueName": "test-event-grid-storage-queue",
+      "connection": "AzureWebJobsStorage"
+    },
+    {
+      "type": "blob",
+      "direction": "out",
+      "name": "$return",
+      "connection": "AzureWebJobsStorage",
+      "path": "python-worker-tests/test-eventgrid-output-binding.txt"
+    }
+  ]
+}

tests/endtoend/eventgrid_functions/eventgrid_output_binding_success/__init__.py

@@ -0,0 +1,8 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import azure.functions as func
+
+
+def main(req: func.HttpRequest, file: func.InputStream) -> str:
+    return file.read().decode('utf-8')