Skip to content

Commit fbfebb6

Browse files
l0lawrencel0lawrence
authored
[EH] update checkpointstore tests/samples (Azure#39498)
* try * tests * skip * update * update * req * bump test * bump * git * try this * add * await * do we need all permissions here? * bump * fix * api version * update * remove * Revert "remove" This reverts commit 503d44a90a48a29a7ae0e6c3b982590cce8001a6. * add * Revert "add" This reverts commit ef03d41f7a9f5a454a38b7eed520d41c72ba8dee. * try * getcred * try * async * dev * try * try * bump * cahnge * tests * Update README.md with new documentation link * why * space * space2 * r
1 parent f0ff4e1 commit fbfebb6

9 files changed

+147
-120
lines changed

‎sdk/eventhub/azure-eventhub-checkpointstoreblob-aio/dev_requirements.txt

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
11
-e ../../../tools/azure-sdk-tools
22
../../core/azure-core
3-
../azure-eventhub
3+
../azure-eventhub
4+
azure-identity~=1.17.0

‎sdk/eventhub/azure-eventhub-checkpointstoreblob-aio/samples/receive_events_using_checkpoint_store_async.py

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,13 +1,15 @@
11
import asyncio
22
import os
3+
from azure.identity.aio import DefaultAzureCredential
34
from azure.eventhub.aio import EventHubConsumerClient
45
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
56

6-
CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"]
7+
FULLY_QUALIFIED_NAMESPACE = os.environ["EVENT_HUB_HOSTNAME"]
78
EVENTHUB_NAME = os.environ['EVENT_HUB_NAME']
8-
STORAGE_CONNECTION_STR = os.environ["AZURE_STORAGE_CONN_STR"]
9+
STORAGE_ACCOUNT = "https://{}.blob.core.windows.net".format(
10+
os.environ["AZURE_STORAGE_ACCOUNT"])
911
BLOB_CONTAINER_NAME = "your-blob-container-name" # Please make sure the blob container resource exists.
10-
12+
STORAGE_SERVICE_API_VERSION = "2019-02-02"
1113

1214
async def on_event(partition_context, event):
1315
# Put your code here.
@@ -21,12 +23,15 @@ async def main(client):
2123
await client.receive(on_event)
2224

2325
if __name__ == '__main__':
24-
checkpoint_store = BlobCheckpointStore.from_connection_string(
25-
STORAGE_CONNECTION_STR,
26+
checkpoint_store = BlobCheckpointStore(
27+
STORAGE_ACCOUNT,
2628
container_name=BLOB_CONTAINER_NAME,
29+
api_version=STORAGE_SERVICE_API_VERSION,
30+
credential=DefaultAzureCredential()
2731
)
28-
client = EventHubConsumerClient.from_connection_string(
29-
CONNECTION_STR,
32+
client = EventHubConsumerClient(
33+
FULLY_QUALIFIED_NAMESPACE,
34+
credential=DefaultAzureCredential(),
3035
consumer_group='$Default',
3136
eventhub_name=EVENTHUB_NAME,
3237
checkpoint_store=checkpoint_store

‎sdk/eventhub/azure-eventhub-checkpointstoreblob-aio/samples/receive_events_using_checkpoint_store_storage_api_version_async.py

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,16 @@
1616

1717
import asyncio
1818
import os
19+
from azure.identity.aio import DefaultAzureCredential
1920
from azure.eventhub.aio import EventHubConsumerClient
2021
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
2122

22-
CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"]
23+
FULLY_QUALIFIED_NAMESPACE = os.environ["EVENT_HUB_HOSTNAME"]
2324
EVENTHUB_NAME = os.environ['EVENT_HUB_NAME']
24-
STORAGE_CONNECTION_STR = os.environ["AZURE_STORAGE_CONN_STR"]
25+
STORAGE_ACCOUNT = "https://{}.blob.core.windows.net".format(
26+
os.environ["AZURE_STORAGE_ACCOUNT"])
2527
BLOB_CONTAINER_NAME = "your-blob-container-name" # Please make sure the blob container resource exists.
26-
STORAGE_SERVICE_API_VERSION = "2017-11-09"
28+
STORAGE_SERVICE_API_VERSION = "2019-02-02"
2729

2830

2931
async def on_event(partition_context, event):
@@ -38,13 +40,15 @@ async def main(client):
3840
await client.receive(on_event)
3941

4042
if __name__ == '__main__':
41-
checkpoint_store = BlobCheckpointStore.from_connection_string(
42-
STORAGE_CONNECTION_STR,
43+
checkpoint_store = BlobCheckpointStore(
44+
STORAGE_ACCOUNT,
4345
container_name=BLOB_CONTAINER_NAME,
44-
api_version=STORAGE_SERVICE_API_VERSION
46+
api_version=STORAGE_SERVICE_API_VERSION,
47+
credential=DefaultAzureCredential()
4548
)
46-
client = EventHubConsumerClient.from_connection_string(
47-
CONNECTION_STR,
49+
client = EventHubConsumerClient(
50+
FULLY_QUALIFIED_NAMESPACE,
51+
credential=DefaultAzureCredential(),
4852
consumer_group='$Default',
4953
eventhub_name=EVENTHUB_NAME,
5054
checkpoint_store=checkpoint_store

‎sdk/eventhub/azure-eventhub-checkpointstoreblob-aio/tests/test_storage_blob_partition_manager_aio.py

Lines changed: 37 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -10,42 +10,34 @@
1010
import uuid
1111
import warnings
1212
import asyncio
13-
13+
from functools import partial
14+
from devtools_testutils import get_credential
1415
from azure.eventhub.extensions.checkpointstoreblobaio import BlobCheckpointStore
15-
from azure.eventhub.extensions.checkpointstoreblobaio._vendor.storage.blob import BlobServiceClient
16+
from azure.eventhub.extensions.checkpointstoreblobaio._vendor.storage.blob.aio import BlobServiceClient
1617

1718
STORAGE_ENV_KEYS = [
18-
"AZURE_STORAGE_CONN_STR",
19-
"AZURE_STORAGE_DATA_LAKE_ENABLED_CONN_STR"
19+
"AZURE_STORAGE_ACCOUNT",
2020
]
2121

2222

23-
def get_live_storage_blob_client(conn_str_env_key):
24-
try:
25-
storage_connection_str = os.environ[conn_str_env_key]
26-
container_name = str(uuid.uuid4())
27-
blob_service_client = BlobServiceClient.from_connection_string(storage_connection_str)
28-
blob_service_client.create_container(container_name)
29-
return storage_connection_str, container_name
30-
except:
31-
pytest.skip("Storage blob client can't be created")
32-
33-
34-
def remove_live_storage_blob_client(storage_connection_str, container_str):
35-
try:
36-
blob_service_client = BlobServiceClient.from_connection_string(storage_connection_str)
37-
blob_service_client.delete_container(container_str)
38-
except:
39-
warnings.warn(UserWarning("storage container teardown failed"))
23+
async def get_live_storage_blob_client( storage_account):
24+
storage_account = "https://{}.blob.core.windows.net".format(
25+
os.environ[storage_account])
26+
container_name = str(uuid.uuid4())
27+
blob_service_client = BlobServiceClient(storage_account, get_credential(is_async=True))
28+
await blob_service_client.create_container(container_name)
29+
return storage_account, container_name
4030

4131

42-
async def _claim_and_list_ownership(connection_str, container_name):
32+
async def _claim_and_list_ownership( storage_account, container_name):
4333
fully_qualified_namespace = 'test_namespace'
4434
eventhub_name = 'eventhub'
4535
consumer_group = '$default'
4636
ownership_cnt = 8
4737

48-
checkpoint_store = BlobCheckpointStore.from_connection_string(connection_str, container_name)
38+
credential = get_credential(is_async=True)
39+
40+
checkpoint_store = BlobCheckpointStore(storage_account, container_name, credential=credential)
4941
async with checkpoint_store:
5042
ownership_list = await checkpoint_store.list_ownership(
5143
fully_qualified_namespace=fully_qualified_namespace,
@@ -78,13 +70,15 @@ async def _claim_and_list_ownership(connection_str, container_name):
7870
assert len(ownership_list) == ownership_cnt
7971

8072

81-
async def _update_checkpoint(connection_str, container_name):
73+
async def _update_checkpoint( storage_account, container_name):
8274
fully_qualified_namespace = 'test_namespace'
8375
eventhub_name = 'eventhub'
8476
consumer_group = '$default'
8577
partition_cnt = 8
8678

87-
checkpoint_store = BlobCheckpointStore.from_connection_string(connection_str, container_name)
79+
credential = get_credential(is_async=True)
80+
81+
checkpoint_store = BlobCheckpointStore(storage_account, container_name, credential=credential)
8882
async with checkpoint_store:
8983
for i in range(partition_cnt):
9084
checkpoint = {
@@ -107,23 +101,21 @@ async def _update_checkpoint(connection_str, container_name):
107101
assert checkpoint['sequence_number'] == 20
108102

109103

110-
@pytest.mark.parametrize("conn_str_env_key", STORAGE_ENV_KEYS)
111-
@pytest.mark.liveTest
112-
def test_claim_and_list_ownership(conn_str_env_key):
113-
storage_connection_str, container_name = get_live_storage_blob_client(conn_str_env_key)
114-
try:
115-
loop = asyncio.get_event_loop()
116-
loop.run_until_complete(_claim_and_list_ownership(storage_connection_str, container_name))
117-
finally:
118-
remove_live_storage_blob_client(storage_connection_str, container_name)
119-
120-
121-
@pytest.mark.parametrize("conn_str_env_key", STORAGE_ENV_KEYS)
122-
@pytest.mark.liveTest
123-
def test_update_checkpoint(conn_str_env_key):
124-
storage_connection_str, container_name = get_live_storage_blob_client(conn_str_env_key)
125-
try:
126-
loop = asyncio.get_event_loop()
127-
loop.run_until_complete(_update_checkpoint(storage_connection_str, container_name))
128-
finally:
129-
remove_live_storage_blob_client(storage_connection_str, container_name)
104+
@pytest.mark.parametrize("storage_account", STORAGE_ENV_KEYS)
105+
@pytest.mark.live_test_only
106+
@pytest.mark.asyncio
107+
async def test_claim_and_list_ownership_async( storage_account):
108+
storage_account, container_name = await get_live_storage_blob_client(storage_account)
109+
await _claim_and_list_ownership(storage_account, container_name)
110+
blob_service_client = BlobServiceClient(storage_account, credential=get_credential(is_async=True))
111+
blob_service_client.delete_container(container_name)
112+
113+
114+
@pytest.mark.parametrize("storage_account", STORAGE_ENV_KEYS)
115+
@pytest.mark.live_test_only
116+
@pytest.mark.asyncio
117+
async def test_update_checkpoint_async( storage_account):
118+
storage_account, container_name = await get_live_storage_blob_client(storage_account)
119+
await _update_checkpoint(storage_account, container_name)
120+
blob_service_client = BlobServiceClient(storage_account, credential=get_credential(is_async=True))
121+
blob_service_client.delete_container(container_name)

‎sdk/eventhub/azure-eventhub-checkpointstoreblob/dev_requirements.txt

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
11
-e ../../../tools/azure-sdk-tools
22
../../core/azure-core
3+
azure-identity~=1.17.0
34
../azure-eventhub

‎sdk/eventhub/azure-eventhub-checkpointstoreblob/samples/receive_events_using_checkpoint_store.py

Lines changed: 12 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,11 +1,14 @@
11
import os
2+
from azure.identity import DefaultAzureCredential
23
from azure.eventhub import EventHubConsumerClient # type: ignore[attr-defined]
34
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
45

5-
CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"]
6+
FULLY_QUALIFIED_NAMESPACE = os.environ["EVENT_HUB_HOSTNAME"]
67
EVENTHUB_NAME = os.environ['EVENT_HUB_NAME']
7-
STORAGE_CONNECTION_STR = os.environ["AZURE_STORAGE_CONN_STR"]
8+
STORAGE_ACCOUNT = "https://{}.blob.core.windows.net".format(
9+
os.environ["AZURE_STORAGE_ACCOUNT"])
810
BLOB_CONTAINER_NAME = "your-blob-container-name" # Please make sure the blob container resource exists.
11+
STORAGE_SERVICE_API_VERSION = "2019-02-02"
912

1013

1114
def on_event(partition_context, event):
@@ -16,12 +19,15 @@ def on_event(partition_context, event):
1619

1720

1821
if __name__ == '__main__':
19-
checkpoint_store = BlobCheckpointStore.from_connection_string(
20-
STORAGE_CONNECTION_STR,
22+
checkpoint_store = BlobCheckpointStore(
23+
STORAGE_ACCOUNT,
2124
container_name=BLOB_CONTAINER_NAME,
25+
api_version=STORAGE_SERVICE_API_VERSION,
26+
credential=DefaultAzureCredential()
2227
)
23-
client = EventHubConsumerClient.from_connection_string(
24-
CONNECTION_STR,
28+
client = EventHubConsumerClient(
29+
FULLY_QUALIFIED_NAMESPACE,
30+
credential=DefaultAzureCredential(),
2531
consumer_group='$Default',
2632
eventhub_name=EVENTHUB_NAME,
2733
checkpoint_store=checkpoint_store

‎sdk/eventhub/azure-eventhub-checkpointstoreblob/samples/receive_events_using_checkpoint_store_storage_api_version.py

Lines changed: 12 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,14 +15,16 @@
1515
"""
1616

1717
import os
18+
from azure.identity import DefaultAzureCredential
1819
from azure.eventhub import EventHubConsumerClient # type: ignore[attr-defined]
1920
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
2021

21-
CONNECTION_STR = os.environ["EVENT_HUB_CONN_STR"]
22+
FULLY_QUALIFIED_NAMESPACE = os.environ["EVENT_HUB_HOSTNAME"]
2223
EVENTHUB_NAME = os.environ['EVENT_HUB_NAME']
23-
STORAGE_CONNECTION_STR = os.environ["AZURE_STORAGE_CONN_STR"]
24+
STORAGE_ACCOUNT = "https://{}.blob.core.windows.net".format(
25+
os.environ["AZURE_STORAGE_ACCOUNT"])
2426
BLOB_CONTAINER_NAME = "your-blob-container-name" # Please make sure the blob container resource exists.
25-
STORAGE_SERVICE_API_VERSION = "2017-11-09"
27+
STORAGE_SERVICE_API_VERSION = "2019-02-02"
2628

2729

2830
def on_event(partition_context, event):
@@ -33,13 +35,15 @@ def on_event(partition_context, event):
3335

3436

3537
if __name__ == '__main__':
36-
checkpoint_store = BlobCheckpointStore.from_connection_string(
37-
STORAGE_CONNECTION_STR,
38+
checkpoint_store = BlobCheckpointStore(
39+
STORAGE_ACCOUNT,
3840
container_name=BLOB_CONTAINER_NAME,
39-
api_version=STORAGE_SERVICE_API_VERSION
41+
api_version=STORAGE_SERVICE_API_VERSION,
42+
credential=DefaultAzureCredential()
4043
)
41-
client = EventHubConsumerClient.from_connection_string(
42-
CONNECTION_STR,
44+
client = EventHubConsumerClient(
45+
FULLY_QUALIFIED_NAMESPACE,
46+
credential=DefaultAzureCredential(),
4347
consumer_group='$Default',
4448
eventhub_name=EVENTHUB_NAME,
4549
checkpoint_store=checkpoint_store

0 commit comments

Comments
 (0)