Azure · YijunXieMS · Oct 3, 2019 · Sep 24, 2019 · Sep 24, 2019 · Sep 24, 2019
@@ -7,6 +7,7 @@
 import logging
 from typing import List, Any
 import time
+from distutils.version import StrictVersion
 
 import uamqp  # type: ignore
 from uamqp import errors, types, utils  # type: ignore
@@ -132,7 +133,7 @@ def _create_handler(self):
         if self._offset is not None:
             source.set_filter(self._offset._selector())  # pylint:disable=protected-access
 
-        if uamqp.__version__ <= "1.2.2":  # backward compatible until uamqp 1.2.3 is released
+        if StrictVersion(uamqp.__version__) < StrictVersion("1.2.3"):  # backward compatible until uamqp 1.2.3 is released
             desired_capabilities = {}
         elif self._track_last_enqueued_event_properties:
                 symbol_array = [types.AMQPSymbol(self._receiver_runtime_metric_symbol)]

@@ -13,6 +13,7 @@
 from uamqp import types, errors, utils  # type: ignore
 from uamqp import ReceiveClient, Source  # type: ignore
 import uamqp
+from distutils.version import StrictVersion
 
 from azure.eventhub.common import EventData, EventPosition
 from azure.eventhub.error import _error_handler
@@ -129,7 +130,7 @@ def _create_handler(self):
         if self._offset is not None:
             source.set_filter(self._offset._selector())  # pylint:disable=protected-access
 
-        if uamqp.__version__ <= "1.2.2":  # backward compatible until uamqp 1.2.3 is released
+        if StrictVersion(uamqp.__version__) < StrictVersion("1.2.3"):  # backward compatible until uamqp 1.2.3 is released
             desired_capabilities = {}
         elif self._track_last_enqueued_event_properties:
                 symbol_array = [types.AMQPSymbol(self._receiver_runtime_metric_symbol)]

@@ -25,7 +25,7 @@
 
 def pytest_addoption(parser):
     parser.addoption(
-        "--sleep", action="store", default="True", help="sleep on reconnect test: True or False"
+        "--sleep", action="store", default="False", help="sleep on reconnect test: True or False"
     )
 
 
@@ -82,7 +82,7 @@ def cleanup_eventhub(eventhub_config, hub_name, client=None):
     client.delete_event_hub(hub_name)
 
 
-@pytest.fixture()
+@pytest.fixture(scope="session")
 def live_eventhub_config():
     try:
         config = {}

@@ -37,7 +37,7 @@
 
 exclude_packages = [
         'tests',
-        "tests.asynctests",
+        'stress',
         'examples',
         # Exclude packages that will be covered by PEP420 or nspkg
         'azure',

@@ -218,19 +218,6 @@ async def test_create_batch_with_invalid_hostname_async(invalid_hostname):
         await sender.close()
 
 
-@pytest.mark.liveTest
-@pytest.mark.asyncio
-async def test_create_batch_with_none_async(connection_str):
-    client = EventHubClient.from_connection_string(connection_str, network_tracing=False)
-    sender = client.create_producer()
-    batch_event_data = await sender.create_batch(max_size=300, partition_key="key")
-    try:
-        with pytest.raises(ValueError):
-            batch_event_data.try_add(EventData(None))
-    finally:
-        await sender.close()
-
-
 @pytest.mark.liveTest
 @pytest.mark.asyncio
 async def test_create_batch_with_too_large_size_async(connection_str):

@@ -325,7 +325,10 @@ async def test_receive_over_websocket_async(connstr_senders):
 @pytest.mark.asyncio
 @pytest.mark.liveTest
 async def test_receive_run_time_metric_async(connstr_senders):
-    pytest.skip("Disabled for uamqp 1.2.2. Will enable after uamqp 1.2.3 is released.")
+    from uamqp import __version__ as uamqp_version
+    from distutils.version import StrictVersion
+    if StrictVersion(uamqp_version) < StrictVersion('1.2.3'):
+        pytest.skip("Disabled for uamqp 1.2.2. Will enable after uamqp 1.2.3 is released.")
     connection_str, senders = connstr_senders
     client = EventHubClient.from_connection_string(connection_str, transport_type=TransportType.AmqpOverWebsocket,
                                                    network_tracing=False)

@@ -42,45 +42,22 @@ async def test_send_with_long_interval_async(connstr_receivers, sleep):
     assert list(received[0].body)[0] == b"A single event"
 
 
-def pump(receiver):
-    messages = []
-    with receiver:
-        batch = receiver.receive(timeout=1)
-        messages.extend(batch)
-        while batch:
-            batch = receiver.receive(timeout=1)
-            messages.extend(batch)
-    return messages
-
-
 @pytest.mark.liveTest
 @pytest.mark.asyncio
 async def test_send_with_forced_conn_close_async(connstr_receivers, sleep):
-    pytest.skip("This test is similar to the above one")
     connection_str, receivers = connstr_receivers
     client = EventHubClient.from_connection_string(connection_str, network_tracing=False)
     sender = client.create_producer()
     try:
         await sender.send(EventData(b"A single event"))
-        if sleep:
-            await asyncio.sleep(300)
-        else:
-            sender._handler._connection._conn.destroy()
-        await sender.send(EventData(b"A single event"))
-        await sender.send(EventData(b"A single event"))
-        if sleep:
-            await asyncio.sleep(300)
-        else:
-            sender._handler._connection._conn.destroy()
-        await sender.send(EventData(b"A single event"))
+        sender._handler._connection._conn.destroy()
         await sender.send(EventData(b"A single event"))
     finally:
         await sender.close()
 
     received = []
     for r in receivers:
-        if not sleep:
-            r._handler._connection._conn.destroy()
-        received.extend(pump(r))
-    assert len(received) == 5
+        r._handler._connection._conn.destroy()
+        received.extend(r.receive(timeout=5))
+    assert len(received) == 2
     assert list(received[0].body)[0] == b"A single event"
@@ -21,8 +21,7 @@
 
 
 @pytest.mark.liveTest
-def test_send_with_invalid_hostname(invalid_hostname, connstr_receivers):
-    _, receivers = connstr_receivers
+def test_send_with_invalid_hostname(invalid_hostname):
     client = EventHubClient.from_connection_string(invalid_hostname, network_tracing=False)
     sender = client.create_producer()
     with pytest.raises(AuthenticationError):
@@ -40,8 +39,7 @@ def test_receive_with_invalid_hostname_sync(invalid_hostname):
 
 
 @pytest.mark.liveTest
-def test_send_with_invalid_key(invalid_key, connstr_receivers):
-    _, receivers = connstr_receivers
+def test_send_with_invalid_key(invalid_key):
     client = EventHubClient.from_connection_string(invalid_key, network_tracing=False)
     sender = client.create_producer()
     with pytest.raises(AuthenticationError):
@@ -60,8 +58,7 @@ def test_receive_with_invalid_key_sync(invalid_key):
 
 
 @pytest.mark.liveTest
-def test_send_with_invalid_policy(invalid_policy, connstr_receivers):
-    _, receivers = connstr_receivers
+def test_send_with_invalid_policy(invalid_policy):
     client = EventHubClient.from_connection_string(invalid_policy, network_tracing=False)
     sender = client.create_producer()
     with pytest.raises(AuthenticationError):
@@ -226,16 +223,6 @@ def test_create_batch_with_invalid_hostname_sync(invalid_hostname):
     sender.close()
 
 
-@pytest.mark.liveTest
-def test_create_batch_with_none_sync(connection_str):
-    client = EventHubClient.from_connection_string(connection_str, network_tracing=False)
-    sender = client.create_producer()
-    batch_event_data = sender.create_batch(max_size=300, partition_key="key")
-    with pytest.raises(ValueError):
-        batch_event_data.try_add(EventData(None))
-    sender.close()
-
-
 @pytest.mark.liveTest
 def test_create_batch_with_too_large_size_sync(connection_str):
     client = EventHubClient.from_connection_string(connection_str, network_tracing=False)

@@ -282,7 +282,10 @@ def test_receive_over_websocket_sync(connstr_senders):
 
 @pytest.mark.liveTest
 def test_receive_run_time_metric(connstr_senders):
-    pytest.skip("Disabled for uamqp 1.2.2. Will enable after uamqp 1.2.3 is released.")
+    from uamqp import __version__ as uamqp_version
+    from distutils.version import StrictVersion
+    if StrictVersion(uamqp_version) < StrictVersion('1.2.3'):
+        pytest.skip("Disabled for uamqp 1.2.2. Will enable after uamqp 1.2.3 is released.")
     connection_str, senders = connstr_senders
     client = EventHubClient.from_connection_string(connection_str, transport_type=TransportType.AmqpOverWebsocket,
                                                    network_tracing=False)

@@ -40,30 +40,18 @@ def test_send_with_long_interval_sync(connstr_receivers, sleep):
 
 @pytest.mark.liveTest
 def test_send_with_forced_conn_close_sync(connstr_receivers, sleep):
-    pytest.skip("This test is similar to the above one")
     connection_str, receivers = connstr_receivers
     client = EventHubClient.from_connection_string(connection_str, network_tracing=False)
     sender = client.create_producer()
     with sender:
         sender.send(EventData(b"A single event"))
         sender._handler._connection._conn.destroy()
-        if sleep:
-            time.sleep(300)
-        else:
-            sender._handler._connection._conn.destroy()
-        sender.send(EventData(b"A single event"))
-        sender.send(EventData(b"A single event"))
-        if sleep:
-            time.sleep(300)
-        else:
-            sender._handler._connection._conn.destroy()
-        sender.send(EventData(b"A single event"))
         sender.send(EventData(b"A single event"))
 
     received = []
     for r in receivers:
         if not sleep:
             r._handler._connection._conn.destroy()
-        received.extend(r.receive(timeout=1))
-    assert len(received) == 5
+        received.extend(r.receive(timeout=5))
+    assert len(received) == 2
     assert list(received[0].body)[0] == b"A single event"
@@ -52,7 +52,7 @@ def test_send_and_receive_large_body_size(connstr_receivers):
 
     received = []
     for r in receivers:
-        received.extend(r.receive(timeout=4))
+        received.extend(r.receive(timeout=10))
 
     assert len(received) == 1
     assert len(list(received[0].body)[0]) == payload

@@ -0,0 +1,43 @@
+import pytest
+from azure.eventhub import EventData, EventDataBatch
+from uamqp import Message
+
+
+@pytest.mark.parametrize("test_input, expected_result",
+                         [("", ""), ("AAA", "AAA"), (None, ValueError), (["a", "b", "c"], "abc"), (b"abc", "abc")])
+def test_constructor(test_input, expected_result):
+    if isinstance(expected_result, type):
+        with pytest.raises(expected_result):
+            EventData(test_input)
+    else:
+        event_data = EventData(test_input)
+        assert event_data.body_as_str() == expected_result
+        assert event_data.partition_key is None
+        assert event_data.application_properties is None
+        assert event_data.enqueued_time is None
+        assert event_data.offset is None
+        assert event_data.sequence_number is None
+        assert event_data.system_properties == {}
+        with pytest.raises(TypeError):
+            event_data.body_as_json()
+
+
+def test_body_json():
+    event_data = EventData('{"a":"b"}')
+    jo = event_data.body_as_json()
+    assert jo["a"] == "b"
+
+
+def test_app_properties():
+    app_props = {"a": "b"}
+    event_data = EventData("")
+    event_data.application_properties = app_props
+    assert event_data.application_properties["a"] == "b"
+
+
+def test_evetn_data_batch():
+    batch = EventDataBatch(max_size=100, partition_key="par")
+    batch.try_add(EventData("A"))
+    assert batch.size == 89 and len(batch) == 1
+    with pytest.raises(ValueError):
+        batch.try_add(EventData("A"))