test_eventhub.py

# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.

import sys
from typing import List, Mapping
import unittest
import json
from unittest.mock import patch
from datetime import datetime

import azure.functions as func
import azure.functions.eventhub as azf_eh
import azure.functions.meta as meta

from tests.utils.testutils import CollectionBytes, CollectionString


class TestEventHub(unittest.TestCase):
    MOCKED_ENQUEUE_TIME = datetime.utcnow()

    def test_eventhub_input_type(self):
        check_input_type = (
            azf_eh.EventHubConverter.check_input_type_annotation
        )
        if sys.version_info >= (3, 9):
            self.assertTrue(check_input_type(list[func.EventHubEvent]))

        self.assertTrue(check_input_type(func.EventHubEvent))
        self.assertTrue(check_input_type(List[func.EventHubEvent]))
        self.assertFalse(check_input_type(str))
        self.assertFalse(check_input_type(bytes))
        self.assertFalse(check_input_type(List[str]))

    def test_eventhub_output_type(self):
        check_output_type = (
            azf_eh.EventHubTriggerConverter.check_output_type_annotation
        )

        if sys.version_info >= (3, 9):
            self.assertTrue(check_output_type(list[str]))

        self.assertTrue(check_output_type(bytes))
        self.assertTrue(check_output_type(str))
        self.assertTrue(check_output_type(List[str]))
        self.assertFalse(check_output_type(func.EventHubEvent))
        self.assertFalse(check_output_type(List[bytes]))
        self.assertFalse(check_output_type(List[func.EventHubEvent]))

    @patch('azure.functions.eventhub.EventHubTriggerConverter'
           '.decode_single_event')
    @patch('azure.functions.eventhub.EventHubTriggerConverter'
           '.decode_multiple_events')
    def test_eventhub_decode_single_event(self, dme_mock, dse_mock):
        azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_single_iothub_datum(),
            trigger_metadata=self._generate_single_trigger_metadatum()
        )
        dse_mock.assert_called_once()
        dme_mock.assert_not_called()

    @patch('azure.functions.eventhub.EventHubTriggerConverter'
           '.decode_single_event')
    @patch('azure.functions.eventhub.EventHubTriggerConverter'
           '.decode_multiple_events')
    def test_eventhub_decode_multiple_events(self, dme_mock, dse_mock):
        azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_multiple_iothub_data(),
            trigger_metadata=self._generate_multiple_trigger_metadata()
        )
        dse_mock.assert_not_called()
        dme_mock.assert_called_once()

    def test_eventhub_trigger_single_event_json(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_single_iothub_datum('json'),
            trigger_metadata=self._generate_single_trigger_metadatum()
        )
        # Result body always has the datatype of bytes
        self.assertEqual(
            result.get_body().decode('utf-8'), '{"device-status": "good"}'
        )
        self.assertEqual(result.enqueued_time, self.MOCKED_ENQUEUE_TIME)

    def test_eventhub_trigger_single_event_bytes(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_single_iothub_datum('bytes'),
            trigger_metadata=self._generate_single_trigger_metadatum()
        )
        self.assertEqual(
            result.get_body().decode('utf-8'), '{"device-status": "good"}'
        )
        self.assertEqual(result.enqueued_time, self.MOCKED_ENQUEUE_TIME)

    def test_iothub_metadata_single_event(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_single_iothub_datum('json'),
            trigger_metadata=self._generate_single_trigger_metadatum()
        )
        self.assertIsNotNone(result.iothub_metadata)
        self.assertEqual(
            result.iothub_metadata['connection-device-id'], 'MyTestDevice'
        )

    def test_eventhub_trigger_single_event_string(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_single_iothub_datum('string'),
            trigger_metadata=self._generate_single_trigger_metadatum()
        )
        self.assertEqual(
            result.get_body().decode('utf-8'), '{"device-status": "good"}'
        )
        self.assertEqual(result.enqueued_time, self.MOCKED_ENQUEUE_TIME)

    def test_eventhub_trigger_multiple_events_json(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_multiple_iothub_data('json'),
            trigger_metadata=self._generate_multiple_trigger_metadata()
        )
        self.assertIsInstance(result, list)
        self.assertEqual(len(result), 2)

        self.assertEqual(result[0].enqueued_time, self.MOCKED_ENQUEUE_TIME)
        self.assertEqual(
            result[0].get_body().decode('utf-8'), '{"device-status": "good1"}'
        )

        self.assertEqual(result[1].enqueued_time, self.MOCKED_ENQUEUE_TIME)
        self.assertEqual(
            result[1].get_body().decode('utf-8'), '{"device-status": "good2"}'
        )

    def test_eventhub_trigger_multiple_events_collection_string(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_multiple_iothub_data('collection_string'),
            trigger_metadata=self._generate_multiple_trigger_metadata()
        )
        self.assertIsInstance(result, list)
        self.assertEqual(len(result), 2)

        self.assertEqual(result[0].enqueued_time, self.MOCKED_ENQUEUE_TIME)
        self.assertEqual(
            result[0].get_body().decode('utf-8'), '{"device-status": "good1"}'
        )

        self.assertEqual(result[1].enqueued_time, self.MOCKED_ENQUEUE_TIME)
        self.assertEqual(
            result[1].get_body().decode('utf-8'), '{"device-status": "good2"}'
        )

    def test_eventhub_trigger_multiple_events_collection_bytes(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_multiple_iothub_data('collection_bytes'),
            trigger_metadata=self._generate_multiple_trigger_metadata()
        )
        self.assertIsInstance(result, list)
        self.assertEqual(len(result), 2)

        self.assertEqual(result[0].enqueued_time, self.MOCKED_ENQUEUE_TIME)
        self.assertEqual(
            result[0].get_body().decode('utf-8'), '{"device-status": "good1"}'
        )

        self.assertEqual(result[1].enqueued_time, self.MOCKED_ENQUEUE_TIME)
        self.assertEqual(
            result[1].get_body().decode('utf-8'), '{"device-status": "good2"}'
        )

    def test_iothub_metadata_events(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_multiple_iothub_data('json'),
            trigger_metadata=self._generate_multiple_trigger_metadata()
        )
        self.assertIsNotNone(result[0].iothub_metadata)
        self.assertEqual(
            result[0].iothub_metadata['connection-device-id'], 'MyTestDevice1'
        )

        self.assertIsNotNone(result[1].iothub_metadata)
        self.assertEqual(
            result[1].iothub_metadata['connection-device-id'], 'MyTestDevice2'
        )

    def test_single_eventhub_trigger_metadata_field(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_single_iothub_datum(),
            trigger_metadata=self._generate_single_trigger_metadatum()
        )

        # Ensure the event enqueue_time property reflects the sys prop
        self.assertEqual(result.enqueued_time, self.MOCKED_ENQUEUE_TIME)

        # System Properties should be propagated in metadata
        metadata_dict = result.metadata
        self.assertIsNotNone(metadata_dict.get('SystemProperties'))

        # EnqueuedTime should be in iso8601 string format
        self.assertEqual(metadata_dict['EnqueuedTimeUtc'],
                         self.MOCKED_ENQUEUE_TIME.isoformat())
        self.assertEqual(metadata_dict['SystemProperties'][
            'iothub-connection-device-id'
        ], 'MyTestDevice')

    def test_multiple_eventhub_triggers_metadata_field(self):
        result = azf_eh.EventHubTriggerConverter.decode(
            data=self._generate_multiple_iothub_data(),
            trigger_metadata=self._generate_multiple_trigger_metadata()
        )

        # Any of the event should contain the full metadata
        event = result[0]
        metadata_dict = event.metadata
        self.assertIsNotNone(metadata_dict)

        # Ensure the event enqueue_time property reflects the sys prop
        self.assertEqual(event.enqueued_time, self.MOCKED_ENQUEUE_TIME)

        # Multiple metadata should be reflected in the list
        self.assertIsNotNone(metadata_dict.get('SystemPropertiesArray'))

        # EnqueuedTimeUtc should be in iso8601 string format
        self.assertEqual(metadata_dict['SystemPropertiesArray'][0][
            'EnqueuedTimeUtc'], self.MOCKED_ENQUEUE_TIME.isoformat())
        self.assertEqual(metadata_dict['SystemPropertiesArray'][0][
            'iothub-connection-device-id'
        ], 'MyTestDevice1')

    def test_eventhub_properties(self):
        """Test if properties from public interface _eventhub.py returns
        the correct values from metadata"""

        result = azf_eh.EventHubTriggerConverter.decode(
            data=meta.Datum(b'body_bytes', 'bytes'),
            trigger_metadata=self._generate_full_metadata()
        )

        self.assertEqual(result.get_body(), b'body_bytes')
        self.assertIsNone(result.partition_key)
        self.assertDictEqual(result.iothub_metadata,
                             {'connection-device-id': 'awesome-device-id'})
        self.assertEqual(result.sequence_number, 47)
        self.assertEqual(result.enqueued_time.isoformat(),
                         '2020-07-14T01:27:55.627000+00:00')
        self.assertEqual(result.offset, '3696')

    @patch('azure.functions.eventhub.EventHubConverter'
           '.decode_single_event')
    @patch('azure.functions.eventhub.EventHubConverter'
           '.decode_multiple_events')
    def test_eventhub_decode_call_single_event(self, dme_mock, dse_mock):
        azf_eh.EventHubConverter.decode(
            data=self._generate_single_iothub_datum(),
            trigger_metadata=self._generate_single_trigger_metadatum()
        )
        dse_mock.assert_called_once()
        dme_mock.assert_not_called()

    @patch('azure.functions.eventhub.EventHubConverter'
           '.decode_single_event')
    @patch('azure.functions.eventhub.EventHubConverter'
           '.decode_multiple_events')
    def test_eventhub_decode_call_multiple_events(self, dme_mock, dse_mock):
        azf_eh.EventHubConverter.decode(
            data=self._generate_multiple_iothub_data(
                data_type='collection_bytes'),
            trigger_metadata=self._generate_multiple_trigger_metadata()
        )
        dse_mock.assert_not_called()
        dme_mock.assert_called_once()

    def test_eventhub_single_event_json(self):
        result = azf_eh.EventHubConverter.decode(
            data=self._generate_single_iothub_datum('json'),
            trigger_metadata=self._generate_single_trigger_metadatum()
        )
        # Result body always has the datatype of bytes
        self.assertEqual(
            result.get_body().decode('utf-8'), '{"device-status": "good"}'
        )

    def test_eventhub_single_event_bytes(self):
        result = azf_eh.EventHubConverter.decode(
            data=self._generate_single_iothub_datum('bytes'),
            trigger_metadata=self._generate_single_trigger_metadatum()
        )
        self.assertEqual(
            result.get_body().decode('utf-8'), '{"device-status": "good"}'
        )

    def test_eventhub_multiple_events_collection_bytes(self):
        result = azf_eh.EventHubConverter.decode(
            data=self._generate_multiple_iothub_data('collection_bytes'),
            trigger_metadata=self._generate_multiple_trigger_metadata()
        )
        self.assertIsInstance(result, list)
        self.assertEqual(len(result), 2)

        self.assertEqual(
            result[0].get_body().decode('utf-8'), '{"device-status": "good1"}'
        )

        self.assertEqual(
            result[1].get_body().decode('utf-8'), '{"device-status": "good2"}'
        )

    def test_eventhub_multiple_events_collection_string(self):
        result = azf_eh.EventHubConverter.decode(
            data=self._generate_multiple_iothub_data('collection_string'),
            trigger_metadata=self._generate_multiple_trigger_metadata()
        )
        self.assertIsInstance(result, list)
        self.assertEqual(len(result), 2)

        self.assertEqual(
            result[0].get_body(), '{"device-status": "good1"}'
        )

        self.assertEqual(
            result[1].get_body(), '{"device-status": "good2"}'
        )

    def test_eventhub_encode_string(self):
        result = azf_eh.EventHubConverter.encode(obj="dummy_string",
                                                 expected_type=None)

        self.assertEqual(result.type, "string")
        self.assertEqual(result.value, "dummy_string")

    def test_eventhub_encode_bytes(self):
        result = azf_eh.EventHubConverter.encode(obj=b"dummy_bytes",
                                                 expected_type=None)

        self.assertEqual(result.type, "bytes")
        self.assertEqual(result.value, b"dummy_bytes")

    def test_eventhub_encode_int(self):
        result = azf_eh.EventHubConverter.encode(obj=1,
                                                 expected_type=None)

        self.assertEqual(result.type, "int")
        self.assertEqual(result.value, 1)

    def test_eventhub_encode_json(self):
        data = ["dummy_val1", "dummy_val2"]
        result = azf_eh.EventHubConverter.encode(obj=data,
                                                 expected_type=None)

        self.assertEqual(result.type, "json")
        self.assertEqual(result.value, json.dumps(data))

    def _generate_full_metadata(self):
        mocked_metadata: Mapping[str, meta.Datum] = {}
        mocked_metadata['Offset'] = meta.Datum(type='string', value='3696')
        mocked_metadata['EnqueuedTimeUtc'] = meta.Datum(
            type='string', value='2020-07-14T01:27:55.627Z')
        mocked_metadata['SequenceNumber'] = meta.Datum(type='int', value=47)
        mocked_metadata['Properties'] = meta.Datum(type='json', value='{}')
        mocked_metadata['sys'] = meta.Datum(type='json', value='''
        {
            "MethodName":"metadata_trigger",
            "UtcNow":"2020-07-14T01:27:55.8940305Z",
            "RandGuid":"db413fd6-8411-4e51-844c-c9b5345e537d"
        }''')
        mocked_metadata['SystemProperties'] = meta.Datum(type='json', value='''
        {
            "x-opt-sequence-number":47,
            "x-opt-offset":"3696",
            "x-opt-enqueued-time":"2020-07-14T01:27:55.627Z",
            "SequenceNumber":47,
            "Offset":"3696",
            "PartitionKey":null,
            "EnqueuedTimeUtc":"2020-07-14T01:27:55.627Z",
            "iothub-connection-device-id":"awesome-device-id"
        }''')
        mocked_metadata['PartitionContext'] = meta.Datum(type='json', value='''
        {
            "CancellationToken":{
            "IsCancellationRequested":false,
            "CanBeCanceled":true,
            "WaitHandle":{
                "Handle":{
                    "value":2472
                },
                "SafeWaitHandle":{
                    "IsInvalid":false,
                    "IsClosed":false
                }
            }
            },
            "ConsumerGroupName":"$Default",
            "EventHubPath":"python-worker-ci-eventhub-one-metadata",
            "PartitionId":"0",
            "Owner":"88cec2e2-94c9-4e08-acb6-4f2b97cd888e",
            "RuntimeInformation":{
            "PartitionId":"0",
            "LastSequenceNumber":0,
            "LastEnqueuedTimeUtc":"0001-01-01T00:00:00",
            "LastEnqueuedOffset":null,
            "RetrievalTime":"0001-01-01T00:00:00"
            }
        }''')

        return mocked_metadata

    def _generate_single_iothub_datum(self, datum_type='json'):
        datum = '{"device-status": "good"}'
        if datum_type == 'bytes':
            datum = datum.encode('utf-8')

        return meta.Datum(datum, datum_type)

    def _generate_multiple_iothub_data(self, data_type='json'):
        data = '[{"device-status": "good1"}, {"device-status": "good2"}]'
        if data_type == 'collection_bytes':
            data = list(
                map(lambda x: json.dumps(x).encode('utf-8'), json.loads(data))
            )
            data = CollectionBytes(data)
        elif data_type == 'collection_string':
            data = list(
                map(lambda x: json.dumps(x), json.loads(data))
            )
            data = CollectionString(data)

        return meta.Datum(data, data_type)

    def _generate_single_trigger_metadatum(self):
        return {
            'EnqueuedTimeUtc': meta.Datum(
                f'"{self.MOCKED_ENQUEUE_TIME.isoformat()}"', 'json'
            ),
            'SystemProperties': meta.Datum(
                '{"iothub-connection-device-id": "MyTestDevice"}', 'json'
            )
        }

    def _generate_multiple_trigger_metadata(self):
        system_props_array = [
            {
                'EnqueuedTimeUtc': self.MOCKED_ENQUEUE_TIME.isoformat(),
                'iothub-connection-device-id': 'MyTestDevice1',
            },
            {
                'EnqueuedTimeUtc': self.MOCKED_ENQUEUE_TIME.isoformat(),
                'iothub-connection-device-id': 'MyTestDevice2',
            }
        ]

        return {
            'SystemPropertiesArray': meta.Datum(
                json.dumps(system_props_array), 'json'
            )
        }