# Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. # Import Local Modules from marvin.cloudstackTestCase import cloudstackTestCase, unittest from marvin.lib.base import (VmSnapshot, Snapshot, DiskOffering, ServiceOffering, VirtualMachine, Account, Volume) from marvin.lib.common import (get_domain, get_zone, get_template) from marvin.lib.utils import validateList, cleanup_resources from marvin.codes import PASS from nose.plugins.attrib import attr class TestSnapshots(cloudstackTestCase): @classmethod def setUpClass(cls): try: cls._cleanup = [] cls.testClient = super(TestSnapshots, cls).getClsTestClient() cls.api_client = cls.testClient.getApiClient() cls.services = cls.testClient.getParsedTestDataConfig() cls.hypervisor = cls.testClient.getHypervisorInfo() if cls.hypervisor.lower() == 'lxc': raise unittest.SkipTest("snapshots are not supported on %s" % cls.hypervisor.lower()) # Get Domain, Zone, Template cls.domain = get_domain(cls.api_client) cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests()) cls.template = get_template( 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["disk_offering"]["storagetype"] = 'local' else: cls.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.hypervisor 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._cleanup.append(cls.disk_offering) cls.service_offering = ServiceOffering.create( cls.api_client, cls.services["service_offerings"]["tiny"] ) cls._cleanup.append(cls.service_offering) 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) cls._cleanup.append(cls.account) # 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, ) except Exception as e: cls.tearDownClass() raise Exception("Warning: Exception in setup : %s" % e) return def setUp(self): self.apiClient = self.testClient.getApiClient() self.cleanup = [] def tearDown(self): # Clean up, terminate the created resources cleanup_resources(self.apiClient, self.cleanup) return @classmethod def tearDownClass(cls): try: cleanup_resources(cls.api_client, cls._cleanup) except Exception as e: raise Exception("Warning: Exception during cleanup : %s" % e) return def __verify_values(self, expected_vals, actual_vals): """ @Desc: Function to verify expected and actual values @Steps: Step1: Initializing return flag to True 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 If not making return flag to False Step4: returning the return flag after all the values are verified """ return_flag = True if len(expected_vals) != len(actual_vals): return False keys = expected_vals.keys() for i in range(0, len(expected_vals)): exp_val = expected_vals[keys[i]] act_val = actual_vals[keys[i]] if exp_val == act_val: 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 )) return return_flag @attr(tags=["advanced", "basic"], required_hardware="true") def test_01_list_volume_snapshots_pagination(self): """ @Desc: Test to List Volume Snapshots pagination @steps: Step1: Listing all the volume snapshots for a user Step2: Verifying that list size is 0 Step3: Creating (page size + 1) number of volume snapshots Step4: Listing all the volume snapshots again for a user Step5: Verifying that list size is (page size + 1) Step6: Listing all the volume snapshots in page1 Step7: Verifying that list size is (page size) Step8: Listing all the volume snapshots in page2 Step9: Verifying that list size is 1 Step10: Deleting the volume snapshot present in page 2 Step11: Listing all the volume snapshots in page2 Step12: Verifying that list size is 0 """ if self.hypervisor.lower() in ['hyperv']: raise unittest.SkipTest("This feature is not supported on existing hypervisor. Hence, skipping the test") # Listing all the volume snapshots for a User list_vol_snaps_before = Snapshot.list( self.userapiclient, listall=self.services["listall"] ) # Verifying list size is 0 self.assertIsNone( list_vol_snaps_before, "Volume snapshots exists for newly created user" ) # Listing the root volumes available for the user volumes_list = Volume.list( self.userapiclient, listall=self.services["listall"] ) status = validateList(volumes_list) self.assertEquals( PASS, status[0], "Root volume did not get created while deploying a VM" ) # Verifying list size to be 1 self.assertEquals( 1, len(volumes_list), "More than 1 root volume created for deployed VM" ) root_volume = volumes_list[0] # Creating pagesize + 1 number of volume snapshots for i in range(0, (self.services["pagesize"] + 1)): snapshot_created = Snapshot.create( self.userapiclient, root_volume.id, ) self.assertIsNotNone( snapshot_created, "Snapshot creation failed" ) self.cleanup.append(snapshot_created) # Listing all the volume snapshots for user again list_vol_snaps_after = Snapshot.list( self.userapiclient, listall=self.services["listall"] ) status = validateList(list_vol_snaps_after) self.assertEquals( PASS, status[0], "Volume snapshot creation failed" ) # Verifying that list size is pagesize + 1 self.assertEquals( self.services["pagesize"] + 1, len(list_vol_snaps_after), "Failed to create pagesize + 1 number of Volume snapshots" ) # Listing all the volume snapshots in page 1 list_vol_snaps_page1 = Snapshot.list( self.userapiclient, listall=self.services["listall"], page=1, pagesize=self.services["pagesize"] ) status = validateList(list_vol_snaps_page1) self.assertEquals( PASS, status[0], "Failed to list volume snapshots in page 1" ) # Verifying the list size to be equal to pagesize self.assertEquals( self.services["pagesize"], len(list_vol_snaps_page1), "Size of volume snapshots in page 1 is not matching" ) # Listing all the volume snapshots in page 2 list_vol_snaps_page2 = Snapshot.list( self.userapiclient, listall=self.services["listall"], page=2, pagesize=self.services["pagesize"] ) status = validateList(list_vol_snaps_page2) self.assertEquals( PASS, status[0], "Failed to list volume snapshots in page 2" ) # Verifying the list size to be equal to pagesize self.assertEquals( 1, len(list_vol_snaps_page2), "Size of volume snapshots in page 2 is not matching" ) # Deleting the volume snapshot present in page 2 Snapshot.delete( snapshot_created, self.userapiclient ) self.cleanup.remove(snapshot_created) # Listing all the snapshots in page 2 again list_vol_snaps_page2 = Snapshot.list( self.userapiclient, listall=self.services["listall"], page=2, pagesize=self.services["pagesize"] ) # Verifying that list size is 0 self.assertIsNone( list_vol_snaps_page2, "Volume snapshot not deleted from page 2" ) return @attr(tags=["advanced", "basic"], required_hardware="true") def test_02_list_volume_snapshots_byid(self): """ @Desc: Test to List Volume Snapshots by Id @Steps: Step1: Listing all the volume snapshots for a user Step2: Verifying that list size is 0 Step3: Creating a volume snapshot Step4: Listing all the volume snapshots again for a user Step5: Verifying that list size is 1 Step6: Listing all the volume snapshots by specifying snapshot id Step7: Verifying that list size is 1 Step8: Verifying details of the listed volume snapshot """ if self.hypervisor.lower() in ['hyperv']: raise unittest.SkipTest("This feature is not supported on existing hypervisor. Hence, skipping the test") # Listing all the volume snapshots for a User list_vol_snaps_before = Snapshot.list( self.userapiclient, listall=self.services["listall"] ) # Verifying list size is 0 self.assertIsNone( list_vol_snaps_before, "Volume snapshots exists for newly created user" ) # Listing the root volumes available for the user volumes_list = Volume.list( self.userapiclient, listall=self.services["listall"] ) status = validateList(volumes_list) self.assertEquals( PASS, status[0], "Root volume did not get created while deploying a VM" ) # Verifying list size to be 1 self.assertEquals( 1, len(volumes_list), "More than 1 root volume created for deployed VM" ) root_volume = volumes_list[0] # Creating a volume snapshot snapshot_created = Snapshot.create( self.userapiclient, root_volume.id, ) self.assertIsNotNone( snapshot_created, "Snapshot creation failed" ) self.cleanup.append(snapshot_created) # Listing all the volume snapshots for user again list_vol_snaps_after = Snapshot.list( self.userapiclient, listall=self.services["listall"] ) status = validateList(list_vol_snaps_after) self.assertEquals( PASS, status[0], "Volume snapshot creation failed" ) # Verifying that list size is 1 self.assertEquals( 1, len(list_vol_snaps_after), "Failed to create Volume snapshot" ) # Listing volume snapshot by id list_vol_snapshot = Snapshot.list( self.userapiclient, listall=self.services["listall"], id=snapshot_created.id ) status = validateList(list_vol_snapshot) self.assertEquals( PASS, status[0], "Failed to list Volume snapshot by Id" ) # Verifying that list size is 1 self.assertEquals( 1, len(list_vol_snapshot), "Size of the list volume snapshot by Id is not matching" ) # Verifying details of the listed snapshot to be same as snapshot created above # Creating expected and actual values dictionaries expected_dict = { "id":snapshot_created.id, "name":snapshot_created.name, "state":snapshot_created.state, "intervaltype":snapshot_created.intervaltype, "account":snapshot_created.account, "domain":snapshot_created.domainid, "volume":snapshot_created.volumeid } actual_dict = { "id":list_vol_snapshot[0].id, "name":list_vol_snapshot[0].name, "state":list_vol_snapshot[0].state, "intervaltype":list_vol_snapshot[0].intervaltype, "account":list_vol_snapshot[0].account, "domain":list_vol_snapshot[0].domainid, "volume":list_vol_snapshot[0].volumeid } vol_snapshot_status = self.__verify_values( expected_dict, actual_dict ) self.assertEqual( True, vol_snapshot_status, "Listed Volume Snapshot details are not as expected" ) return @attr(tags=["advanced", "basic"], required_hardware="true") def test_03_list_vm_snapshots_pagination(self): """ @Desc: Test to List VM Snapshots pagination @Steps: Step1: Listing all the VM snapshots for a user Step2: Verifying that list size is 0 Step3: Creating (page size + 1) number of VM snapshots Step4: Listing all the VM snapshots again for a user Step5: Verifying that list size is (page size + 1) Step6: Listing all the VM snapshots in page1 Step7: Verifying that list size is (page size) Step8: Listing all the VM snapshots in page2 Step9: Verifying that list size is 1 Step10: Deleting the VM snapshot present in page 2 Step11: Listing all the volume snapshots in page2 Step12: Verifying that list size is 0 """ if self.hypervisor.lower() in ['kvm', 'hyperv']: raise unittest.SkipTest("This feature is not supported on existing hypervisor. Hence, skipping the test") # Listing all the VM snapshots for a User list_vm_snaps_before = VmSnapshot.list( self.userapiclient, listall=self.services["listall"] ) # Verifying list size is 0 self.assertIsNone( list_vm_snaps_before, "VM snapshots exists for newly created user" ) # Creating pagesize + 1 number of VM snapshots for i in range(0, (self.services["pagesize"] + 1)): snapshot_created = VmSnapshot.create( self.userapiclient, self.virtual_machine.id, ) self.assertIsNotNone( snapshot_created, "Snapshot creation failed" ) # Listing all the VM snapshots for user again list_vm_snaps_after = VmSnapshot.list( self.userapiclient, listall=self.services["listall"] ) status = validateList(list_vm_snaps_after) self.assertEquals( PASS, status[0], "VM snapshot creation failed" ) # Verifying that list size is pagesize + 1 self.assertEquals( self.services["pagesize"] + 1, len(list_vm_snaps_after), "Failed to create pagesize + 1 number of VM snapshots" ) # Listing all the VM snapshots in page 1 list_vm_snaps_page1 = VmSnapshot.list( self.userapiclient, listall=self.services["listall"], page=1, pagesize=self.services["pagesize"] ) status = validateList(list_vm_snaps_page1) self.assertEquals( PASS, status[0], "Failed to list vm snapshots in page 1" ) # Verifying the list size to be equal to pagesize self.assertEquals( self.services["pagesize"], len(list_vm_snaps_page1), "Size of vm snapshots in page 1 is not matching" ) # Listing all the vm snapshots in page 2 list_vm_snaps_page2 = VmSnapshot.list( self.userapiclient, listall=self.services["listall"], page=2, pagesize=self.services["pagesize"] ) status = validateList(list_vm_snaps_page2) self.assertEquals( PASS, status[0], "Failed to list vm snapshots in page 2" ) # Verifying the list size to be equal to pagesize self.assertEquals( 1, len(list_vm_snaps_page2), "Size of vm snapshots in page 2 is not matching" ) # Deleting the vm snapshot present in page 2 VmSnapshot.deleteVMSnapshot( self.userapiclient, snapshot_created.id ) # Listing all the snapshots in page 2 again list_vm_snaps_page2 = VmSnapshot.list( self.userapiclient, listall=self.services["listall"], page=2, pagesize=self.services["pagesize"] ) # Verifying that list size is 0 self.assertIsNone( list_vm_snaps_page2, "VM snapshot not deleted from page 2" ) # Deleting all the existing VM snapshots list_vm_snaps = VmSnapshot.list( self.userapiclient, listall=self.services["listall"], ) status = validateList(list_vm_snaps) self.assertEquals( PASS, status[0], "All VM snapshots deleted" ) # Verifying that list size is equal to page size self.assertEquals( self.services["pagesize"], len(list_vm_snaps), "VM Snapshots count is not matching" ) # Deleting all the existing VM snapshots for i in range(0, len(list_vm_snaps)): VmSnapshot.deleteVMSnapshot( self.userapiclient, list_vm_snaps[i].id ) return @attr(tags=["advanced", "basic"], required_hardware="true") def test_04_list_vm_snapshots_byid(self): """ @summary: Test to List VM Snapshots by Id Step1: Listing all the VM snapshots for a user Step2: Verifying that list size is 0 Step3: Creating a VM snapshot Step4: Listing all the VM snapshots again for a user Step5: Verifying that list size is 1 Step6: Listing all the VM snapshots by specifying snapshot id Step7: Verifying that list size is 1 Step8: Verifying details of the listed VM snapshot """ if self.hypervisor.lower() in ['kvm', 'hyperv']: raise unittest.SkipTest("This feature is not supported on existing hypervisor. Hence, skipping the test") # Listing all the VM snapshots for a User list_vm_snaps_before = VmSnapshot.list( self.userapiclient, listall=self.services["listall"] ) # Verifying list size is 0 self.assertIsNone( list_vm_snaps_before, "VM snapshots exists for newly created user" ) # Creating a VM snapshot snapshot_created = VmSnapshot.create( self.userapiclient, self.virtual_machine.id, ) self.assertIsNotNone( snapshot_created, "Snapshot creation failed" ) # Listing all the VM snapshots for user again list_vm_snaps_after = VmSnapshot.list( self.userapiclient, listall=self.services["listall"] ) status = validateList(list_vm_snaps_after) self.assertEquals( PASS, status[0], "VM snapshot creation failed" ) # Verifying that list size is 1 self.assertEquals( 1, len(list_vm_snaps_after), "Failed to create VM snapshot" ) # Listing vm snapshot by id list_vm_snapshot = VmSnapshot.list( self.userapiclient, listall=self.services["listall"], vmsnapshotid=snapshot_created.id ) status = validateList(list_vm_snapshot) self.assertEquals( PASS, status[0], "Failed to list VM snapshot by Id" ) # Verifying that list size is 1 self.assertEquals( 1, len(list_vm_snapshot), "Size of the list vm snapshot by Id is not matching" ) # Verifying details of the listed snapshot to be same as snapshot created above # Creating expected and actual values dictionaries expected_dict = { "id":snapshot_created.id, "name":snapshot_created.name, "state":snapshot_created.state, "vmid":snapshot_created.virtualmachineid, } actual_dict = { "id":list_vm_snapshot[0].id, "name":list_vm_snapshot[0].name, "state":list_vm_snapshot[0].state, "vmid":list_vm_snapshot[0].virtualmachineid, } vm_snapshot_status = self.__verify_values( expected_dict, actual_dict ) self.assertEqual( True, vm_snapshot_status, "Listed VM Snapshot details are not as expected" ) return