# 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 from marvin.cloudstackAPI import (createVolume, createTemplate) from marvin.lib.base import (Volume, Iso, VirtualMachine, Template, Snapshot, SecurityGroup, Account, Zone, Network, NetworkOffering, DiskOffering, ServiceOffering, VmSnapshot, SnapshotPolicy, SSHKeyPair, Resources, Configurations, VpnCustomerGateway, Hypervisor, VpcOffering, VPC, NetworkACL) from marvin.lib.common import (get_zone, get_domain, get_template, list_os_types) from marvin.lib.utils import (validateList, cleanup_resources, random_gen) from marvin.codes import (PASS, FAIL, EMPTY_LIST) from nose.plugins.attrib import attr import time 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() # 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.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._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, ) cls._cleanup.append(cls.virtual_machine) 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", "provisioning"]) 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 """ # 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 ) # 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", "provisioning"]) 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 """ # 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", "provisioning"]) 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 """ # 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", "provisioning"]) 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 """ # 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