mirror of
				https://github.com/apache/cloudstack.git
				synced 2025-10-26 08:42:29 +01:00 
			
		
		
		
	Signed-off-by: Santhosh Edukulla <santhosh.edukulla@gmail.com> (cherry picked from commit d7dea70e89e45b0864bfbcaab3b58f896a6c75c1)
		
			
				
	
	
		
			636 lines
		
	
	
		
			30 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			636 lines
		
	
	
		
			30 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
| # 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()
 | |
|             # 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
 |