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
|
#-------------------------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License. See License.txt in the project root for
# license information.
#--------------------------------------------------------------------------
import asyncio
import pytest
import time
from azure.eventhub import EventData, TransportType
from azure.eventhub.exceptions import EventHubError
from azure.eventhub.aio import EventHubConsumerClient
@pytest.mark.liveTest
@pytest.mark.asyncio
async def test_receive_end_of_stream_async(connstr_senders):
async def on_event(partition_context, event):
if partition_context.partition_id == "0":
assert event.body_as_str() == "Receiving only a single event"
assert list(event.body)[0] == b"Receiving only a single event"
on_event.called = True
on_event.called = False
connection_str, senders = connstr_senders
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group='$default')
async with client:
task = asyncio.ensure_future(client.receive(on_event, partition_id="0", starting_position="@latest"))
await asyncio.sleep(10)
assert on_event.called is False
senders[0].send(EventData(b"Receiving only a single event"))
await asyncio.sleep(10)
assert on_event.called is True
await task
@pytest.mark.parametrize("position, inclusive, expected_result",
[("offset", False, "Exclusive"),
("offset", True, "Inclusive"),
("sequence", False, "Exclusive"),
("sequence", True, "Inclusive"),
("enqueued_time", False, "Exclusive")])
@pytest.mark.liveTest
@pytest.mark.asyncio
async def test_receive_with_event_position_async(connstr_senders, position, inclusive, expected_result):
async def on_event(partition_context, event):
assert partition_context.last_enqueued_event_properties.get('sequence_number') == event.sequence_number
assert partition_context.last_enqueued_event_properties.get('offset') == event.offset
assert partition_context.last_enqueued_event_properties.get('enqueued_time') == event.enqueued_time
assert partition_context.last_enqueued_event_properties.get('retrieval_time') is not None
if position == "offset":
on_event.event_position = event.offset
elif position == "sequence":
on_event.event_position = event.sequence_number
else:
on_event.event_position = event.enqueued_time
on_event.event = event
on_event.event_position = None
connection_str, senders = connstr_senders
senders[0].send(EventData(b"Inclusive"))
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group='$default')
async with client:
task = asyncio.ensure_future(client.receive(on_event,
starting_position="-1",
starting_position_inclusive=inclusive,
track_last_enqueued_event_properties=True))
await asyncio.sleep(10)
assert on_event.event_position is not None
await task
senders[0].send(EventData(expected_result))
client2 = EventHubConsumerClient.from_connection_string(connection_str, consumer_group='$default')
async with client2:
task = asyncio.ensure_future(
client2.receive(on_event,
starting_position=on_event.event_position, starting_position_inclusive=inclusive,
track_last_enqueued_event_properties=True))
await asyncio.sleep(10)
assert on_event.event.body_as_str() == expected_result
await task
@pytest.mark.liveTest
@pytest.mark.asyncio
async def test_receive_owner_level_async(connstr_senders):
app_prop = {"raw_prop": "raw_value"}
async def on_event(partition_context, event):
pass
async def on_error(partition_context, error):
on_error.error = error
on_error.error = None
connection_str, senders = connstr_senders
client1 = EventHubConsumerClient.from_connection_string(connection_str, consumer_group='$default')
client2 = EventHubConsumerClient.from_connection_string(connection_str, consumer_group='$default')
async with client1, client2:
task1 = asyncio.ensure_future(client1.receive(on_event,
partition_id="0", starting_position="-1",
on_error=on_error))
for i in range(5):
ed = EventData("Event Number {}".format(i))
senders[0].send(ed)
await asyncio.sleep(10)
task2 = asyncio.ensure_future(client2.receive(on_event,
partition_id="0", starting_position="-1",
owner_level=1))
for i in range(5):
ed = EventData("Event Number {}".format(i))
senders[0].send(ed)
await asyncio.sleep(10)
await task1
await task2
assert isinstance(on_error.error, EventHubError)
@pytest.mark.liveTest
@pytest.mark.asyncio
async def test_receive_over_websocket_async(connstr_senders):
app_prop = {"raw_prop": "raw_value"}
async def on_event(partition_context, event):
on_event.received.append(event)
on_event.app_prop = event.properties
on_event.received = []
on_event.app_prop = None
connection_str, senders = connstr_senders
client = EventHubConsumerClient.from_connection_string(connection_str, consumer_group='$default',
transport_type=TransportType.AmqpOverWebsocket)
event_list = []
for i in range(5):
ed = EventData("Event Number {}".format(i))
ed.properties = app_prop
event_list.append(ed)
senders[0].send(event_list)
async with client:
task = asyncio.ensure_future(client.receive(on_event,
partition_id="0", starting_position="-1"))
await asyncio.sleep(10)
await task
assert len(on_event.received) == 5
for ed in on_event.received:
assert ed.properties[b"raw_prop"] == b"raw_value"
|