[Event Hubs] Use EventHubManagementClient to create test resources

sdk/eventhub/azure-eventhub/conftest.py

@@ -3,31 +3,37 @@
 # Licensed under the MIT License. See License.txt in the project root for
 # license information.
 #--------------------------------------------------------------------------
-
+import sys
 import os
 import pytest
 import logging
-import sys
 import uuid
 import warnings
 from logging.handlers import RotatingFileHandler
 
+from azure.identity import EnvironmentCredential
+from azure.mgmt.resource import ResourceManagementClient
+from azure.mgmt.eventhub import EventHubManagementClient
+from azure.eventhub import EventHubProducerClient
+from uamqp import ReceiveClient
+from uamqp.authentication import SASTokenAuth
+
+
 # Ignore async tests for Python < 3.5
 collect_ignore = []
 if sys.version_info < (3, 5):
     collect_ignore.append("tests/livetest/asynctests")
     collect_ignore.append("tests/unittest/asynctests")
     collect_ignore.append("features")
     collect_ignore.append("samples/async_samples")
-    collect_ignore.append("examples/async_examples")
-
-from azure.servicebus._control_client import ServiceBusService, EventHub
-from azure.eventhub import EventHubProducerClient
-import uamqp
-from uamqp import authentication
-
 PARTITION_COUNT = 2
 CONN_STR = "Endpoint=sb://{}/;SharedAccessKeyName={};SharedAccessKey={};EntityPath={}"
+RES_GROUP_PREFIX = "eh-res-group"
+NAMESPACE_PREFIX = "eh-ns"
+EVENTHUB_PREFIX = "eh"
+EVENTHUB_DEFAULT_AUTH_RULE_NAME = 'RootManageSharedAccessKey'
+LOCATION = "westus"
+
 
 def pytest_addoption(parser):
     parser.addoption(
@@ -66,59 +72,81 @@ def get_logger(filename, level=logging.INFO):
 log = get_logger(None, logging.DEBUG)
 
 
-def create_eventhub(eventhub_config, client=None):
-    hub_name = str(uuid.uuid4())
-    hub_value = EventHub(partition_count=PARTITION_COUNT)
-    client = client or ServiceBusService(
-        service_namespace=eventhub_config['namespace'],
-        shared_access_key_name=eventhub_config['key_name'],
-        shared_access_key_value=eventhub_config['access_key'])
-    if client.create_event_hub(hub_name, hub=hub_value, fail_on_exist=True):
-        return hub_name
-    raise ValueError("EventHub creation failed.")
-
-
-def cleanup_eventhub(eventhub_config, hub_name, client=None):
-    client = client or ServiceBusService(
-        service_namespace=eventhub_config['namespace'],
-        shared_access_key_name=eventhub_config['key_name'],
-        shared_access_key_value=eventhub_config['access_key'])
-    client.delete_event_hub(hub_name)
+@pytest.fixture(scope="session")
+def resource_group():
+    try:
+        SUBSCRIPTION_ID = os.environ["AZURE_SUBSCRIPTION_ID"]
+    except KeyError:
+        pytest.skip('AZURE_SUBSCRIPTION_ID undefined')
+        return
+    resource_client = ResourceManagementClient(EnvironmentCredential(), SUBSCRIPTION_ID)
+    resource_group_name = RES_GROUP_PREFIX + str(uuid.uuid4())
+    try:
+        rg = resource_client.resource_groups.create_or_update(
+           resource_group_name, {"location": LOCATION}
+        )
+        yield rg
+    finally:
+        try:
+            resource_client.resource_groups.begin_delete(resource_group_name)
+        except:
+            warnings.warn(UserWarning("resource group teardown failed"))
 
 
 @pytest.fixture(scope="session")
-def live_eventhub_config():
+def eventhub_namespace(resource_group):
     try:
-        config = {}
-        config['hostname'] = os.environ['EVENT_HUB_HOSTNAME']
-        config['event_hub'] = os.environ['EVENT_HUB_NAME']
-        config['key_name'] = os.environ['EVENT_HUB_SAS_POLICY']
-        config['access_key'] = os.environ['EVENT_HUB_SAS_KEY']
-        config['namespace'] = os.environ['EVENT_HUB_NAMESPACE']
-        config['consumer_group'] = "$Default"
-        config['partition'] = "0"
-        config['connection_str'] = CONN_STR
+        SUBSCRIPTION_ID = os.environ["AZURE_SUBSCRIPTION_ID"]
     except KeyError:
-        pytest.skip("Live EventHub configuration not found.")
-    else:
-        return config
+        pytest.skip('AZURE_SUBSCRIPTION_ID defined')
+        return
+    resource_client = EventHubManagementClient(EnvironmentCredential(), SUBSCRIPTION_ID)
+    namespace_name = NAMESPACE_PREFIX + str(uuid.uuid4())
+    try:
+        namespace = resource_client.namespaces.begin_create_or_update(
+            resource_group.name, namespace_name, {"location": LOCATION}
+        ).result()
+        key = resource_client.namespaces.list_keys(resource_group.name, namespace_name, EVENTHUB_DEFAULT_AUTH_RULE_NAME)
+        connection_string = key.primary_connection_string
+        key_name = key.key_name
+        primary_key = key.primary_key
+        yield namespace.name, connection_string, key_name, primary_key
+    finally:
+        try:
+            resource_client.namespaces.begin_delete(resource_group.name, namespace_name).wait()
+        except:
+            warnings.warn(UserWarning("eventhub namespace teardown failed"))
 
 
 @pytest.fixture()
-def live_eventhub(live_eventhub_config):  # pylint: disable=redefined-outer-name
-    client = ServiceBusService(
-        service_namespace=live_eventhub_config['namespace'],
-        shared_access_key_name=live_eventhub_config['key_name'],
-        shared_access_key_value=live_eventhub_config['access_key'])
+def live_eventhub(resource_group, eventhub_namespace):  # pylint: disable=redefined-outer-name
+    try:
+        SUBSCRIPTION_ID = os.environ["AZURE_SUBSCRIPTION_ID"]
+    except KeyError:
+        pytest.skip('AZURE_SUBSCRIPTION_ID defined')
+        return
+    resource_client = EventHubManagementClient(EnvironmentCredential(), SUBSCRIPTION_ID)
+    eventhub_name = EVENTHUB_PREFIX + str(uuid.uuid4())
+    eventhub_ns_name, connection_string, key_name, primary_key = eventhub_namespace
     try:
-        hub_name = create_eventhub(live_eventhub_config, client=client)
-        print("Created EventHub {}".format(hub_name))
-        live_eventhub_config['event_hub'] = hub_name
+        eventhub = resource_client.event_hubs.create_or_update(
+            resource_group.name, eventhub_ns_name, eventhub_name, {"partition_count": PARTITION_COUNT}
+        )
+        live_eventhub_config = {
+            'resource_group': resource_group.name,
+            'hostname': "{}.servicebus.windows.net".format(eventhub_ns_name),
+            'key_name': key_name,
+            'access_key': primary_key,
+            'namespace': eventhub_ns_name,
+            'event_hub': eventhub.name,
+            'consumer_group': '$Default',
+            'partition': '0',
+            'connection_str': connection_string + ";EntityPath="+eventhub.name
+        }
         yield live_eventhub_config
     finally:
         try:
-            cleanup_eventhub(live_eventhub_config, hub_name, client=client)
-            print("Deleted EventHub {}".format(hub_name))
+            resource_client.event_hubs.delete(resource_group.name, eventhub_ns_name, eventhub_name)
         except:
             warnings.warn(UserWarning("eventhub teardown failed"))
 
@@ -133,55 +161,48 @@ def connection_str(live_eventhub):
 
 
 @pytest.fixture()
-def invalid_hostname(live_eventhub_config):
+def invalid_hostname(live_eventhub):
     return CONN_STR.format(
         "invalid123.servicebus.windows.net",
-        live_eventhub_config['key_name'],
-        live_eventhub_config['access_key'],
-        live_eventhub_config['event_hub'])
+        live_eventhub['key_name'],
+        live_eventhub['access_key'],
+        live_eventhub['event_hub'])
 
 
 @pytest.fixture()
-def invalid_key(live_eventhub_config):
+def invalid_key(live_eventhub):
     return CONN_STR.format(
-        live_eventhub_config['hostname'],
-        live_eventhub_config['key_name'],
+        live_eventhub['hostname'],
+        live_eventhub['key_name'],
         "invalid",
-        live_eventhub_config['event_hub'])
+        live_eventhub['event_hub'])
 
 
 @pytest.fixture()
-def invalid_policy(live_eventhub_config):
+def invalid_policy(live_eventhub):
     return CONN_STR.format(
-        live_eventhub_config['hostname'],
+        live_eventhub['hostname'],
         "invalid",
-        live_eventhub_config['access_key'],
-        live_eventhub_config['event_hub'])
-
-
-@pytest.fixture()
-def aad_credential():
-    try:
-        return os.environ['AZURE_CLIENT_ID'], os.environ['AZURE_CLIENT_SECRET'], os.environ['AZURE_TENANT_ID']
-    except KeyError:
-        pytest.skip('No Azure Active Directory credential found')
+        live_eventhub['access_key'],
+        live_eventhub['event_hub'])
 
 
 @pytest.fixture()
-def connstr_receivers(connection_str, live_eventhub_config):
+def connstr_receivers(live_eventhub):
+    connection_str = live_eventhub["connection_str"]
     partitions = [str(i) for i in range(PARTITION_COUNT)]
     receivers = []
     for p in partitions:
-        uri = "sb://{}/{}".format(live_eventhub_config['hostname'], live_eventhub_config['event_hub'])
-        sas_auth = authentication.SASTokenAuth.from_shared_access_key(
-            uri, live_eventhub_config['key_name'], live_eventhub_config['access_key'])
+        uri = "sb://{}/{}".format(live_eventhub['hostname'], live_eventhub['event_hub'])
+        sas_auth = SASTokenAuth.from_shared_access_key(
+            uri, live_eventhub['key_name'], live_eventhub['access_key'])
 
         source = "amqps://{}/{}/ConsumerGroups/{}/Partitions/{}".format(
-            live_eventhub_config['hostname'],
-            live_eventhub_config['event_hub'],
-            live_eventhub_config['consumer_group'],
+            live_eventhub['hostname'],
+            live_eventhub['event_hub'],
+            live_eventhub['consumer_group'],
             p)
-        receiver = uamqp.ReceiveClient(source, auth=sas_auth, debug=False, timeout=0, prefetch=500)
+        receiver = ReceiveClient(source, auth=sas_auth, debug=False, timeout=0, prefetch=500)
         receiver.open()
         receivers.append(receiver)
     yield connection_str, receivers
@@ -190,7 +211,8 @@ def connstr_receivers(connection_str, live_eventhub_config):
 
 
 @pytest.fixture()
-def connstr_senders(connection_str):
+def connstr_senders(live_eventhub):
+    connection_str = live_eventhub["connection_str"]
     client = EventHubProducerClient.from_connection_string(connection_str)
     partitions = client.get_partition_ids()
 

sdk/eventhub/azure-eventhub/dev_requirements.txt

@@ -1,7 +1,8 @@
 -e ../../../tools/azure-sdk-tools
 ../../core/azure-core
 -e ../../identity/azure-identity
--e ../../servicebus/azure-servicebus
+-e ../azure-mgmt-eventhub
+-e ../../resources/azure-mgmt-resource
 aiohttp>=3.0; python_version >= '3.5'
 docutils>=0.14
 pygments>=2.2.0

sdk/eventhub/azure-eventhub/tests/livetest/asynctests/test_auth_async.py

@@ -7,19 +7,14 @@
 import pytest
 import asyncio
 
+from azure.identity.aio import EnvironmentCredential
 from azure.eventhub import EventData
 from azure.eventhub.aio import EventHubConsumerClient, EventHubProducerClient
 
 
 @pytest.mark.liveTest
 @pytest.mark.asyncio
-async def test_client_secret_credential_async(aad_credential, live_eventhub):
-    try:
-        from azure.identity.aio import EnvironmentCredential
-    except ImportError:
-        pytest.skip("No azure identity library")
-        return
-
+async def test_client_secret_credential_async(live_eventhub):
     credential = EnvironmentCredential()
     producer_client = EventHubProducerClient(fully_qualified_namespace=live_eventhub['hostname'],
                                              eventhub_name=live_eventhub['event_hub'],

sdk/eventhub/azure-eventhub/tests/livetest/asynctests/test_negative_async.py

@@ -40,11 +40,11 @@ async def test_send_with_invalid_hostname_async(invalid_hostname, connstr_receiv
                          ["hostname", "key_name", "access_key", "event_hub", "partition"])
 @pytest.mark.liveTest
 @pytest.mark.asyncio
-async def test_receive_with_invalid_param_async(live_eventhub_config, invalid_place):
-    eventhub_config = live_eventhub_config.copy()
+async def test_receive_with_invalid_param_async(live_eventhub, invalid_place):
+    eventhub_config = live_eventhub.copy()
     if invalid_place != "partition":
         eventhub_config[invalid_place] = "invalid " + invalid_place
-    conn_str = live_eventhub_config["connection_str"].format(
+    conn_str = live_eventhub["connection_str"].format(
         eventhub_config['hostname'],
         eventhub_config['key_name'],
         eventhub_config['access_key'],

sdk/eventhub/azure-eventhub/tests/livetest/synctests/test_auth.py

@@ -7,15 +7,12 @@
 import time
 import threading
 
+from azure.identity import EnvironmentCredential
 from azure.eventhub import EventData, EventHubProducerClient, EventHubConsumerClient
 
 
 @pytest.mark.liveTest
-def test_client_secret_credential(aad_credential, live_eventhub):
-    try:
-        from azure.identity import EnvironmentCredential
-    except:
-        pytest.skip("No azure identity library")
+def test_client_secret_credential(live_eventhub):
     credential = EnvironmentCredential()
     producer_client = EventHubProducerClient(fully_qualified_namespace=live_eventhub['hostname'],
                                              eventhub_name=live_eventhub['event_hub'],

sdk/eventhub/azure-eventhub/tests/livetest/synctests/test_negative.py

@@ -50,13 +50,8 @@ def on_event(partition_context, event):
 
 
 @pytest.mark.liveTest
-def test_send_batch_with_invalid_key(live_eventhub):
-    conn_str = live_eventhub["connection_str"].format(
-        live_eventhub['hostname'],
-        live_eventhub['key_name'],
-        'invalid',
-        live_eventhub['event_hub'])
-    client = EventHubProducerClient.from_connection_string(conn_str)
+def test_send_batch_with_invalid_key(invalid_key):
+    client = EventHubProducerClient.from_connection_string(invalid_key)
     try:
         with pytest.raises(ConnectError):
             batch = EventDataBatch()

sdk/eventhub/tests.yml

@@ -33,3 +33,4 @@ jobs:
         AZURE_CLIENT_ID: $(python-eh-livetest-event-hub-aad-client-id)
         AZURE_TENANT_ID: $(python-eh-livetest-event-hub-aad-tenant-id)
         AZURE_CLIENT_SECRET: $(python-eh-livetest-event-hub-aad-secret)
+        AZURE_SUBSCRIPTION_ID: $(python-eh-livetest-event-hub-subscription-id)