Repository: cloudstack Updated Branches: refs/heads/master 071004e3e -> 032bce5b5
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/032bce5b/test/integration/component/test_escalations_volumes.py ---------------------------------------------------------------------- diff --git a/test/integration/component/test_escalations_volumes.py b/test/integration/component/test_escalations_volumes.py index db4c3d8..0d78c58 100644 --- a/test/integration/component/test_escalations_volumes.py +++ b/test/integration/component/test_escalations_volumes.py @@ -15,18 +15,24 @@ # specific language governing permissions and limitations # under the License. -#Import Local Modules -from marvin.cloudstackTestCase import * -from marvin.cloudstackException import * -from marvin.cloudstackAPI import * -from marvin.sshClient import SshClient -from marvin.lib.utils import * -from marvin.lib.base import * -from marvin.lib.common import * -from marvin.lib.utils import checkVolumeSize -from marvin.codes import SUCCESS +# Import Local Modules +from marvin.cloudstackTestCase import cloudstackTestCase +from marvin.cloudstackAPI import createVolume, createTemplate +from marvin.lib.utils import (cleanup_resources, + random_gen) +from marvin.lib.base import (Account, + VirtualMachine, + ServiceOffering, + Volume, + DiskOffering, + Snapshot, + Template, + SnapshotPolicy) +from marvin.lib.common import (get_domain, + get_zone, + get_template) from nose.plugins.attrib import attr -from time import sleep + class TestVolumes(cloudstackTestCase): @@ -39,55 +45,42 @@ class TestVolumes(cloudstackTestCase): cls.services = cls.testClient.getParsedTestDataConfig() # Get Domain, Zone, Template cls.domain = get_domain(cls.api_client) - cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) + cls.zone = get_zone( + cls.api_client, + cls.testClient.getZoneForTests()) cls.template = get_template( - cls.api_client, - cls.zone.id, - cls.services["ostype"] - ) + cls.api_client, + cls.zone.id, + cls.services["ostype"] + ) if cls.zone.localstorageenabled: cls.storagetype = 'local' - cls.services["service_offerings"]["tiny"]["storagetype"] = 'local' + cls.services["service_offerings"][ + "tiny"]["storagetype"] = 'local' cls.services["disk_offering"]["storagetype"] = 'local' else: cls.storagetype = 'shared' - cls.services["service_offerings"]["tiny"]["storagetype"] = 'shared' + cls.services["service_offerings"][ + "tiny"]["storagetype"] = 'shared' cls.services["disk_offering"]["storagetype"] = 'shared' cls.services['mode'] = cls.zone.networktype - cls.services["virtual_machine"]["hypervisor"] = cls.testClient.getHypervisorInfo() + cls.services["virtual_machine"][ + "hypervisor"] = cls.testClient.getHypervisorInfo() cls.services["virtual_machine"]["zoneid"] = cls.zone.id cls.services["virtual_machine"]["template"] = cls.template.id cls.services["custom_volume"]["zoneid"] = cls.zone.id # Creating Disk offering, Service Offering and Account cls.disk_offering = DiskOffering.create( - cls.api_client, - cls.services["disk_offering"] - ) + cls.api_client, + cls.services["disk_offering"] + ) cls.service_offering = ServiceOffering.create( - cls.api_client, - cls.services["service_offerings"]["tiny"] - ) - cls.account = Account.create( - cls.api_client, - cls.services["account"], - domainid=cls.domain.id - ) - # Getting authentication for user in newly created Account - cls.user = cls.account.user[0] - cls.userapiclient = cls.testClient.getUserApiClient(cls.user.username, cls.domain.name) - # Creating Virtual Machine - cls.virtual_machine = VirtualMachine.create( - cls.userapiclient, - cls.services["virtual_machine"], - accountid=cls.account.name, - domainid=cls.account.domainid, - serviceofferingid=cls.service_offering.id, - ) - cls._cleanup.append(cls.virtual_machine) + cls.api_client, + cls.services["service_offerings"]["tiny"] + ) cls._cleanup.append(cls.disk_offering) cls._cleanup.append(cls.service_offering) - cls._cleanup.append(cls.account) except Exception as e: cls.tearDownClass() raise Exception("Warning: Exception in setup : %s" % e) @@ -96,10 +89,28 @@ class TestVolumes(cloudstackTestCase): def setUp(self): self.apiClient = self.testClient.getApiClient() - self.cleanup = [] + self.account = Account.create( + self.apiClient, + self.services["account"], + domainid=self.domain.id + ) + # Getting authentication for user in newly created Account + self.user = self.account.user[0] + self.userapiclient = self.testClient.getUserApiClient( + self.user.username, + self.domain.name) + # Creating Virtual Machine + self.virtual_machine = VirtualMachine.create( + self.userapiclient, + self.services["virtual_machine"], + accountid=self.account.name, + domainid=self.account.domainid, + serviceofferingid=self.service_offering.id, + ) + self.cleanup = [self.account, ] def tearDown(self): - #Clean up, terminate the created volumes + # Clean up, terminate the created volumes cleanup_resources(self.apiClient, self.cleanup) return @@ -110,15 +121,16 @@ class TestVolumes(cloudstackTestCase): except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) - def __verify_values(self, expected_vals, actual_vals): """ @summary: Function to verify expected and actual values Step1: Initializing return flag to True - Step1: Verifying length of expected and actual dictionaries is matching. + Step1: Verifying length of expected and actual dictionaries is + matching If not matching returning false Step2: Listing all the keys from expected dictionary - Step3: Looping through each key from step2 and verifying expected and actual dictionaries have same value + Step3: Looping through each key from step2 and verifying expected + and actual dictionaries have same value If not making return flag to False Step4: returning the return flag after all the values are verified """ @@ -135,24 +147,26 @@ class TestVolumes(cloudstackTestCase): return_flag = return_flag and True else: return_flag = return_flag and False - self.debug("expected Value: %s, is not matching with actual value: %s" % ( - exp_val, - act_val - )) + self.debug( + "expected Value: %s, is not matching with actual value:\ + %s" % + (exp_val, act_val)) return return_flag @attr(tags=["advanced", "basic"], required_hardware="true") def test_01_list_volumes_pagination(self): - """ + """ @summary: Test List Volumes pagination - + Step1: Listing all the volumes for a user Step2: Verifying listed volumes for account created at class level - Step3: If number of volumes is less than (page size + 1), then creating them + Step3: If number of volumes is less than (page size + 1), + then creating them Step4: Listing all the volumes again after creation of volumes Step5: Verifying the length of the volumes is (page size + 1) Step6: Listing all the volumes in page1 - Step7: Verifying that the length of the volumes in page 1 is (page size) + Step7: Verifying that the length of the volumes in page 1 is + (page size) Step8: Listing all the volumes in page2 Step9: Verifying that the length of the volumes in page 2 is 1 Step10: Deleting the volume present in page 2 @@ -160,119 +174,124 @@ class TestVolumes(cloudstackTestCase): Step12: Verifying that there are no volumes present in page 2 """ # Listing all the volumes for a user - list_volumes_before = Volume.list(self.userapiclient, listall=self.services["listall"]) + list_volumes_before = Volume.list( + self.userapiclient, + listall=self.services["listall"]) # Verifying listed volumes for account created at class level self.assertIsNotNone( - list_volumes_before, - "create volume from VM failed at class setup method" - ) + list_volumes_before, + "create volume from VM failed at class setup method" + ) self.assertEqual( - len(list_volumes_before), - 1, - "more than 1 volume created from VM at class level" - ) + len(list_volumes_before), + 1, + "more than 1 volume created from VM at class level" + ) - # If number of volumes is less than (pagesize + 1), then creating them + # If number of volumes is less than (pagesize + 1), then creating them for i in range(0, (self.services["pagesize"])): volume_created = Volume.create( - self.userapiclient, - self.services["volume"], - zoneid=self.zone.id, - diskofferingid=self.disk_offering.id - ) + self.userapiclient, + self.services["volume"], + zoneid=self.zone.id, + diskofferingid=self.disk_offering.id + ) self.assertIsNotNone( - volume_created, - "Volume is not created" - ) - if(i < (self.services["pagesize"] - 1)): - self.cleanup.append(volume_created) - + volume_created, + "Volume is not created" + ) + self.assertEqual( - self.services["volume"]["diskname"], - volume_created.name, - "Newly created volume name and the test data volume name are not matching" - ) + self.services["volume"]["diskname"], + volume_created.name, + "Newly created volume name and the test data\ + volume name are not matching" + ) - # Listing all the volumes again after creation of volumes - list_volumes_after = Volume.list(self.userapiclient, listall=self.services["listall"]) + # Listing all the volumes again after creation of volumes + list_volumes_after = Volume.list( + self.userapiclient, + listall=self.services["listall"]) # Verifying the length of the volumes is (page size + 1) self.assertEqual( - len(list_volumes_after), - (self.services["pagesize"] + 1), - "Number of volumes created is not matching expected" - ) + len(list_volumes_after), + (self.services["pagesize"] + 1), + "Number of volumes created is not matching expected" + ) # Listing all the volumes in page1 list_volumes_page1 = Volume.list( - self.userapiclient, - listall=self.services["listall"], - page=1, - pagesize=self.services["pagesize"] - ) + self.userapiclient, + listall=self.services["listall"], + page=1, + pagesize=self.services["pagesize"] + ) self.assertIsNotNone( - list_volumes_page1, - "No volumes found in Page 1" - ) + list_volumes_page1, + "No volumes found in Page 1" + ) # Verifying that the length of the volumes in page 1 is (page size) self.assertEqual( - len(list_volumes_page1), - self.services["pagesize"], - "List Volume response is not matching with the page size length for page 1" - ) + len(list_volumes_page1), + self.services["pagesize"], + "List Volume response is not matching with\ + the page size length for page 1" + ) # Listing all the volumes in page2 list_volumes_page2 = Volume.list( - self.userapiclient, - listall=self.services["listall"], - page=2, - pagesize=self.services["pagesize"] - ) + self.userapiclient, + listall=self.services["listall"], + page=2, + pagesize=self.services["pagesize"] + ) self.assertIsNotNone( - list_volumes_page2, - "No volumes found in Page 2" - ) + list_volumes_page2, + "No volumes found in Page 2" + ) # Verifying that the length of the volumes in page 2 is 1 self.assertEqual( - len(list_volumes_page2), - 1, - "List Volume response is not matching with the page size length for page 2" - ) + len(list_volumes_page2), + 1, + "List Volume response is not matching with\ + the page size length for page 2" + ) volume_page2 = list_volumes_page2[0] # Verifying that the volume on page 2 is not present in page1 for i in range(0, len(list_volumes_page1)): volume_page1 = list_volumes_page1[i] self.assertNotEquals( - volume_page2.id, - volume_page1.id, - "Volume listed in page 2 is also listed in page 1" - ) + volume_page2.id, + volume_page1.id, + "Volume listed in page 2 is also listed in page 1" + ) # Deleting a single volume Volume.delete(volume_created, self.userapiclient) # Listing the volumes in page 2 list_volume_response = Volume.list( - self.userapiclient, - listall=self.services["listall"], - page=2, - pagesize=self.services["pagesize"] - ) + self.userapiclient, + listall=self.services["listall"], + page=2, + pagesize=self.services["pagesize"] + ) # verifying that volume does not exists on page 2 self.assertEqual( - list_volume_response, - None, - "Volume was not deleted" - ) + list_volume_response, + None, + "Volume was not deleted" + ) return - @attr(tags=["advanced", "basic"], required_hardware="true") + @attr(tags=["advanced", "basic"], required_hardware="true") def test_02_list_volume_byid(self): - """ + """ @summary: Test List Volumes with Id - + Step1: Listing all the volumes for a user before creating a data volume Step2: Verifying the length of the list as 1 Step3: Creating a data volume @@ -285,157 +304,156 @@ class TestVolumes(cloudstackTestCase): """ # Listing all the volumes for a user before creating a data volume list_volumes_before = Volume.list( - self.userapiclient, - listall=self.services["listall"] - ) + self.userapiclient, + listall=self.services["listall"] + ) self.assertIsNotNone( - list_volumes_before, - "create volume from VM failed at class setup method") + list_volumes_before, + "create volume from VM failed at class setup method") # Verifying the length of the list as 1 self.assertEqual( - len(list_volumes_before), - 1, - "more than 1 volume created at class level" - ) + len(list_volumes_before), + 1, + "more than 1 volume created at class level" + ) root_volume = list_volumes_before[0] - + # Creating a data volume volume_created = Volume.create( - self.userapiclient, - self.services["volume"], - zoneid=self.zone.id, - diskofferingid=self.disk_offering.id - ) + self.userapiclient, + self.services["volume"], + zoneid=self.zone.id, + diskofferingid=self.disk_offering.id + ) self.assertIsNotNone( - volume_created, - "Volume is not created" - ) - self.cleanup.append(volume_created) - + volume_created, + "Volume is not created" + ) self.assertEqual( - self.services["volume"]["diskname"], - volume_created.name, - "Newly created volume name and the test data volume name are not matching" - ) + self.services["volume"]["diskname"], + volume_created.name, + "Newly created volume name and\ + the test data volume name are not matching" + ) # Listing all the volumes for a user after creating a data volume list_volumes_after = Volume.list( - self.userapiclient, - listall=self.services["listall"] - ) + self.userapiclient, + listall=self.services["listall"] + ) self.assertIsNotNone( - list_volumes_after, - "Volume creation failed" - ) + list_volumes_after, + "Volume creation failed" + ) # Verifying the list volume size is increased by 1 self.assertEqual( - len(list_volumes_before) + 1, - len(list_volumes_after), - "list volume is not matching with Number of volumes created" - ) - + len(list_volumes_before) + 1, + len(list_volumes_after), + "list volume is not matching with Number of volumes created" + ) + # Listing a Root Volume by Id and verifying the volume details list_volumes_by_id = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=root_volume.id - ) + self.userapiclient, + listall=self.services["listall"], + id=root_volume.id + ) self.assertIsNotNone( - list_volumes_by_id, - "Root volume is not listed" - ) + list_volumes_by_id, + "Root volume is not listed" + ) self.assertEqual( - 1, - len(list_volumes_by_id), - "list volume is not matching with Number of volumes created" - ) + 1, + len(list_volumes_by_id), + "list volume is not matching with Number of volumes created" + ) obtained_volume = list_volumes_by_id[0] - - #Creating expected and actual values dictionaries + + # Creating expected and actual values dictionaries expected_dict = { - "id":root_volume.id, - "name":root_volume.name, - "vmname":self.virtual_machine.name, - "state":"Ready", - "type":"ROOT", - "zoneid":self.zone.id, - "account":self.account.name, - "storagetype":self.storagetype, - "size":self.template.size - } + "id": root_volume.id, + "name": root_volume.name, + "vmname": self.virtual_machine.name, + "state": "Ready", + "type": "ROOT", + "zoneid": self.zone.id, + "account": self.account.name, + "storagetype": self.storagetype, + "size": self.template.size + } actual_dict = { - "id":obtained_volume.id, - "name":obtained_volume.name, - "vmname":obtained_volume.vmname, - "state":obtained_volume.state, - "type":obtained_volume.type, - "zoneid":obtained_volume.zoneid, - "account":obtained_volume.account, - "storagetype":obtained_volume.storagetype, - "size":obtained_volume.size, - } + "id": obtained_volume.id, + "name": obtained_volume.name, + "vmname": obtained_volume.vmname, + "state": obtained_volume.state, + "type": obtained_volume.type, + "zoneid": obtained_volume.zoneid, + "account": obtained_volume.account, + "storagetype": obtained_volume.storagetype, + "size": obtained_volume.size, + } root_volume_status = self.__verify_values( - expected_dict, - actual_dict - ) + expected_dict, + actual_dict + ) self.assertEqual( - True, - root_volume_status, - "Listed Root Volume details are not as expected" - ) + True, + root_volume_status, + "Listed Root Volume details are not as expected" + ) # Listing a Data Volume by Id and verifying the volume details list_volumes_by_id = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) self.assertIsNotNone( - list_volumes_by_id, - "Data volume is not listed" - ) + list_volumes_by_id, + "Data volume is not listed" + ) self.assertEqual( - len(list_volumes_by_id), - 1, - "list volume is not matching with Number of volumes created" - ) + len(list_volumes_by_id), + 1, + "list volume is not matching with Number of volumes created" + ) obtained_volume = list_volumes_by_id[0] - - #Creating expected and actual values dictionaries + + # Creating expected and actual values dictionaries expected_dict = { - "id":volume_created.id, - "name":volume_created.name, - "state":"Allocated", - "type":"DATADISK", - "zoneid":self.zone.id, - "account":self.account.name, - "storagetype":self.storagetype, - "size":self.disk_offering.disksize - } + "id": volume_created.id, + "name": volume_created.name, + "state": "Allocated", + "type": "DATADISK", + "zoneid": self.zone.id, + "account": self.account.name, + "storagetype": self.storagetype, + "size": self.disk_offering.disksize + } actual_dict = { - "id":obtained_volume.id, - "name":obtained_volume.name, - "state":obtained_volume.state, - "type":obtained_volume.type, - "zoneid":obtained_volume.zoneid, - "account":obtained_volume.account, - "storagetype":obtained_volume.storagetype, - "size":obtained_volume.size/(1024*1024*1024), - } + "id": obtained_volume.id, + "name": obtained_volume.name, + "state": obtained_volume.state, + "type": obtained_volume.type, + "zoneid": obtained_volume.zoneid, + "account": obtained_volume.account, + "storagetype": obtained_volume.storagetype, + "size": obtained_volume.size / (1024 * 1024 * 1024), + } root_volume_status = self.__verify_values( - expected_dict, - actual_dict - ) + expected_dict, + actual_dict + ) self.assertEqual( - True, - root_volume_status, - "Listed Data Volume details are not as expected" - ) + True, + root_volume_status, + "Listed Data Volume details are not as expected" + ) return - + @attr(tags=["advanced", "basic"], required_hardware="true") def test_03_data_volume_resize(self): - """ + """ @summary: Test to verify creation and resize of data volume - + Step1: Listing the volumes for a user before creating data volume Step2: Creating a data volume Step3: Listing the volumes for a user after creating data volume @@ -445,111 +463,116 @@ class TestVolumes(cloudstackTestCase): Step6: Resizing data volume """ # Listing volumes for a user before creating a volume - list_volumes_before = Volume.list(self.userapiclient, listall=self.services["listall"]) - + list_volumes_before = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + # Creating a data volume volume_created = Volume.create( - self.userapiclient, - self.services["volume"], - zoneid=self.zone.id, - diskofferingid=self.disk_offering.id - ) + self.userapiclient, + self.services["volume"], + zoneid=self.zone.id, + diskofferingid=self.disk_offering.id + ) self.assertIsNotNone(volume_created, "Data volume creation failed") - - self.cleanup.append(volume_created) - + # Listing volumes for a user after creating data volume - list_volumes_after = Volume.list(self.userapiclient, listall=self.services["listall"]) + list_volumes_after = Volume.list( + self.userapiclient, + listall=self.services["listall"]) self.assertEquals( - len(list_volumes_before) + 1, - len(list_volumes_after), - "Data volume creation failed" - ) - + len(list_volumes_before) + 1, + len(list_volumes_after), + "Data volume creation failed" + ) + # Attaching data volume created to Virtual Machine self.virtual_machine.attach_volume( - self.userapiclient, - volume_created - ) - + self.userapiclient, + volume_created + ) + list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) attached_volume = list_volumes[0] - + self.assertIsNotNone( - attached_volume.vmname, - "VM is not attached to Volume" - ) + attached_volume.vmname, + "VM is not attached to Volume" + ) self.assertEquals( - self.virtual_machine.name, - attached_volume.vmname, - "VM Name is not matching with attached vm" - ) - + self.virtual_machine.name, + attached_volume.vmname, + "VM Name is not matching with attached vm" + ) + # Detaching data volume from Virtual Machine self.virtual_machine.detach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) detached_volume = list_volumes[0] self.assertIsNone( - detached_volume.vmname, - "VM is not detached from volume" - ) - - # Verifying if there exists a disk offering with higher size. If not present creating it + detached_volume.vmname, + "VM is not detached from volume" + ) + + # Verifying if there exists a disk offering with higher size. If not + # present creating it list_disk_offerings = DiskOffering.list(self.apiClient) - + large_disk_offering_exists = False # Converting disk_size in bytes to GB - current_disk_size = volume_created.size/(1024*1024*1024) - + current_disk_size = volume_created.size / (1024 * 1024 * 1024) + for disk_offering in list_disk_offerings: - if ((disk_offering.disksize > current_disk_size) and (not disk_offering.iscustomized) and disk_offering.storagetype == self.storagetype): + if ((disk_offering.disksize > current_disk_size) and ( + not disk_offering.iscustomized) and + disk_offering.storagetype == self.storagetype): new_disk_offering = disk_offering large_disk_offering_exists = True break - - if large_disk_offering_exists == False: - new_size = (volume_created.size/(1024*1024*1024)) + 1 + + if not large_disk_offering_exists: + new_size = (volume_created.size / (1024 * 1024 * 1024)) + 1 self.services["disk_offering"]["disksize"] = new_size new_disk_offering = DiskOffering.create( - self.apiClient, - self.services["disk_offering"] - ) + self.apiClient, + self.services["disk_offering"] + ) if new_disk_offering is not None: self.cleanup.append(new_disk_offering) else: new_size = new_disk_offering.disksize - + # Resizing data volume resized_volume = volume_created.resize( - self.userapiclient, - diskofferingid=new_disk_offering.id, - shrinkok='false', - ) + self.userapiclient, + diskofferingid=new_disk_offering.id, + shrinkok='false', + ) self.assertIsNotNone(resized_volume, "Resize Volume failed") # Verifying data volume size is increased self.assertEquals( - new_size, - (resized_volume.size/(1024*1024*1024)), - "volume not resized to expected value" - ) + new_size, + (resized_volume.size / (1024 * 1024 * 1024)), + "volume not resized to expected value" + ) return - + @attr(tags=["advanced", "basic"], required_hardware="true") def test_04_custom_volume_resize(self): - """ + """ @summary: Test to verify creation and resize of custom volume - + Step1: Checking if Custom disk offering already exists. If not present then creating custom Disk Offering Step2: Listing the volumes for a user before creating custom volume @@ -560,261 +583,271 @@ class TestVolumes(cloudstackTestCase): """ # Listing all the disk offerings list_disk_offerings = DiskOffering.list(self.apiClient) - + custom_disk_offering_exists = False - + # Verifying if a custom disk offering already exists if list_disk_offerings is not None: for disk_offering in list_disk_offerings: - if (disk_offering.iscustomized and disk_offering.storagetype == self.storagetype): + if (disk_offering.iscustomized and disk_offering.storagetype == + self.storagetype): custom_disk_offering = disk_offering custom_disk_offering_exists = True break - - # If a custom disk offering does not exists, then creating a custom disk offering - if custom_disk_offering_exists == False: + + # If a custom disk offering does not exists, then creating a custom + # disk offering + if not custom_disk_offering_exists: custom_disk_offering = DiskOffering.create( - self.apiClient, - self.services["disk_offering"], - custom=True - ) + self.apiClient, + self.services["disk_offering"], + custom=True + ) if custom_disk_offering is not None: self.cleanup.append(custom_disk_offering) - + # Listing the volumes for a user before creating custom volume - list_volumes_before = Volume.list(self.userapiclient, listall=self.services["listall"]) - + list_volumes_before = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + # Creating a custom volume volume_created = Volume.create_custom_disk( - self.userapiclient, - self.services["custom_volume"], - account=self.account.name, - domainid=self.account.domainid, - diskofferingid=custom_disk_offering.id - ) + self.userapiclient, + self.services["custom_volume"], + account=self.account.name, + domainid=self.account.domainid, + diskofferingid=custom_disk_offering.id + ) self.assertIsNotNone( - volume_created, - "Custom volume did not get created" - ) - - self.cleanup.append(volume_created) - + volume_created, + "Custom volume did not get created" + ) + # Listing the volumes for a user after creating custom volume - list_volumes_after = Volume.list(self.userapiclient, listall=self.services["listall"]) - - # Verifyign that volume list is increased by 1 after creation of custion volume + list_volumes_after = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + + # Verifyign that volume list is increased by 1 after creation of + # custion volume self.assertEquals( - len(list_volumes_before) + 1, - len(list_volumes_after), - "Custom volume did not get created" - ) - + len(list_volumes_before) + 1, + len(list_volumes_after), + "Custom volume did not get created" + ) + # Attaching custom volume created to Virtual Machine self.virtual_machine.attach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) attached_volume = list_volumes[0] - + self.assertIsNotNone( - attached_volume.vmname, - "VM is not attached to Volume" - ) + attached_volume.vmname, + "VM is not attached to Volume" + ) self.assertEquals( - self.virtual_machine.name, - attached_volume.vmname, - "VM Name is not matching with attached vm" - ) - + self.virtual_machine.name, + attached_volume.vmname, + "VM Name is not matching with attached vm" + ) + # Detaching custom volume from Virtual Machine self.virtual_machine.detach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) detached_volume = list_volumes[0] self.assertIsNone( - detached_volume.vmname, - "VM is not detached from volume" - ) - + detached_volume.vmname, + "VM is not detached from volume" + ) + # Resizing custom volume # Increasing custom disk size by 1 new_size = self.services["custom_volume"]["customdisksize"] + 1 resized_volume = volume_created.resize( - self.userapiclient, - diskofferingid=custom_disk_offering.id, - shrinkok='false', - size=new_size) + self.userapiclient, + diskofferingid=custom_disk_offering.id, + shrinkok='false', + size=new_size) self.assertIsNotNone(resized_volume, "Resize Volume failed") # Verifying that custom disk size is increased self.assertEquals( - new_size, - (resized_volume.size/(1024*1024*1024)), - "volume not resized to expected value" - ) + new_size, + (resized_volume.size / (1024 * 1024 * 1024)), + "volume not resized to expected value" + ) return - + @attr(tags=["advanced", "basic"], required_hardware="true") def test_05_volume_snapshot(self): - """ - @summary: Test to verify creation of snapshot from volume and creation of template, volume from snapshot - + """ + @summary: Test to verify creation of snapshot from volume + and creation of template, volume from snapshot + Step1: Creating a volume Step2: Attaching and Detaching custom volume created to Virtual Machine Step3: Creating Snapshot from volume Step4: Creating Volume from snapshot Step5: Creating Template from Snapshot """ - list_volumes_before = Volume.list(self.userapiclient, listall=self.services["listall"]) - + list_volumes_before = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + volume_created = Volume.create( - self.userapiclient, - self.services["volume"], - zoneid=self.zone.id, - diskofferingid=self.disk_offering.id - ) - + self.userapiclient, + self.services["volume"], + zoneid=self.zone.id, + diskofferingid=self.disk_offering.id + ) + self.assertIsNotNone(volume_created, "Volume not created") - - if volume_created is not None: - self.cleanup.append(volume_created) - - list_volumes_after = Volume.list(self.userapiclient, listall=self.services["listall"]) - + + list_volumes_after = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + self.assertEquals( - len(list_volumes_before) + 1, - len(list_volumes_after), - "Volume not created" - ) + len(list_volumes_before) + 1, + len(list_volumes_after), + "Volume not created" + ) # Attaching and Detaching custom volume created to Virtual Machine self.virtual_machine.attach_volume( - self.userapiclient, - volume_created - ) - + self.userapiclient, + volume_created + ) + list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) attached_volume = list_volumes[0] - + self.assertIsNotNone( - attached_volume.vmname, - "VM is not attached to Volume" - ) + attached_volume.vmname, + "VM is not attached to Volume" + ) self.assertEquals( - self.virtual_machine.name, - attached_volume.vmname, - "VM Name is not matching with attached vm" - ) + self.virtual_machine.name, + attached_volume.vmname, + "VM Name is not matching with attached vm" + ) self.virtual_machine.detach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) detached_volume = list_volumes[0] self.assertIsNone( - detached_volume.vmname, - "VM is not detached from volume" - ) + detached_volume.vmname, + "VM is not detached from volume" + ) # Creating Snapshot from volume snapshot_created = Snapshot.create( - self.userapiclient, - volume_created.id, - ) - + self.userapiclient, + volume_created.id, + ) + self.assertIsNotNone(snapshot_created, "Snapshot not created") - + self.cleanup.append(snapshot_created) - - #Creating expected and actual values dictionaries + + # Creating expected and actual values dictionaries expected_dict = { - "id":volume_created.id, - "intervaltype":"MANUAL", - "snapshottype":"MANUAL", - "volumetype":volume_created.type, - "domain":self.domain.id - } + "id": volume_created.id, + "intervaltype": "MANUAL", + "snapshottype": "MANUAL", + "volumetype": volume_created.type, + "domain": self.domain.id + } actual_dict = { - "id":snapshot_created.volumeid, - "intervaltype":snapshot_created.intervaltype, - "snapshottype":snapshot_created.snapshottype, - "volumetype":snapshot_created.volumetype, - "domain":snapshot_created.domainid, - } + "id": snapshot_created.volumeid, + "intervaltype": snapshot_created.intervaltype, + "snapshottype": snapshot_created.snapshottype, + "volumetype": snapshot_created.volumetype, + "domain": snapshot_created.domainid, + } status = self.__verify_values( - expected_dict, - actual_dict - ) + expected_dict, + actual_dict + ) self.assertEqual( - True, - status, - "Snapshot created from Volume details are not as expected" - ) + True, + status, + "Snapshot created from Volume details are not as expected" + ) # Creating Volume from snapshot cmd = createVolume.createVolumeCmd() - cmd.name = "-".join([self.services["volume"]["diskname"], random_gen()]) + cmd.name = "-".join([self.services["volume"] + ["diskname"], random_gen()]) cmd.snapshotid = snapshot_created.id - - volume_from_snapshot = Volume(self.userapiclient.createVolume(cmd).__dict__) - + + volume_from_snapshot = Volume( + self.userapiclient.createVolume(cmd).__dict__) + self.assertIsNotNone( - volume_from_snapshot, - "Volume creation failed from snapshot" - ) - self.cleanup.append(volume_from_snapshot) - - #Creating expected and actual values dictionaries + volume_from_snapshot, + "Volume creation failed from snapshot" + ) + + # Creating expected and actual values dictionaries expected_dict = { - "snapshotid":snapshot_created.id, - "volumetype":snapshot_created.volumetype, - "size":self.disk_offering.disksize, - "accounr":self.account.name, - "domain":self.domain.id, - "storagetype":self.storagetype, - "zone":self.zone.id - } + "snapshotid": snapshot_created.id, + "volumetype": snapshot_created.volumetype, + "size": self.disk_offering.disksize, + "accounr": self.account.name, + "domain": self.domain.id, + "storagetype": self.storagetype, + "zone": self.zone.id + } actual_dict = { - "snapshotid":volume_from_snapshot.snapshotid, - "volumetype":volume_from_snapshot.type, - "size":volume_from_snapshot.size/(1024*1024*1024), - "accounr":volume_from_snapshot.account, - "domain":volume_from_snapshot.domainid, - "storagetype":volume_from_snapshot.storagetype, - "zone":volume_from_snapshot.zoneid, - } + "snapshotid": volume_from_snapshot.snapshotid, + "volumetype": volume_from_snapshot.type, + "size": volume_from_snapshot.size / (1024 * 1024 * 1024), + "accounr": volume_from_snapshot.account, + "domain": volume_from_snapshot.domainid, + "storagetype": volume_from_snapshot.storagetype, + "zone": volume_from_snapshot.zoneid, + } status = self.__verify_values( - expected_dict, - actual_dict - ) + expected_dict, + actual_dict + ) self.assertEqual( - True, - status, - "Volume created from Snapshot details are not as expected" - ) + True, + status, + "Volume created from Snapshot details are not as expected" + ) # Creating Template from Snapshot - list_templates_before = Template.list(self.userapiclient, templatefilter='self') - + list_templates_before = Template.list( + self.userapiclient, + templatefilter='self') + if list_templates_before is None: templates_before_size = 0 else: templates_before_size = len(list_templates_before) - + cmd = createTemplate.createTemplateCmd() cmd.name = self.services["ostype"] cmd.displaytext = self.services["ostype"] @@ -822,63 +855,67 @@ class TestVolumes(cloudstackTestCase): cmd.snapshotid = snapshot_created.id cmd.ispublic = False cmd.passwordenabled = False - - template_from_snapshot = Template(self.userapiclient.createTemplate(cmd).__dict__) - + + template_from_snapshot = Template( + self.userapiclient.createTemplate(cmd).__dict__) + self.assertIsNotNone( - template_from_snapshot, - "Template creation failed from snapshot" - ) - + template_from_snapshot, + "Template creation failed from snapshot" + ) + self.cleanup.append(template_from_snapshot) - - #Creating expected and actual values dictionaries + + # Creating expected and actual values dictionaries expected_dict = { - "name":self.services["ostype"], - "ostypeid":self.template.ostypeid, - "type":"USER", - "zone":self.zone.id, - "domain":self.domain.id, - "account":self.account.name, - "passwordenabled":False, - "ispublic":False, - "size":self.disk_offering.disksize - } + "name": self.services["ostype"], + "ostypeid": self.template.ostypeid, + "type": "USER", + "zone": self.zone.id, + "domain": self.domain.id, + "account": self.account.name, + "passwordenabled": False, + "ispublic": False, + "size": self.disk_offering.disksize + } actual_dict = { - "name":template_from_snapshot.name, - "ostypeid":template_from_snapshot.ostypeid, - "type":template_from_snapshot.templatetype, - "zone":template_from_snapshot.zoneid, - "domain":template_from_snapshot.domainid, - "account":template_from_snapshot.account, - "passwordenabled":template_from_snapshot.passwordenabled, - "ispublic":template_from_snapshot.ispublic, - "size":template_from_snapshot.size/(1024*1024*1024) - } + "name": template_from_snapshot.name, + "ostypeid": template_from_snapshot.ostypeid, + "type": template_from_snapshot.templatetype, + "zone": template_from_snapshot.zoneid, + "domain": template_from_snapshot.domainid, + "account": template_from_snapshot.account, + "passwordenabled": template_from_snapshot.passwordenabled, + "ispublic": template_from_snapshot.ispublic, + "size": template_from_snapshot.size / (1024 * 1024 * 1024) + } status = self.__verify_values( - expected_dict, - actual_dict - ) + expected_dict, + actual_dict + ) self.assertEqual( - True, - status, - "Template created from Snapshot details are not as expected" - ) - - list_templates_after = Template.list(self.userapiclient, templatefilter='self') - + True, + status, + "Template created from Snapshot details are not as expected" + ) + + list_templates_after = Template.list( + self.userapiclient, + templatefilter='self') + self.assertEquals( - templates_before_size + 1, - len(list_templates_after), - "Template creation failed from snapshot" - ) + templates_before_size + 1, + len(list_templates_after), + "Template creation failed from snapshot" + ) return - + @attr(tags=["advanced", "basic"], required_hardware="true") def test_06_volume_snapshot_policy_hourly(self): - """ - @summary: Test to verify creation of Hourly Snapshot policies from volume - + """ + @summary: Test to verify creation of Hourly Snapshot policies + from volume + Step1: Creating a Volume. Step2: Attaching volume created in Step2 to virtual machine Step3: Detaching the volume created in step2 from virtual machine @@ -887,122 +924,130 @@ class TestVolumes(cloudstackTestCase): Step6: Listing snapshot policies for a volume created in step1 again Step7: Verifyign that the list snapshot policy length is increased by 1 """ - list_volumes_before = Volume.list(self.userapiclient, listall=self.services["listall"]) - + list_volumes_before = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + volume_created = Volume.create( - self.userapiclient, - self.services["volume"], - zoneid=self.zone.id, - diskofferingid=self.disk_offering.id - ) - + self.userapiclient, + self.services["volume"], + zoneid=self.zone.id, + diskofferingid=self.disk_offering.id + ) + self.assertIsNotNone(volume_created, "Volume not created") - self.cleanup.append(volume_created) - - list_volumes_after = Volume.list(self.userapiclient, listall=self.services["listall"]) - + + list_volumes_after = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + self.assertEquals( - len(list_volumes_before) + 1, - len(list_volumes_after), - "Volume not created" - ) - + len(list_volumes_before) + 1, + len(list_volumes_after), + "Volume not created" + ) + # Attaching volume created to Virtual Machine self.virtual_machine.attach_volume( - self.userapiclient, - volume_created - ) - + self.userapiclient, + volume_created + ) + list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) attached_volume = list_volumes[0] - + self.assertIsNotNone( - attached_volume.vmname, - "VM is not attached to Volume" - ) + attached_volume.vmname, + "VM is not attached to Volume" + ) self.assertEquals( - self.virtual_machine.name, - attached_volume.vmname, - "VM Name is not matching with attached vm" - ) + self.virtual_machine.name, + attached_volume.vmname, + "VM Name is not matching with attached vm" + ) # Detaching volume created from Virtual Machine self.virtual_machine.detach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) detached_volume = list_volumes[0] self.assertIsNone( - detached_volume.vmname, - "VM is not detached from volume" - ) + detached_volume.vmname, + "VM is not detached from volume" + ) # Creating Hourly Snapshot Policy from volume self.services["recurring_snapshot"]["intervaltype"] = 'hourly' self.services["recurring_snapshot"]["schedule"] = '1' - - list_snapshot_policy_before = SnapshotPolicy.list(self.userapiclient, volumeid=volume_created.id) - + + list_snapshot_policy_before = SnapshotPolicy.list( + self.userapiclient, + volumeid=volume_created.id) + snapshot_policy_before_size = 0 - + if list_snapshot_policy_before is not None: snapshot_policy_before_size = len(list_snapshot_policy_before) - + snapshot_policy_hourly = SnapshotPolicy.create( - self.userapiclient, - volume_created.id, - self.services["recurring_snapshot"] - ) + self.userapiclient, + volume_created.id, + self.services["recurring_snapshot"] + ) self.assertIsNotNone( - snapshot_policy_hourly, - "Hourly Snapshot policy creation failed" - ) - #Creating expected and actual values dictionaries + snapshot_policy_hourly, + "Hourly Snapshot policy creation failed" + ) + # Creating expected and actual values dictionaries expected_dict = { - "schedule":self.services["recurring_snapshot"]["schedule"], - "intervaltype":0, - "volumeid":volume_created.id - } + "schedule": self.services["recurring_snapshot"]["schedule"], + "intervaltype": 0, + "volumeid": volume_created.id + } actual_dict = { - "schedule":snapshot_policy_hourly.schedule, - "intervaltype":snapshot_policy_hourly.intervaltype, - "volumeid":snapshot_policy_hourly.volumeid - } + "schedule": snapshot_policy_hourly.schedule, + "intervaltype": snapshot_policy_hourly.intervaltype, + "volumeid": snapshot_policy_hourly.volumeid + } status = self.__verify_values( - expected_dict, - actual_dict - ) + expected_dict, + actual_dict + ) self.assertEqual( - True, - status, - "Hourly Snapshot Policy details are not as expected" - ) - - list_snapshot_policy_after = SnapshotPolicy.list(self.userapiclient, volumeid=volume_created.id) - + True, + status, + "Hourly Snapshot Policy details are not as expected" + ) + + list_snapshot_policy_after = SnapshotPolicy.list( + self.userapiclient, + volumeid=volume_created.id) + self.assertIsNotNone( - list_snapshot_policy_after, - "Hourly Snapshot policy creation failed" - ) + list_snapshot_policy_after, + "Hourly Snapshot policy creation failed" + ) self.assertEquals( - snapshot_policy_before_size + 1, - len(list_snapshot_policy_after), - "Hourly Snapshot policy creation failed" - ) + snapshot_policy_before_size + 1, + len(list_snapshot_policy_after), + "Hourly Snapshot policy creation failed" + ) return - + @attr(tags=["advanced", "basic"], required_hardware="true") def test_07_volume_snapshot_policy_daily(self): - """ - @summary: Test to verify creation of Daily Snapshot policies from volume - + """ + @summary: Test to verify creation of Daily Snapshot policies + from volume + Step1: Creating a Volume. Step2: Attaching volume created in Step2 to virtual machine Step3: Detaching the volume created in step2 from virtual machine @@ -1011,121 +1056,129 @@ class TestVolumes(cloudstackTestCase): Step6: Listing snapshot policies for a volume created in step1 again Step7: Verifyign that the list snapshot policy length is increased by 1 """ - list_volumes_before = Volume.list(self.userapiclient, listall=self.services["listall"]) - + list_volumes_before = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + volume_created = Volume.create( - self.userapiclient, - self.services["volume"], - zoneid=self.zone.id, - diskofferingid=self.disk_offering.id - ) - + self.userapiclient, + self.services["volume"], + zoneid=self.zone.id, + diskofferingid=self.disk_offering.id + ) + self.assertIsNotNone(volume_created, "Volume not created") - self.cleanup.append(volume_created) - - list_volumes_after = Volume.list(self.userapiclient, listall=self.services["listall"]) - + + list_volumes_after = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + self.assertEquals( - len(list_volumes_before) + 1, - len(list_volumes_after), - "Volume not created" - ) + len(list_volumes_before) + 1, + len(list_volumes_after), + "Volume not created" + ) # Attaching volume created to Virtual Machine self.virtual_machine.attach_volume( - self.userapiclient, - volume_created - ) - + self.userapiclient, + volume_created + ) + list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) attached_volume = list_volumes[0] - + self.assertIsNotNone( - attached_volume.vmname, - "VM is not attached to Volume" - ) + attached_volume.vmname, + "VM is not attached to Volume" + ) self.assertEquals( - self.virtual_machine.name, - attached_volume.vmname, - "VM Name is not matching with attached vm" - ) + self.virtual_machine.name, + attached_volume.vmname, + "VM Name is not matching with attached vm" + ) # Detaching volume created from Virtual Machine self.virtual_machine.detach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) detached_volume = list_volumes[0] self.assertIsNone( - detached_volume.vmname, - "VM is not detached from volume" - ) + detached_volume.vmname, + "VM is not detached from volume" + ) # Creating Daily Snapshot Policy from volume self.services["recurring_snapshot"]["intervaltype"] = 'daily' self.services["recurring_snapshot"]["schedule"] = '00:00' - - list_snapshot_policy_before = SnapshotPolicy.list(self.userapiclient, volumeid=volume_created.id) - + + list_snapshot_policy_before = SnapshotPolicy.list( + self.userapiclient, + volumeid=volume_created.id) + snapshot_policy_before_size = 0 - + if list_snapshot_policy_before is not None: snapshot_policy_before_size = len(list_snapshot_policy_before) - + snapshot_policy_daily = SnapshotPolicy.create( - self.userapiclient, - volume_created.id, - self.services["recurring_snapshot"] - ) + self.userapiclient, + volume_created.id, + self.services["recurring_snapshot"] + ) self.assertIsNotNone( - snapshot_policy_daily, - "Daily Snapshot policy creation failed" - ) - #Creating expected and actual values dictionaries + snapshot_policy_daily, + "Daily Snapshot policy creation failed" + ) + # Creating expected and actual values dictionaries expected_dict = { - "schedule":self.services["recurring_snapshot"]["schedule"], - "intervaltype":1, - "volumeid":volume_created.id - } + "schedule": self.services["recurring_snapshot"]["schedule"], + "intervaltype": 1, + "volumeid": volume_created.id + } actual_dict = { - "schedule":snapshot_policy_daily.schedule, - "intervaltype":snapshot_policy_daily.intervaltype, - "volumeid":snapshot_policy_daily.volumeid - } + "schedule": snapshot_policy_daily.schedule, + "intervaltype": snapshot_policy_daily.intervaltype, + "volumeid": snapshot_policy_daily.volumeid + } status = self.__verify_values( - expected_dict, - actual_dict - ) + expected_dict, + actual_dict + ) self.assertEqual( - True, - status, - "Daily Snapshot Policy details are not as expected" - ) - - list_snapshot_policy_after = SnapshotPolicy.list(self.userapiclient, volumeid=volume_created.id) - + True, + status, + "Daily Snapshot Policy details are not as expected" + ) + + list_snapshot_policy_after = SnapshotPolicy.list( + self.userapiclient, + volumeid=volume_created.id) + self.assertIsNotNone( - list_snapshot_policy_after, - "Daily Snapshot policy creation failed" - ) + list_snapshot_policy_after, + "Daily Snapshot policy creation failed" + ) self.assertEquals( - snapshot_policy_before_size + 1, - len(list_snapshot_policy_after), - "Daily Snapshot policy creation failed" - ) + snapshot_policy_before_size + 1, + len(list_snapshot_policy_after), + "Daily Snapshot policy creation failed" + ) return - + @attr(tags=["advanced", "basic"], required_hardware="true") def test_08_volume_snapshot_policy_weekly(self): - """ - @summary: Test to verify creation of Weekly Snapshot policies from volume - + """ + @summary: Test to verify creation of Weekly Snapshot policies + from volume + Step1: Creating a Volume. Step2: Attaching volume created in Step2 to virtual machine Step3: Detaching the volume created in step2 from virtual machine @@ -1134,121 +1187,129 @@ class TestVolumes(cloudstackTestCase): Step6: Listing snapshot policies for a volume created in step1 again Step7: Verifyign that the list snapshot policy length is increased by 1 """ - list_volumes_before = Volume.list(self.userapiclient, listall=self.services["listall"]) - + list_volumes_before = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + volume_created = Volume.create( - self.userapiclient, - self.services["volume"], - zoneid=self.zone.id, - diskofferingid=self.disk_offering.id - ) - + self.userapiclient, + self.services["volume"], + zoneid=self.zone.id, + diskofferingid=self.disk_offering.id + ) + self.assertIsNotNone(volume_created, "Volume not created") - self.cleanup.append(volume_created) - - list_volumes_after = Volume.list(self.userapiclient, listall=self.services["listall"]) - + + list_volumes_after = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + self.assertEquals( - len(list_volumes_before) + 1, - len(list_volumes_after), - "Volume not created" - ) - + len(list_volumes_before) + 1, + len(list_volumes_after), + "Volume not created" + ) + # Attaching volume created to Virtual Machine self.virtual_machine.attach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) attached_volume = list_volumes[0] - + self.assertIsNotNone( - attached_volume.vmname, - "VM is not attached to Volume" - ) + attached_volume.vmname, + "VM is not attached to Volume" + ) self.assertEquals( - self.virtual_machine.name, - attached_volume.vmname, - "VM Name is not matching with attached vm" - ) + self.virtual_machine.name, + attached_volume.vmname, + "VM Name is not matching with attached vm" + ) # Detaching volume created to Virtual Machine self.virtual_machine.detach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) detached_volume = list_volumes[0] self.assertIsNone( - detached_volume.vmname, - "VM is not detached from volume" - ) + detached_volume.vmname, + "VM is not detached from volume" + ) # Creating Weekly Snapshot Policy from volume self.services["recurring_snapshot"]["intervaltype"] = 'weekly' self.services["recurring_snapshot"]["schedule"] = '00:00:1' - - list_snapshot_policy_before = SnapshotPolicy.list(self.userapiclient, volumeid=volume_created.id) - + + list_snapshot_policy_before = SnapshotPolicy.list( + self.userapiclient, + volumeid=volume_created.id) + snapshot_policy_before_size = 0 - + if list_snapshot_policy_before is not None: snapshot_policy_before_size = len(list_snapshot_policy_before) - + snapshot_policy_weekly = SnapshotPolicy.create( - self.userapiclient, - volume_created.id, - self.services["recurring_snapshot"] - ) + self.userapiclient, + volume_created.id, + self.services["recurring_snapshot"] + ) self.assertIsNotNone( - snapshot_policy_weekly, - "Weekly Snapshot policy creation failed" - ) - #Creating expected and actual values dictionaries + snapshot_policy_weekly, + "Weekly Snapshot policy creation failed" + ) + # Creating expected and actual values dictionaries expected_dict = { - "schedule":self.services["recurring_snapshot"]["schedule"], - "intervaltype":2, - "volumeid":volume_created.id - } + "schedule": self.services["recurring_snapshot"]["schedule"], + "intervaltype": 2, + "volumeid": volume_created.id + } actual_dict = { - "schedule":snapshot_policy_weekly.schedule, - "intervaltype":snapshot_policy_weekly.intervaltype, - "volumeid":snapshot_policy_weekly.volumeid - } + "schedule": snapshot_policy_weekly.schedule, + "intervaltype": snapshot_policy_weekly.intervaltype, + "volumeid": snapshot_policy_weekly.volumeid + } status = self.__verify_values( - expected_dict, - actual_dict - ) + expected_dict, + actual_dict + ) self.assertEqual( - True, - status, - "Weekly Snapshot Policy details are not as expected" - ) - - list_snapshot_policy_after = SnapshotPolicy.list(self.userapiclient, volumeid=volume_created.id) - + True, + status, + "Weekly Snapshot Policy details are not as expected" + ) + + list_snapshot_policy_after = SnapshotPolicy.list( + self.userapiclient, + volumeid=volume_created.id) + self.assertIsNotNone( - list_snapshot_policy_after, - "Weekly Snapshot policy creation failed" - ) + list_snapshot_policy_after, + "Weekly Snapshot policy creation failed" + ) self.assertEquals( - snapshot_policy_before_size + 1, - len(list_snapshot_policy_after), - "Weekly Snapshot policy creation failed" - ) + snapshot_policy_before_size + 1, + len(list_snapshot_policy_after), + "Weekly Snapshot policy creation failed" + ) return - + @attr(tags=["advanced", "basic"], required_hardware="true") def test_09_volume_snapshot_policy_monthly(self): - """ - @summary: Test to verify creation of Monthly Snapshot policies from volume - + """ + @summary: Test to verify creation of Monthly Snapshot policies + from volume + Step1: Creating a Volume. Step2: Attaching volume created in Step2 to virtual machine Step3: Detaching the volume created in step2 from virtual machine @@ -1258,304 +1319,323 @@ class TestVolumes(cloudstackTestCase): Step7: Verifyign that the list snapshot policy length is increased by 1 Step8: Deleting monthly snapshot policy created in step5 Step9: List snapshot policies for a volume again - Step10: Verifying that the list snapshot policy length is decreased by 1 + Step10:Verifying that the list snapshot policy length is decreased + by 1 """ - list_volumes_before = Volume.list(self.userapiclient, listall=self.services["listall"]) - + list_volumes_before = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + volume_created = Volume.create( - self.userapiclient, - self.services["volume"], - zoneid=self.zone.id, - diskofferingid=self.disk_offering.id - ) + self.userapiclient, + self.services["volume"], + zoneid=self.zone.id, + diskofferingid=self.disk_offering.id + ) self.assertIsNotNone(volume_created, "Volume not created") - self.cleanup.append(volume_created) - - list_volumes_after = Volume.list(self.userapiclient, listall=self.services["listall"]) - + + list_volumes_after = Volume.list( + self.userapiclient, + listall=self.services["listall"]) + self.assertEquals( - len(list_volumes_before) + 1, - len(list_volumes_after), - "Volume not created" - ) + len(list_volumes_before) + 1, + len(list_volumes_after), + "Volume not created" + ) # Attaching and Detaching custom volume created to Virtual Machine self.virtual_machine.attach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) attached_volume = list_volumes[0] - + self.assertIsNotNone( - attached_volume.vmname, - "VM is not attached to Volume" - ) + attached_volume.vmname, + "VM is not attached to Volume" + ) self.assertEquals( - self.virtual_machine.name, - attached_volume.vmname, - "VM Name is not matching with attached vm" - ) + self.virtual_machine.name, + attached_volume.vmname, + "VM Name is not matching with attached vm" + ) self.virtual_machine.detach_volume( - self.userapiclient, - volume_created - ) + self.userapiclient, + volume_created + ) list_volumes = Volume.list( - self.userapiclient, - listall=self.services["listall"], - id=volume_created.id - ) + self.userapiclient, + listall=self.services["listall"], + id=volume_created.id + ) detached_volume <TRUNCATED>
