Smoke Test Sample for Track 2 libraries

Smoke Test for Indentity, Key Vault Secrets, Storage Blobs, Event Hubs and Cosmos DB

samples/smoketest/CosmosDB.py

@@ -0,0 +1,88 @@
+import azure.cosmos.cosmos_client as cosmos_client
+import azure.cosmos.errors as errors
+from azure.cosmos.partition_key import PartitionKey
+import os
+
+class CosmosDB:
+    def __init__(self):
+        URL = os.environ["COSMOS_ENDPOINT"]
+        KEY = os.environ["COSMOS_KEY"]
+        self.client = cosmos_client.CosmosClient(URL,{'masterKey': KEY})
+
+    def createDatabase(self):
+        dbName="pySolarSystem"
+        print("Creating '{0}' database...".format(dbName))
+        return self.client.create_database(dbName)
+
+    def createContainer(self, db):
+        collectionName = "Planets"
+        print("Creating '{0}' collection...".format(collectionName))
+        partition_key = PartitionKey(path='/id', kind='Hash')
+        return db.create_container(id="Planets", partition_key=partition_key)
+
+    def createDocuments(self, container):
+        planets = []
+
+        # Cosmos will look for an 'id' field in the items, if the 'id' is not specify Cosmos is going to assing a random key.
+        planets.append({
+            'id' : "Earth",
+            'HasRings' : False,
+            'Radius' : 3959,
+            'Moons' : 
+                [
+                    {
+                        'Name' : "Moon"
+                    }
+                ]
+                })
+
+        planets.append({
+                "id" : "Mars",
+                "HasRings" : False,
+                "Radius" : 2106,
+                "Moons" : 
+                    [
+                        {
+                            "Name" : "Phobos"
+                        },
+                        {
+                            "Name" : "Deimos"
+                        }
+                    ]
+            })
+
+        print("Inserting items in the collection...")
+        for planet in planets:
+            container.create_item(planet)
+            print("\t'{0}' created".format(planet['id']))
+        print("\tdone")
+
+    def deleteDatabase(self):
+        print("Cleaning up the resource...")
+        self.client.delete_database("pySolarSystem")
+        print("\tdone")
+
+    def Run(self):
+        print()
+        print("------------------------")
+        print("Cosmos DB")
+        print("------------------------")
+        print("1) Create a Database")
+        print("2) Create a Container in the database")
+        print("3) Insert Documents (items) into the Container")
+        print("4) Delete Database (Clean up the resource)")
+        print()
+
+        # Ensure that the database does not exists
+        try:
+            self.deleteDatabase()
+        except:
+            pass
+
+        try:
+            db = self.createDatabase()
+            container = self.createContainer(db=db)
+            self.createDocuments(container=container)            
+        finally:
+            # if something goes wrong, the resource should be cleaned anyway
+            self.deleteDatabase()

samples/smoketest/EventHubs.py

@@ -0,0 +1,52 @@
+from azure.eventhub import EventHubClient, EventData, EventPosition
+from datetime import datetime
+import os
+
+class EventHub:
+    def __init__(self):
+        # This test requires a previusly created Event Hub.
+        # In this example the name is "myeventhub", but it could be change below
+        connectionString = os.environ["EVENT_HUBS_CONNECTION_STRING"]
+        eventHubName = 'myeventhub'
+        self.client = EventHubClient.from_connection_string(connectionString, eventHubName)
+
+    def getPartitionIds(self):
+        print("Getting partitions id...")
+        partition_ids = self.client.get_partition_ids()
+        print("\tdone")
+        return partition_ids      
+
+    def sendAndReceiveEvents(self, partitionID):
+        consumer = self.client.create_consumer(consumer_group="$default", partition_id=partitionID, event_position=EventPosition(datetime.utcnow()))
+
+        print("Sending events...")
+        producer = self.client.create_producer(partition_id=partitionID)
+        event_list = [EventData(b"Test Event 1 in Python"),EventData(b"Test Event 2 in Python"),EventData(b"Test Event 3 in Python")]
+        producer.send(event_list)
+        producer.close()
+        print("\tdone")
+
+        print("Receiving events...")
+        received = consumer.receive(max_batch_size=len(event_list), timeout=2)
+        for event_data in received:
+            print("\tEvent Received: " + event_data.body_as_str())
+        consumer.close()
+        print("\tdone")
+
+        if(len(received) != len(event_list)):
+            raise Exception("Error, expecting {0} events, but {1} were received.".format(str(len(event_list)),str(len(received))))
+
+    def Run(self):
+        print()
+        print("------------------------")
+        print("Event Hubs")
+        print("------------------------")
+        print("1) Get partition ID")
+        print("2) Send Events")
+        print("3) Consume Events")
+        print()
+
+        partitionID = self.getPartitionIds()
+        #In this sample the same partition id is going to be used for the producer and consumer,
+        #It is the first one, but it could be any (is not relevant as long as it is the same in both producer and consumer)
+        self.sendAndReceiveEvents(partitionID[0])

samples/smoketest/KeyVaultSecrets.py

@@ -0,0 +1,43 @@
+from azure.identity import DefaultAzureCredential
+from azure.keyvault.secrets import SecretClient
+import os
+
+class KeyVault:
+    def __init__(self):
+        # DefaultAzureCredential() expects the following environment variables:
+        # * AZURE_CLIENT_ID
+        # * AZURE_CLIENT_SECRET
+        # * AZURE_TENANT_ID
+        credential = DefaultAzureCredential()
+        self.secret_client = SecretClient(vault_url=os.environ["AZURE_PROJECT_URL"], credential=credential)
+
+    def setSecret(self):
+        print("Setting a secret...")
+        self.secret_client.set_secret("secret-name", "secret-value")
+        print("\tdone")
+
+    def getSecret(self):
+        print("Getting a secret...")
+        secret = self.secret_client.get_secret("secret-name")
+        print("\tdone: " + secret.name)
+
+    def deleteSecret(self):
+        print("Deleting a secret...")
+        deleted_secret = self.secret_client.delete_secret("secret-name")
+        print("\tdone: " + deleted_secret.name)
+
+    def Run(self):
+        print()
+        print("------------------------")
+        print("Key Vault - Secrets\nIdentity - Credential")
+        print("------------------------")
+        print("1) Set a secret")
+        print("2) Get that secret")
+        print("3) Delete that secret (Clean up the resource)")
+        print()
+
+        try:
+            self.setSecret()
+            self.getSecret()
+        finally:
+            self.deleteSecret()

samples/smoketest/Program.py

@@ -0,0 +1,15 @@
+# from azure.identity import DefaultAzureCredential
+# from azure.keyvault.secrets import SecretClient
+from KeyVaultSecrets import KeyVault
+from StorageBlobs import StorageBlob
+from EventHubs import EventHub
+from CosmosDB import CosmosDB
+
+print("==========================================")
+print("      AZURE TRACK 2 SDKs SMOKE TEST")
+print("==========================================")
+
+KeyVault().Run()
+StorageBlob().Run()
+EventHub().Run()
+CosmosDB().Run()

samples/smoketest/StorageBlobs.py

@@ -0,0 +1,47 @@
+from azure.storage.blob import BlobClient
+import os
+
+class StorageBlob:
+    def __init__(self):
+        connectionString = os.environ["STORAGE_CONNECTION_STRING"]
+        self.blob = BlobClient.from_connection_string(connectionString, container="mycontainer", blob="pyTestBlob.txt")
+
+    def uploadBLob(self):
+        print("uploading blob...")
+        self.data = "This is a sample data for Python Test"
+        self.blob.upload_blob(self.data)
+        print("\tdone")
+
+    def downloadBlob(self):
+        print("downloading blob...")
+        with open("./downloadedBlob.txt", "wb+") as my_blob:
+            my_blob.writelines(self.blob.download_blob())
+
+        print("\tdone")
+
+    def deleteBlob(self):
+        print("Cleaning up the resource...")
+        self.blob.delete_blob()
+        print("\tdone")
+
+    def Run(self):
+        print()
+        print("------------------------")
+        print("Storage - Blob")
+        print("------------------------")
+        print("1) Upload a Blob")
+        print("2) Download a Blob")
+        print("3) Delete that Blob (Clean up the resource)")
+        print()
+
+        #Ensure that the blob does not exists before the tests
+        try:
+            self.deleteBlob()
+        except:
+            pass
+
+        try:
+            self.uploadBLob()
+            self.downloadBlob()
+        finally:
+            self.deleteBlob()