Tests for Create Bucket with internal methods added

cloudbridge/cloud/providers/azure/mock_azure_client.py

@@ -1,5 +1,5 @@
 from azure.mgmt.network.models import NetworkSecurityGroup
-from azure.storage.blob.models import Container
+from azure.storage.blob.models import Container, Blob
 from azure.mgmt.resource.resources.models import ResourceGroup
 from azure.mgmt.network.models import SecurityRule
 
@@ -38,12 +38,25 @@ class MockAzureClient:
 
     security_groups = [sec_gr1, sec_gr2, sec_gr3]
 
+
     container1 = Container()
     container1.name = "container1"
+
     container2 = Container()
     container2.name = "container2"
+
     containers = [container1, container2]
 
+    block1 = Blob()
+    block1.name = "block1"
+    block1.content = "blob1Content"
+
+    block2 = Blob()
+    block2.name = "block2"
+    block2.content = "blob2Content"
+
+    blocks = [block1, block2]
+
     rg = ResourceGroup(location='westus')
     rg.name = "testResourceGroup"
 
@@ -97,7 +110,49 @@ class MockAzureClient:
 
     def create_container(self, container_name):
         new_container = Container()
-        new_container.name = "newContainerCreate"
+        new_container.name = container_name
+        return new_container
+
+    def delete_container(self, container_name):
+        for cont in self.containers:
+            if cont.name == container_name:
+                self.containers.remove(cont)
+        return None
+
+    def create_blob_from_text(self, container_name, blob_name, text):
+        new_blob = Blob()
+        new_blob.name = blob_name
+        new_blob.content = text
+        self.blocks.append(new_blob)
+        return None
+
+    def get_blob(self, container_name, blob_name):
+        for blob in self.blocks:
+            if blob.name == blob_name:
+                return blob
+        return None
+
+    def list_blobs(self,container_name):
+        return self.blocks
+
+    def get_blob_content(self, container_name, blob_name):
+        for blob in self.blocks:
+            if blob.name == blob_name:
+                return blob
+        return None
+
+    def delete_blob(self,container_name, blob_name):
+        for blob in self.blocks:
+            if blob.name == blob_name:
+                self.blocks.remove(blob)
+
+    def create_blob_from_file(self, container_name, blob_name, file_path):
+        new_blob = Blob()
+        new_blob.name = blob_name
+        new_blob.content = "FileUploadText"
+        self.blocks.append(new_blob)
         return None
 
+    def get_blob_url(self, container_name, blob_name):
+        return 'https://cloudbridgeazure.blob.core.windows.net/vhds/block1'
 

test/test_azure_object_store_service.py

@@ -16,14 +16,13 @@ class AzureObjectStoreServiceTestCase(ProviderTestBase):
     @helpers.skipIfNoService(['object_store'])
     def test_azure_bucket_create(self):
         container = self.provider.object_store.create("container3")
-        print(container)
-        self.assertTrue(
-            container == None,
-            "Object create returned value should be None")
+        print("Create - " + str(container))
+        self.assertEqual(
+            str(container) , "<CB-AzureBucket: container3>")
 
     @helpers.skipIfNoService(['object_store'])
     def test_azure_bucket_find_Exist(self):
-        container = self.provider.object_store.find("container1")
+        container = self.provider.object_store.find("container2")
         print("Find Exist - " + str(container))
         self.assertEqual(
             len(container) ,1)
@@ -50,3 +49,116 @@ class AzureObjectStoreServiceTestCase(ProviderTestBase):
         print("Get Not Exist - " + str(container))
         self.assertEqual(
             str(container) , 'None')
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_delete(self):
+        containers = self.provider.object_store.find("container1")
+        cont = containers[0]
+        contDel = cont.delete()
+        print("Bucket delete - " + str(contDel))
+        self.assertEqual(
+            contDel, None)
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_create_object(self):
+        containers = self.provider.object_store.find("container1")
+        cont = containers[0]
+        contDel = cont.create_object("block1")
+        print("Create object  - " + str(contDel))
+        self.assertEqual(
+            str(contDel), '<CB-AzureBucketObject: block1>')
+
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_object_exists__internalE(self):
+        containers = self.provider.object_store.find("container2")
+        cont = containers[0]
+        contDel = cont.exists("block2")
+        print("List object  - " + str(contDel))
+        self.assertEqual(
+            str(contDel), 'True')
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_object_exists__internalNE(self):
+        containers = self.provider.object_store.find("container2")
+        cont = containers[0]
+        contDel = cont.exists("blob3")
+        print("List object  - " + str(contDel))
+        self.assertEqual(
+            str(contDel), 'False')
+
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_object_list(self):
+        containers = self.provider.object_store.find("container2")
+        cont = containers[0]
+        contDel = cont.list()
+        print("List object  - " + str(contDel))
+        self.assertEqual(
+            len(contDel), 1)
+
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_object_get(self):
+        containers = self.provider.object_store.find("container2")
+        cont = containers[0]
+        contDel = cont.get("block2")
+        print("List object  - " + str(contDel))
+        self.assertEqual(
+            str(contDel), "<CB-AzureBucketObject: block2>")
+
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_object_iter_content(self):
+        containers = self.provider.object_store.find("container2")
+        cont = containers[0]
+        blocks = cont.list()
+        block = blocks[0]
+        content = block.iter_content()
+        print("Iter content  - " + str(content))
+        self.assertEqual(
+            str(content),'blob2Content' )
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_object_upload(self):
+        containers = self.provider.object_store.find("container2")
+        cont = containers[0]
+        blocks = cont.list()
+        block = blocks[0]
+        block.upload('blob1Content')
+        self.assertEqual(
+            block.iter_content(), 'blob2Content')
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_object_delete(self):
+        containers = self.provider.object_store.find("container2")
+        cont = containers[0]
+        blocks = cont.list()
+        block = blocks[0]
+        block.delete()
+        self.assertEqual(
+            len(cont.list()), 1)
+
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_object_upload_from_file(self):
+        containers = self.provider.object_store.find("container2")
+        cont = containers[0]
+        blocks = cont.list()
+        block = blocks[0]
+        block.upload_from_file('blob2Content')
+        self.assertEqual(
+            block.iter_content(), 'blob2Content')
+
+
+    @helpers.skipIfNoService(['object_store'])
+    def test_azure_bucket_object_generate_url(self):
+        containers = self.provider.object_store.find("container2")
+        cont = containers[0]
+        blocks = cont.list()
+        block = blocks[0]
+        url = block.generate_url()
+        print(str(url))
+        self.assertEqual(
+            str(url), 'https://cloudbridgeazure.blob.core.windows.net/vhds/block1')
+