Include iothub metadata in EventHub event trigger metadata

Closes: #288

Author: elprans

azure/functions_worker/bindings/eventhub.py

@@ -84,6 +84,13 @@ def from_proto(cls, data: protos.TypedData, *,
             raise NotImplementedError(
                 f'unsupported event data payload type: {data_type}')
 
+        iothub_metadata = {}
+        for f in trigger_metadata:
+            if f.startswith('iothub-'):
+                v = cls._decode_trigger_metadata_field(
+                    trigger_metadata, f, python_type=str)
+                iothub_metadata[f[len('iothub-'):]] = v
+
         return _eventhub.EventHubEvent(
             body=body,
             enqueued_time=cls._parse_datetime_metadata(
@@ -93,5 +100,6 @@ def from_proto(cls, data: protos.TypedData, *,
             sequence_number=cls._decode_trigger_metadata_field(
                 trigger_metadata, 'SequenceNumber', python_type=int),
             offset=cls._decode_trigger_metadata_field(
-                trigger_metadata, 'Offset', python_type=str)
+                trigger_metadata, 'Offset', python_type=str),
+            iothub_metadata=iothub_metadata
         )

azure/functions_worker/testutils.py

@@ -326,7 +326,14 @@ async def load_function(self, name):
         return func.id, r
 
     async def invoke_function(
-            self, name, input_data: typing.List[protos.ParameterBinding]):
+            self,
+            name,
+            input_data: typing.List[protos.ParameterBinding],
+            metadata: typing.Optional[
+                typing.Mapping[str, protos.TypedData]]=None):
+
+        if metadata is None:
+            metadata = {}
 
         if name not in self._available_functions:
             raise RuntimeError(f'cannot load function {name}')
@@ -339,7 +346,10 @@ async def invoke_function(
                 invocation_request=protos.InvocationRequest(
                     invocation_id=invocation_id,
                     function_id=func.id,
-                    input_data=input_data)),
+                    input_data=input_data,
+                    trigger_metadata=metadata,
+                )
+            ),
             wait_for='invocation_response')
 
         return invocation_id, r

setup.py

@@ -225,7 +225,7 @@ def run(self):
     install_requires=[
         'grpcio~=1.19.0',
         'grpcio-tools~=1.19.0',
-        'azure-functions==1.0.0b3',
+        'azure-functions==1.0.0b4',
     ],
     extras_require={
         'dev': [

tests/eventhub_functions/eventhub_output/function.json

@@ -13,7 +13,7 @@
       "name": "event",
       "direction": "out",
       "eventHubName": "python-worker-ci",
-      "connection": "AzureWebJobsEventHubConnectionString",
+      "connection": "AzureWebJobsEventHubConnectionString"
     },
     {
       "direction": "out",

tests/eventhub_functions/eventhub_trigger/function.json

@@ -8,14 +8,14 @@
       "name": "event",
       "direction": "in",
       "eventHubName": "python-worker-ci",
-      "connection": "AzureWebJobsEventHubConnectionString",
+      "connection": "AzureWebJobsEventHubConnectionString"
     },
     {
       "type": "blob",
       "direction": "out",
       "name": "$return",
       "connection": "AzureWebJobsStorage",
       "path": "python-worker-tests/test-eventhub-triggered.txt"
-    },
+    }
   ]
 }

tests/eventhub_functions/eventhub_trigger_iot/__init__.py

@@ -0,0 +1,7 @@
+import json
+
+import azure.functions as func
+
+
+def main(event: func.EventHubEvent) -> str:
+    return json.dumps(event.iothub_metadata)

tests/eventhub_functions/eventhub_trigger_iot/function.json

@@ -0,0 +1,21 @@
+{
+  "scriptFile": "__init__.py",
+  "disabled": false,
+
+  "bindings": [
+    {
+      "type": "eventHubTrigger",
+      "name": "event",
+      "direction": "in",
+      "eventHubName": "python-worker-ci",
+      "connection": "AzureWebJobsEventHubConnectionString"
+    },
+    {
+      "type": "blob",
+      "direction": "out",
+      "name": "$return",
+      "connection": "AzureWebJobsStorage",
+      "path": "python-worker-tests/test-eventhub-iot-triggered.txt"
+    }
+  ]
+}

tests/eventhub_functions/get_eventhub_triggered/function.json

@@ -17,7 +17,7 @@
     {
       "type": "http",
       "direction": "out",
-      "name": "$return",
+      "name": "$return"
     }
   ]
 }

tests/test_eventhub_functions.py

@@ -1,6 +1,7 @@
 import json
 import time
 
+from azure.functions_worker import protos
 from azure.functions_worker import testutils
 
 
@@ -39,3 +40,48 @@ def test_eventhub_trigger(self):
                     raise
             else:
                 break
+
+
+class TestEventHubMockFunctions(testutils.AsyncTestCase):
+
+    async def test_mock_eventhub_trigger_iot(self):
+        async with testutils.start_mockhost(
+                script_root='eventhub_functions') as host:
+
+            func_id, r = await host.load_function('eventhub_trigger_iot')
+
+            self.assertEqual(r.response.function_id, func_id)
+            self.assertEqual(r.response.result.status,
+                             protos.StatusResult.Success)
+
+            async def call_and_check():
+                _, r = await host.invoke_function(
+                    'eventhub_trigger_iot',
+                    [
+                        protos.ParameterBinding(
+                            name='event',
+                            data=protos.TypedData(
+                                json=json.dumps({
+                                    'id': 'foo'
+                                })
+                            ),
+                        ),
+                    ],
+                    metadata={
+                        'iothub-device-id': protos.TypedData(
+                            string='mock-iothub-device-id'
+                        ),
+                        'iothub-auth-data': protos.TypedData(
+                            string='mock-iothub-auth-data'
+                        )
+                    }
+                )
+
+                self.assertEqual(r.response.result.status,
+                                 protos.StatusResult.Success)
+                self.assertIn(
+                    'mock-iothub-device-id',
+                    r.response.return_value.string
+                )
+
+            await call_and_check()