-
Notifications
You must be signed in to change notification settings - Fork 68
/
Copy pathtest_eventgrid.py
166 lines (144 loc) · 6.43 KB
/
test_eventgrid.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
from datetime import datetime
import unittest
from typing import List
import azure.functions as func
import azure.functions.eventgrid as azf_event_grid
class MyTestCase(unittest.TestCase):
def test_eventgrid_input_type(self):
check_input_type = azf_event_grid.EventGridEventInConverter.\
check_input_type_annotation
self.assertTrue(check_input_type(func.EventGridEvent))
self.assertFalse(check_input_type(List[func.EventGridEvent]))
self.assertFalse(check_input_type(str))
self.assertFalse(check_input_type(bytes))
def test_eventgrid_output_type(self):
check_output_type = azf_event_grid.EventGridEventOutConverter.\
check_output_type_annotation
self.assertTrue(check_output_type(func.EventGridOutputEvent))
self.assertTrue(check_output_type(List[func.EventGridOutputEvent]))
self.assertTrue(check_output_type(str))
self.assertTrue(check_output_type(bytes))
self.assertTrue(check_output_type(List[str]))
def test_eventgrid_decode(self):
eventGridEvent = azf_event_grid.EventGridEventInConverter.decode(
data=self._generate_single_eventgrid_datum(), trigger_metadata=None
)
self.assertEqual(
eventGridEvent.id,
"00010001-0001-0001-0001-000100010001")
self.assertEqual(eventGridEvent.subject, "eventhubs/test")
self.assertEqual(eventGridEvent.event_type, "captureFileCreated")
self.assertEqual(eventGridEvent.topic, "/TestTopic/namespaces/test")
self.assertIsNotNone(eventGridEvent.get_json())
def test_eventgrid_decode_with_null_data(self):
eventGridEvent = azf_event_grid.EventGridEventInConverter.decode(
data=self._generate_single_eventgrid_datum(
with_data=False), trigger_metadata=None)
self.assertEqual(
eventGridEvent.id,
"00010001-0001-0001-0001-000100010001")
self.assertEqual(eventGridEvent.subject, "eventhubs/test")
self.assertEqual(eventGridEvent.event_type, "captureFileCreated")
self.assertEqual(eventGridEvent.topic, "/TestTopic/namespaces/test")
self.assertIsNone(eventGridEvent.get_json())
def test_eventgrid_encode_with_str_data(self):
example_data = self._generate_single_eventgrid_str()
eventGridDatum = azf_event_grid.EventGridEventOutConverter.encode(
example_data, expected_type=type(example_data))
self.assertEqual(eventGridDatum.type, "string")
def test_eventgrid_encode_with_bytes_data(self):
example_data = self._generate_single_eventgrid_str(True)
eventGridDatum = azf_event_grid.EventGridEventOutConverter.encode(
example_data, expected_type=type(example_data))
self.assertEqual(eventGridDatum.type, "bytes")
def test_eventgrid_encode_with_EventGridData(self):
example_data = self._generate_single_eventgrid_event()
event_grid_datum = azf_event_grid.EventGridEventOutConverter.encode(
example_data, expected_type=type(example_data))
self.assertEqual(event_grid_datum.type, "json")
def test_eventgrid_encode_with_multiple_EventGridData(self):
example_data = self._generate_multiple_eventgrid_event()
event_grid_datum = azf_event_grid.EventGridEventOutConverter.encode(
example_data, expected_type=type(example_data))
self.assertEqual(event_grid_datum.type, "json")
@staticmethod
def _generate_single_eventgrid_datum(with_data=True, datum_type='json'):
datum_with_data = """
{
"topic": "/TestTopic/namespaces/test",
"subject": "eventhubs/test",
"eventType": "captureFileCreated",
"eventTime": "2017-07-14T23:10:27.7689666Z",
"id": "00010001-0001-0001-0001-000100010001",
"data": {
"fileUrl": "https://test.blob.core.windows.net/debugging/testblob.txt",
"fileType": "AzureBlockBlob",
"partitionId": "1",
"sizeInBytes": 0,
"eventCount": 0,
"firstSequenceNumber": -1,
"lastSequenceNumber": -1,
"firstEnqueueTime": "0001-01-01T00:00:00",
"lastEnqueueTime": "0001-01-01T00:00:00"
},
"dataVersion": "",
"metadataVersion": "1"
}
"""
datum_without_data = """
{
"topic": "/TestTopic/namespaces/test",
"subject": "eventhubs/test",
"eventType": "captureFileCreated",
"eventTime": "2017-07-14T23:10:27.7689666Z",
"id": "00010001-0001-0001-0001-000100010001",
"dataVersion": "",
"metadataVersion": "1"
}"""
datum = datum_with_data if with_data else datum_without_data
if datum_type == 'bytes':
datum = datum.encode('utf-8')
return func.meta.Datum(datum, datum_type)
@staticmethod
def _generate_single_eventgrid_event(with_date=True):
return azf_event_grid.azf_eventgrid.EventGridOutputEvent(
id="id",
subject='subject',
event_type='eventType',
event_time=datetime.utcnow(),
data={"tag1": "value1", "tag2": "value2"} if with_date else {},
data_version='dataVersion',
)
@staticmethod
def _generate_multiple_eventgrid_event(with_date=True):
return [azf_event_grid.azf_eventgrid.EventGridOutputEvent(
id="id1",
subject='subject1',
event_type='eventType1',
event_time=datetime.utcnow(),
data={"tag1": "value1", "tag2": "value2"} if with_date else {},
data_version='dataVersion',
), azf_event_grid.azf_eventgrid.EventGridOutputEvent(
id="id2",
subject='subject2',
event_type='eventType2',
event_time=datetime.utcnow(),
data={"tag1": "value1", "tag2": "value2"} if with_date else {},
data_version='dataVersion',
)]
@staticmethod
def _generate_single_eventgrid_str(in_bytes=False):
string_representation = '{"id": "id", ' \
'"subject": "subject", ' \
'"dataVersion": "dataVersion", ' \
'"eventType": "eventType", ' \
'"data": {"tag1": "value1", ' \
'"tag2": "value2"}, ' \
'"eventTime": "2020-04-22T18:19:19Z"}'
return string_representation.encode('utf-8') \
if in_bytes \
else string_representation
if __name__ == '__main__':
unittest.main()