KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference
How to use migrate_volume method in tempest

Best Python code snippet using tempest_python
testpath_volumelifecycle.py
Source:testpath_volumelifecycle.py
1# Licensed to the Apache Software Foundation (ASF) under one2# or more contributor license agreements.  See the NOTICE file3# distributed with this work for additional information4# regarding copyright ownership.  The ASF licenses this file5# to you under the Apache License, Version 2.0 (the6# "License"); you may not use this file except in compliance7# with the License.  You may obtain a copy of the License at8#9#   http://www.apache.org/licenses/LICENSE-2.010#11# Unless required by applicable law or agreed to in writing,12# software distributed under the License is distributed on an13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY14# KIND, either express or implied.  See the License for the15# specific language governing permissions and limitations16# under the License.17"""Utilities functions18"""19# All tests inherit from cloudstackTestCase20from marvin.cloudstackTestCase import cloudstackTestCase, unittest21# Import Integration Libraries22from marvin.codes import FAILED, PASS23# base - contains all resources as entities and defines create, delete,24# list operations on them25from marvin.lib.base import (Account,26                             VirtualMachine,27                             ServiceOffering,28                             User,29                             DiskOffering,30                             Volume,31                             Template,32                             StoragePool,33                             Resources)34from marvin.lib.utils import cleanup_resources, validateList35#common - commonly used methods for all tests are listed here36from marvin.lib.common import (get_zone,37                               get_domain,38                               get_template,39                               list_virtual_machines,40                               find_storage_pool_type)41from nose.plugins.attrib import attr42import os43import urllib44import tempfile45def verify_attach_volume(self, vmid, volid):46    list_volumes = Volume.list(self.userapiclient,47                               id=volid48                               )49    self.assertEqual(50        validateList(list_volumes)[0],51        PASS,52        "Check List volume response for volume %s" %53        volid)54    self.assertEqual(55        len(list_volumes),56        1,57        "There is no data disk attached to vm id:%s" %58        vmid)59    self.assertEqual(60        list_volumes[0].virtualmachineid,61        vmid,62        "Check if volume state (attached) is reflected")63    self.debug("volume id:%s successfully attached to vm id%s" % (volid, vmid))64    return65def verify_detach_volume(self, vmid, volid):66    list_volumes = Volume.list(self.userapiclient,67                               id=volid68                               )69    self.assertEqual(70        validateList(list_volumes)[0],71        PASS,72        "Check List volume response for volume %s" %73        volid)74    self.assertEqual(75        len(list_volumes),76        1,77        "Detach data disk id: %s  for vm id :%s was not successful" %78        (volid,79         vmid))80    self.assertEqual(81        list_volumes[0].virtualmachineid,82        None,83        "Check if volume state (attached) is reflected")84    self.debug(85        "volume id: %s successfully detached from vm id:%s" %86        (volid, vmid))87def verify_vm(self, vmid):88    list_vm = list_virtual_machines(self.userapiclient,89                                    account=self.account.name,90                                    domainid=self.account.domainid,91                                    id=vmid92                                    )93    self.assertEqual(94        validateList(list_vm)[0],95        PASS,96        "Check List vm response for vmid: %s" %97        vmid)98    self.assertGreater(99        len(list_vm),100        0,101        "Check the list vm response for vm id:  %s" %102        vmid)103    vm = list_vm[0]104    self.assertEqual(105        vm.id,106        str(vmid),107        "Vm deployed is different from the test")108    self.assertEqual(vm.state, "Running", "VM is not in Running state")109    self.debug("VM got created successfully %s" % vmid)110class TestPathVolume(cloudstackTestCase):111    @classmethod112    def setUpClass(cls):113        testClient = super(TestPathVolume, cls).getClsTestClient()114        cls.apiclient = testClient.getApiClient()115        cls.testdata = testClient.getParsedTestDataConfig()116        # Get Zone,Domain and templates117        cls.domain = get_domain(cls.apiclient)118        cls.zone = get_zone(cls.apiclient)119        cls.testdata["mode"] = cls.zone.networktype120        cls.hypervisor = testClient.getHypervisorInfo()121        cls._cleanup = []122        cls.insuffStorage = False123        cls.unsupportedHypervisor = False124        #for LXC if the storage pool of type 'rbd' ex: ceph is not available, skip the test125        if cls.hypervisor.lower() == 'lxc':126            if not find_storage_pool_type(cls.apiclient, storagetype='rbd'):127                cls.insuffStorage   = True128                return129        cls.template = get_template(130            cls.apiclient,131            cls.zone.id,132            cls.testdata["ostype"])133        cls.testdata["template"]["ostypeid"] = cls.template.ostypeid134        if cls.template == FAILED:135            cls.fail(136                "get_template() failed to return template with description \137                %s" %138                cls.testdata["ostype"])139        try:140            cls.account = Account.create(cls.apiclient,141                                         cls.testdata["account"],142                                         domainid=cls.domain.id143                                         )144            cls._cleanup.append(cls.account)145            # createa two service offerings146            cls.service_offering_1 = ServiceOffering.create(147                cls.apiclient,148                cls.testdata["service_offerings"]["small"])149            cls._cleanup.append(cls.service_offering_1)150            # Create Disk offerings151            cls.disk_offering_1 = DiskOffering.create(152                cls.apiclient,153                cls.testdata["disk_offering"])154            cls._cleanup.append(cls.disk_offering_1)155            # check if zone wide storage is enable156            cls.list_storage = StoragePool.list(cls.apiclient,157                                                scope="ZONE"158                                                )159            if cls.list_storage:160                cls.zone_wide_storage = cls.list_storage[0]161                cls.debug(162                    "zone wide storage id is %s" %163                    cls.zone_wide_storage.id)164                cls.testdata["tags"] = "zp"165                update1 = StoragePool.update(cls.apiclient,166                                             id=cls.zone_wide_storage.id,167                                             tags=cls.testdata["tags"]168                                             )169                cls.debug(170                    "Storage %s pool tag%s" %171                    (cls.zone_wide_storage.id, update1.tags))172                cls.testdata["service_offerings"]["tags"] = "zp"173                cls.tagged_so = ServiceOffering.create(174                    cls.apiclient,175                    cls.testdata["service_offerings"])176                cls.testdata["service_offerings"]["tags"] = " "177                cls._cleanup.append(cls.tagged_so)178                # create tagged disk offerings179                cls.testdata["disk_offering"]["tags"] = "zp"180                cls.disk_offering_tagged = DiskOffering.create(181                    cls.apiclient,182                    cls.testdata["disk_offering"])183                cls._cleanup.append(cls.disk_offering_tagged)184            else:185                cls.debug("No zone wide storage found")186            # check if local storage is enable187            if cls.zone.localstorageenabled:188                cls.testdata["disk_offering"]["tags"] = " "189                cls.testdata["service_offerings"]["storagetype"] = 'local'190                cls.service_offering_2 = ServiceOffering.create(191                    cls.apiclient,192                    cls.testdata["service_offerings"])193                cls._cleanup.append(cls.service_offering_2)194                # craete a compute offering with local storage195                cls.testdata["disk_offering"]["storagetype"] = 'local'196                cls.disk_offering_local = DiskOffering.create(197                    cls.apiclient,198                    cls.testdata["disk_offering"])199                cls._cleanup.append(cls.disk_offering_local)200                cls.testdata["disk_offering"]["storagetype"] = ' '201            else:202                cls.debug("No local storage found")203            cls.userapiclient = testClient.getUserApiClient(204                UserName=cls.account.name,205                DomainName=cls.account.domain)206            # Check if login is successful with new account207            response = User.login(cls.userapiclient,208                                  username=cls.account.name,209                                  password=cls.testdata["account"]["password"]210                                  )211            assert response.sessionkey is not None212            #response should have non null value213        except Exception as e:214                cls.tearDownClass()215                raise e216        return217    def setUp(self):218        self.apiclient = self.testClient.getApiClient()219        self.dbclient = self.testClient.getDbConnection()220        if self.unsupportedHypervisor or self.insuffStorage:221            self.skipTest("Skipping test because of insuff resources\222                    %s" % self.hypervisor)223    @classmethod224    def tearDownClass(cls):225        try:226            cleanup_resources(cls.apiclient, cls._cleanup)227        except Exception as e:228            raise Exception("Warning:Exception during cleanup: %s" % e)229    @attr(230        tags=[231            "advanced",232            "advancedsg",233            "basic",234        ],235        required_hardware="True")236    def test_01_positive_path(self):237        """238        positive test for volume life cycle239        # 1. Deploy a vm [vm1] with shared storage and data disk240        # 2. Deploy a vm [vm2]with shared storage without data disk241        # 3. TBD242        # 4. Create a new volume and attache to vm2243        # 5. Detach data disk from vm1 and download it244        #  Variance(1-9)245        # 6. Upload volume by providing url of downloaded volume in step 5246        # 7. Attach the volume to a different vm - vm2247        # 8. Try to delete an attached volume248        # 9. Create template from root volume of VM1249        # 10. Create new VM using the template created in step 9250        # 11. Delete the template251        # 12. Detach the disk from VM2 and re-attach the disk to VM1252        # 13.TBD253        # 14.TBD254        # 15.Migrate volume(detached) and then attach to a vm and live-migrate255        # 16.Upload volume of size smaller  than256            storage.max.volume.upload.size(leaving the negative case)257        # 17.TBD258        # 18.TBD259        # 19.TBD260        # 20.Detach data disks from VM2 and delete volume261        """262        if self.hypervisor.lower() in ['lxc']:263            self.skipTest(264                "feature is not supported in %s" %265                self.hypervisor)266        # 1. Deploy a vm [vm1] with shared storage and data disk267        self.virtual_machine_1 = VirtualMachine.create(268            self.userapiclient,269            self.testdata["small"],270            templateid=self.template.id,271            accountid=self.account.name,272            domainid=self.account.domainid,273            serviceofferingid=self.service_offering_1.id,274            zoneid=self.zone.id,275            diskofferingid=self.disk_offering_1.id,276            mode=self.testdata["mode"])277        verify_vm(self, self.virtual_machine_1.id)278        # List data volume for vm1279        list_volume = Volume.list(self.userapiclient,280                                  virtualmachineid=self.virtual_machine_1.id,281                                  type='DATADISK'282                                  )283        self.assertEqual(284            validateList(list_volume)[0],285            PASS,286            "Check List volume response for vm id  %s" %287            self.virtual_machine_1.id)288        list_data_volume_for_vm1 = list_volume[0]289        self.assertEqual(290            len(list_volume),291            1,292            "There is no data disk attached to vm id:%s" %293            self.virtual_machine_1.id)294        self.assertEqual(295            list_data_volume_for_vm1.virtualmachineid, str(296                self.virtual_machine_1.id),297            "Check if volume state (attached) is reflected")298        # 2. Deploy a vm [vm2]with shared storage without data disk299        self.virtual_machine_2 = VirtualMachine.create(300            self.userapiclient,301            self.testdata["small"],302            templateid=self.template.id,303            accountid=self.account.name,304            domainid=self.account.domainid,305            serviceofferingid=self.service_offering_1.id,306            zoneid=self.zone.id,307            mode=self.testdata["mode"])308        verify_vm(self, self.virtual_machine_2.id)309        # 4. Create a new volume and attache to vm2310        self.volume = Volume.create(self.userapiclient,311                                    services=self.testdata["volume"],312                                    diskofferingid=self.disk_offering_1.id,313                                    zoneid=self.zone.id314                                    )315        list_data_volume = Volume.list(self.userapiclient,316                                       id=self.volume.id317                                       )318        self.assertEqual(319            validateList(list_data_volume)[0],320            PASS,321            "Check List volume response for volume %s" %322            self.volume.id)323        self.assertEqual(324            list_data_volume[0].id,325            self.volume.id,326            "check list volume response for volume id:  %s" %327            self.volume.id)328        self.debug(329            "volume id %s got created successfully" %330            list_data_volume[0].id)331        # Attach volume to vm2332        self.virtual_machine_2.attach_volume(self.userapiclient,333                                             self.volume334                                             )335        verify_attach_volume(self, self.virtual_machine_2.id, self.volume.id)336        # Variance337        if self.zone.localstorageenabled:338            # V1.Create vm3 with local storage offering339            self.virtual_machine_local_3 = VirtualMachine.create(340                self.userapiclient,341                self.testdata["small"],342                templateid=self.template.id,343                accountid=self.account.name,344                domainid=self.account.domainid,345                serviceofferingid=self.service_offering_2.id,346                zoneid=self.zone.id,347                mode=self.testdata["mode"])348            verify_vm(self, self.virtual_machine_local_3.id)349            # V2.create two data disk on local storage350            self.local_volumes = []351            for i in range(2):352                local_volume = Volume.create(353                    self.userapiclient,354                    services=self.testdata["volume"],355                    diskofferingid=self.disk_offering_local.id,356                    zoneid=self.zone.id)357                list_local_data_volume = Volume.list(self.userapiclient,358                                                     id=local_volume.id359                                                     )360                self.assertEqual(361                    validateList(list_local_data_volume)[0],362                    PASS,363                    "Check List volume response for volume %s" %364                    local_volume.id)365                self.assertEqual(366                    list_local_data_volume[0].id,367                    local_volume.id,368                    "check list volume response for volume id:  %s" %369                    local_volume.id)370                self.debug(371                    "volume id %s got created successfully" %372                    list_local_data_volume[0].id)373                self.local_volumes.append(local_volume)374            # V3.Attach local disk to vm1375            self.virtual_machine_1.attach_volume(self.userapiclient,376                                                 self.local_volumes[0]377                                                 )378            verify_attach_volume(379                self,380                self.virtual_machine_1.id,381                self.local_volumes[0].id)382        if self.list_storage:383            # V4.create vm4 with zone wide storage384            self.virtual_machine_zone_4 = VirtualMachine.create(385                self.userapiclient,386                self.testdata["small"],387                templateid=self.template.id,388                accountid=self.account.name,389                domainid=self.account.domainid,390                serviceofferingid=self.tagged_so.id,391                zoneid=self.zone.id,392                mode=self.testdata["mode"])393            verify_vm(self, self.virtual_machine_zone_4.id)394            # V5.Create two data disk on zone  wide storage395            self.zone_volumes = []396            for i in range(2):397                zone_volume = Volume.create(398                    self.userapiclient,399                    services=self.testdata["volume"],400                    diskofferingid=self.disk_offering_tagged.id,401                    zoneid=self.zone.id)402                list_zone_data_volume = Volume.list(self.userapiclient,403                                                    id=zone_volume.id404                                                    )405                self.assertEqual(406                    validateList(list_zone_data_volume)[0],407                    PASS,408                    "Check List volume response for volume %s" %409                    zone_volume.id)410                self.assertEqual(411                    list_zone_data_volume[0].id,412                    zone_volume.id,413                    "check list volume response for volume id:  %s" %414                    zone_volume.id)415                self.debug(416                    "volume id:%s got created successfully" %417                    list_zone_data_volume[0].id)418                self.zone_volumes.append(zone_volume)419            # V6.Attach data disk running on ZWPS to VM1 (root disk on shared)420            self.virtual_machine_1.attach_volume(self.userapiclient,421                                                 self.zone_volumes[0]422                                                 )423            verify_attach_volume(424                self,425                self.virtual_machine_1.id,426                self.zone_volumes[0].id)427            # V7. Create a cluster wide volume and attach to vm running on zone428            # wide storage429            self.cluster_volume = Volume.create(430                self.userapiclient,431                services=self.testdata["volume"],432                diskofferingid=self.disk_offering_1.id,433                zoneid=self.zone.id)434            list_cluster_volume = Volume.list(self.userapiclient,435                                              id=self.cluster_volume.id436                                              )437            self.assertEqual(438                validateList(list_cluster_volume)[0],439                PASS,440                "Check List volume response for volume %s" %441                self.cluster_volume.id)442            self.assertEqual(443                list_cluster_volume[0].id, str(444                    self.cluster_volume.id), "volume does not exist %s" %445                self.cluster_volume.id)446            self.debug(447                "volume id %s got created successfuly" %448                list_cluster_volume[0].id)449            self.virtual_machine_zone_4.attach_volume(self.userapiclient,450                                                      self.cluster_volume451                                                      )452            verify_attach_volume(453                self,454                self.virtual_machine_zone_4.id,455                self.cluster_volume.id)456        if self.list_storage and self.zone.localstorageenabled:457            # V8.Attach zone wide volume to vm running on local storage458            self.virtual_machine_local_3.attach_volume(self.userapiclient,459                                                       self.zone_volumes[1]460                                                       )461            verify_attach_volume(462                self,463                self.virtual_machine_local_3.id,464                self.zone_volumes[1].id)465            # V9.Attach local volume to a vm running on zone wide storage466            self.virtual_machine_zone_4.attach_volume(self.userapiclient,467                                                      self.local_volumes[1]468                                                      )469            verify_attach_volume(470                self,471                self.virtual_machine_zone_4.id,472                self.local_volumes[1].id)473        # 5. Detach data disk from vm1 and download it474        self.virtual_machine_1.detach_volume(self.userapiclient,475                                             volume=list_data_volume_for_vm1476                                             )477        verify_detach_volume(478            self,479            self.virtual_machine_1.id,480            list_data_volume_for_vm1.id)481        # download detached volume482        self.extract_volume = Volume.extract(483            self.userapiclient,484            volume_id=list_data_volume_for_vm1.id,485            zoneid=self.zone.id,486            mode='HTTP_DOWNLOAD')487        self.debug("extracted url is%s  :" % self.extract_volume.url)488        try:489            formatted_url = urllib.unquote_plus(self.extract_volume.url)490            self.debug(491                "Attempting to download volume at url %s" %492                formatted_url)493            response = urllib.urlopen(formatted_url)494            self.debug("response from volume url %s" % response.getcode())495            fd, path = tempfile.mkstemp()496            self.debug(497                "Saving volume %s to path %s" %498                (list_data_volume_for_vm1.id, path))499            os.close(fd)500            with open(path, 'wb') as fd:501                fd.write(response.read())502            self.debug("Saved volume successfully")503        except Exception:504            self.fail(505                "Extract Volume Failed with invalid URL %s (vol id: %s)" %506                (self.extract_volume, list_data_volume_for_vm1.id))507        # checking  format of  downloaded volume and assigning to508        # testdata["volume_upload"]509        if "OVA" in self.extract_volume.url.upper():510            self.testdata["configurableData"]["upload_volume"]["format"] = "OVA"511        if "QCOW2" in self.extract_volume.url.upper():512            self.testdata["configurableData"]["upload_volume"]["format"] = "QCOW2"513        # 6. Upload volume by providing url of downloaded volume in step 5514        self.upload_response = Volume.upload(515            self.userapiclient,516            zoneid=self.zone.id,517            url=self.extract_volume.url,518            services=self.testdata["configurableData"]["upload_volume"])519        self.upload_response.wait_for_upload(self.userapiclient520                                             )521        self.debug("uploaded volume id is %s" % self.upload_response.id)522        # 7. Attach the volume to a different vm - vm2523        self.virtual_machine_2.attach_volume(self.userapiclient,524                                             volume=self.upload_response525                                             )526        verify_attach_volume(527            self,528            self.virtual_machine_2.id,529            self.upload_response.id)530        # 8. Try to delete an attached volume531        try:532            self.volume.delete(self.userapiclient533                               )534            self.fail(535                "Volume got deleted in attached state %s " %536                self.volume.id)537        except Exception as e:538            self.debug("Attached volume deletion failed because  %s" % e)539        # 9. Create template from root volume of VM1(stop VM->create template540        # -> start vm)541        self.virtual_machine_1.stop(self.userapiclient542                                    )543        self.list_root_disk_for_vm1 = Volume.list(544            self.userapiclient,545            virtualmachineid=self.virtual_machine_1.id,546            type='ROOT')547        self.assertEqual(548            validateList(549                self.list_root_disk_for_vm1)[0],550            PASS,551            "Check List volume response for vm %s" %552            self.virtual_machine_1.id)553        self.assertEqual(554            len(555                self.list_root_disk_for_vm1),556            1,557            "list root disk for vm1 is empty : %s" %558            self.virtual_machine_1.id)559        self.template_from_vm1_root_disk = Template.create(560            self.userapiclient,561            self.testdata["template"],562            self.list_root_disk_for_vm1[0].id,563            account=self.account.name,564            domainid=self.account.domainid)565        list_template = Template.list(566            self.userapiclient,567            templatefilter=self.testdata["templatefilter"],568            id=self.template_from_vm1_root_disk.id)569        self.assertEqual(570            validateList(list_template)[0],571            PASS,572            "Check List template response for template id %s" %573            self.template_from_vm1_root_disk.id)574        self.assertEqual(575            len(list_template),576            1,577            "list template response is empty for template id  : %s" %578            list_template[0].id)579        self.assertEqual(580            list_template[0].id,581            self.template_from_vm1_root_disk.id,582            "list template id is not same as created template")583        self.debug(584            "Template id:%s got created successfully" %585            self.template_from_vm1_root_disk.id)586        self.virtual_machine_1.start(self.userapiclient587                                     )588        # 10. Deploy a vm using template ,created  from vm1's root disk589        self.virtual_machine_3 = VirtualMachine.create(590            self.userapiclient,591            self.testdata["small"],592            templateid=self.template_from_vm1_root_disk.id,593            accountid=self.account.name,594            domainid=self.account.domainid,595            serviceofferingid=self.service_offering_1.id,596            zoneid=self.zone.id,597            mode=self.testdata["mode"])598        verify_vm(self, self.virtual_machine_3.id)599        # 11.delete the template created from root disk of vm1600        try:601            self.template_from_vm1_root_disk.delete(self.userapiclient602                                                    )603            self.debug(604                "Template id: %s got deleted successfuly" %605                self.template_from_vm1_root_disk.id)606        except Exception as e:607            raise Exception("Template deletion failed with error %s" % e)608        list_template = Template.list(609            self.userapiclient,610            templatefilter=self.testdata["templatefilter"],611            id=self.template_from_vm1_root_disk.id)612        self.assertEqual(613            list_template,614            None,615            "Template is not deleted, id %s:" %616            self.template_from_vm1_root_disk.id)617        self.debug(618            "Template id%s got deleted successfully" %619            self.template_from_vm1_root_disk.id)620        # List vm and check the state of vm621        verify_vm(self, self.virtual_machine_3.id)622        # 12.Detach the disk from VM2 and re-attach the disk to VM1623        self.virtual_machine_2.detach_volume(self.userapiclient,624                                             volume=self.upload_response625                                             )626        verify_detach_volume(627            self,628            self.virtual_machine_2.id,629            self.upload_response.id)630        self.virtual_machine_1.attach_volume(self.userapiclient,631                                             volume=self.upload_response632                                             )633        verify_attach_volume(634            self,635            self.virtual_machine_1.id,636            self.upload_response.id)637        # 15.Migrate volume(detached) and then attach to a vm and live-migrate638        self.migrate_volume = Volume.create(639            self.userapiclient,640            services=self.testdata["volume"],641            diskofferingid=self.disk_offering_1.id,642            zoneid=self.zone.id)643        list_volume = Volume.list(self.apiclient,644                                  id=self.migrate_volume.id645                                  )646        self.assertEqual(647            validateList(list_volume)[0],648            PASS,649            "Check List volume response for volume %s" %650            self.migrate_volume.id)651        self.assertEqual(652            list_volume[0].id, str(653                self.migrate_volume.id), "volume does not exist %s" %654            self.migrate_volume.id)655        self.debug("volume id %s got created successfuly" % list_volume[0].id)656        self.virtual_machine_1.attach_volume(self.userapiclient,657                                             self.migrate_volume658                                             )659        verify_attach_volume(660            self,661            self.virtual_machine_1.id,662            self.migrate_volume.id)663        self.virtual_machine_1.detach_volume(self.userapiclient,664                                             volume=self.migrate_volume665                                             )666        verify_detach_volume(667            self,668            self.virtual_machine_1.id,669            self.migrate_volume.id)670        list_volume = Volume.list(self.apiclient,671                                  id=self.migrate_volume.id672                                  )673        self.assertEqual(674            validateList(list_volume)[0],675            PASS,676            "Check List volume response for volume %s" %677            self.migrate_volume.id)678        self.assertEqual(679            list_volume[0].id, str(680                self.migrate_volume.id), "volume does not exist %s" %681            self.migrate_volume.id)682        self.debug("volume id %s got created successfuly" % list_volume[0].id)683        list_pool = StoragePool.list(self.apiclient,684                                     id=list_volume[0].storageid685                                     )686        self.assertEqual(687            validateList(list_pool)[0],688            PASS,689            "Check List pool response for storage id %s" %690            list_volume[0].storageid)691        self.assertGreater(692            len(list_pool),693            0,694            "Check the list list storagepoolresponse for vm id:  %s" %695            list_volume[0].storageid)696        list_pools = StoragePool.list(self.apiclient,697                                      scope=list_pool[0].scope698                                      )699        self.assertEqual(700            validateList(list_pools)[0],701            PASS,702            "Check List pool response for scope %s" %703            list_pool[0].scope)704        self.assertGreater(705            len(list_pools),706            0,707            "Check the list vm response for scope :%s" %708            list_volume[0].scope)709        storagepoolid = None710        for i in range(len(list_pools)):711            if list_volume[0].storageid != list_pools[i].id:712                storagepoolid = list_pools[i].id713                break714            else:715                self.debug("No pool available for volume migration ")716        if storagepoolid is not None:717            try:718                volume_migrate = Volume.migrate(self.apiclient,719                                                storageid=storagepoolid,720                                                volumeid=self.migrate_volume.id721                                                )722            except Exception as e:723                raise Exception("Volume migration failed with error %s" % e)724            self.virtual_machine_2.attach_volume(self.userapiclient,725                                                 self.migrate_volume726                                                 )727            verify_attach_volume(728                self,729                self.virtual_machine_2.id,730                self.migrate_volume.id)731            pool_for_migration = StoragePool.listForMigration(732                self.apiclient,733                id=self.migrate_volume.id)734            self.assertEqual(735                validateList(pool_for_migration)[0],736                PASS,737                "Check list pool For Migration response for volume %s" %738                self.migrate_volume.id)739            self.assertGreater(740                len(pool_for_migration),741                0,742                "Check the listForMigration response for volume :%s" %743                self.migrate_volume.id)744            try:745                volume_migrate = Volume.migrate(746                    self.apiclient,747                    storageid=pool_for_migration[0].id,748                    volumeid=self.migrate_volume.id,749                    livemigrate=True)750            except Exception as e:751                raise Exception("Volume migration failed with error %s" % e)752        else:753            try:754                self.migrate_volume.delete(self.userapiclient755                                           )756                self.debug(757                    "volume id:%s got deleted successfully " %758                    self.migrate_volume.id)759            except Exception as e:760                raise Exception("Volume deletion failed with error %s" % e)761        # 16.Upload volume of size smaller  than762        # storage.max.volume.upload.size(leaving the negative case)763        self.testdata["configurableData"]["upload_volume"]["format"] = "VHD"764        volume_upload = Volume.upload(self.userapiclient,765                                      self.testdata["configurableData"]["upload_volume"],766                                      zoneid=self.zone.id767                                      )768        volume_upload.wait_for_upload(self.userapiclient769                                      )770        self.debug(771            "volume id :%s got uploaded successfully is " %772            volume_upload.id)773        # 20.Detach data disk from vm 2 and delete the volume774        self.virtual_machine_2.detach_volume(self.userapiclient,775                                             volume=self.volume776                                             )777        verify_detach_volume(self, self.virtual_machine_2.id, self.volume.id)778        try:779            self.volume.delete(self.userapiclient780                               )781            self.debug("volume id:%s got deleted successfully " %782                       self.volume.id)783        except Exception as e:784            raise Exception("Volume deletion failed with error %s" % e)785    @attr(786        tags=[787            "advanced",788            "advancedsg",789            "basic",790        ],791        required_hardware="True")792    def test_02_negative_path(self):793        """794        negative test for volume life cycle795        # 1. Deploy a vm [vm1] with shared storage and  data disk796        #v1. Create VM2 with local storage offering disk offerings797        # 2.TBD798        # 3. Detach the data disk from VM1 and Download the volume799        # 4.TBD800        # 5. Attach volume with deviceid = 0801        # 6. Attach volume, specify a VM which is destroyed802        # 7.TBD803        # 8.TBD804        # 9.TBD805        # 10.TBD806        # 11.Upload the volume from T3 by providing the URL of the downloaded807             volume, but specify a wrong format (not supported by the808             hypervisor)809        # 12.Upload the same volume from T4 by providing a wrong URL810        # 13.Upload volume, provide wrong checksum811        # 14.Upload a volume when maximum limit for the account is reached812        # 15.TBD813        # 16.Upload volume with all correct parameters814             (covered in positive test path)815        # 17.TBD816        # 18.TBD817        # 19.Now attach the volume with all correct parameters818            (covered in positive test path)819        # 20.Destroy and expunge all VMs820        """821        # 1. Deploy a vm [vm1] with shared storage and data disk822        self.virtual_machine_1 = VirtualMachine.create(823            self.userapiclient,824            self.testdata["small"],825            templateid=self.template.id,826            accountid=self.account.name,827            domainid=self.account.domainid,828            serviceofferingid=self.service_offering_1.id,829            zoneid=self.zone.id,830            diskofferingid=self.disk_offering_1.id,831            mode=self.testdata["mode"])832        verify_vm(self, self.virtual_machine_1.id)833        # List data volume for vm1834        list_volume = Volume.list(self.userapiclient,835                                  virtualmachineid=self.virtual_machine_1.id,836                                  type='DATADISK'837                                  )838        self.assertEqual(839            validateList(list_volume)[0],840            PASS,841            "Check List volume response for vm id  %s" %842            self.virtual_machine_1.id)843        list_data_volume_for_vm1 = list_volume[0]844        self.assertEqual(845            len(list_volume),846            1,847            "There is no data disk attached to vm id:%s" %848            self.virtual_machine_1.id)849        self.assertEqual(850            list_data_volume_for_vm1.virtualmachineid, str(851                self.virtual_machine_1.id),852            "Check if volume state (attached) is reflected")853        # Variance854        if self.zone.localstorageenabled:855            # V1.Create vm3 with local storage offering856            self.virtual_machine_local_2 = VirtualMachine.create(857                self.userapiclient,858                self.testdata["small"],859                templateid=self.template.id,860                accountid=self.account.name,861                domainid=self.account.domainid,862                serviceofferingid=self.service_offering_2.id,863                zoneid=self.zone.id,864                mode=self.testdata["mode"])865            verify_vm(self, self.virtual_machine_local_2.id)866        # 3. Detach the data disk from VM1 and Download the volume867        self.virtual_machine_1.detach_volume(self.userapiclient,868                                             volume=list_data_volume_for_vm1869                                             )870        verify_detach_volume(871            self,872            self.virtual_machine_1.id,873            list_data_volume_for_vm1.id)874        # download detached volume875        self.extract_volume = Volume.extract(876            self.userapiclient,877            volume_id=list_data_volume_for_vm1.id,878            zoneid=self.zone.id,879            mode='HTTP_DOWNLOAD')880        self.debug("extracted url is%s  :" % self.extract_volume.url)881        try:882            formatted_url = urllib.unquote_plus(self.extract_volume.url)883            self.debug(884                "Attempting to download volume at url %s" %885                formatted_url)886            response = urllib.urlopen(formatted_url)887            self.debug("response from volume url %s" % response.getcode())888            fd, path = tempfile.mkstemp()889            self.debug(890                "Saving volume %s to path %s" %891                (list_data_volume_for_vm1.id, path))892            os.close(fd)893            with open(path, 'wb') as fd:894                fd.write(response.read())895            self.debug("Saved volume successfully")896        except Exception:897            self.fail(898                "Extract Volume Failed with invalid URL %s (vol id: %s)" %899                (self.extract_volume, list_data_volume_for_vm1.id))900        # 6. Attach volume, specify a VM which is destroyed901        self.virtual_machine_2 = VirtualMachine.create(902            self.userapiclient,903            self.testdata["small"],904            templateid=self.template.id,905            accountid=self.account.name,906            domainid=self.account.domainid,907            serviceofferingid=self.service_offering_1.id,908            zoneid=self.zone.id,909            mode=self.testdata["mode"])910        verify_vm(self, self.virtual_machine_2.id)911        try:912            self.virtual_machine_2.delete(self.apiclient)913        except Exception as e:914            raise Exception("Vm deletion failed with error %s" % e)915        # Create a new volume916        self.volume = Volume.create(self.userapiclient,917                                    services=self.testdata["volume"],918                                    diskofferingid=self.disk_offering_1.id,919                                    zoneid=self.zone.id920                                    )921        list_data_volume = Volume.list(self.userapiclient,922                                       id=self.volume.id923                                       )924        self.assertEqual(925            validateList(list_data_volume)[0],926            PASS,927            "Check List volume response for volume %s" %928            self.volume.id)929        self.assertEqual(930            list_data_volume[0].id,931            self.volume.id,932            "check list volume response for volume id:  %s" %933            self.volume.id)934        self.debug(935            "volume id %s got created successfully" %936            list_data_volume[0].id)937        # try  Attach volume to vm2938        try:939            self.virtual_machine_2.attach_volume(self.userapiclient,940                                                 self.volume941                                                 )942            self.fail("Volume got attached to a destroyed vm ")943        except Exception:944            self.debug("Volume cant not be attached to a destroyed vm ")945        # 11.Upload the volume  by providing the URL of the downloaded946        # volume, but specify a wrong format (not supported by the hypervisor)947        if "OVA" in self.extract_volume.url.upper():948            self.testdata["configurableData"]["upload_volume"]["format"] = "VHD"949        else:950            self.testdata["configurableData"]["upload_volume"]["format"] = "OVA"951        try:952            self.upload_response = Volume.upload(953                self.userapiclient,954                zoneid=self.zone.id,955                url=self.extract_volume.url,956                services=self.testdata["configurableData"]["upload_volume"])957            self.fail("Volume got uploaded with invalid format")958        except Exception as e:959            self.debug("upload volume failed due %s" % e)960        # 12. Upload the same volume from T4 by providing a wrong URL961        self.testdata["configurableData"]["upload_volume"]["format"] = "VHD"962        if "OVA" in self.extract_volume.url.upper():963            self.testdata["configurableData"]["upload_volume"]["format"] = "OVA"964        if "QCOW2" in self.extract_volume.url.upper():965            self.testdata["configurableData"]["upload_volume"]["format"] = "QCOW2"966        u1 = self.extract_volume.url.split('.')967        u1[-2] = "wrong"968        wrong_url = ".".join(u1)969        try:970            self.upload_response = Volume.upload(971                self.userapiclient,972                zoneid=self.zone.id,973                url=wrong_url,974                services=self.testdata["configurableData"]["upload_volume"])975            self.upload_response.wait_for_upload(self.userapiclient976                                                 )977            self.fail("volume got uploaded with wrong url")978        except Exception as e:979            self.debug("upload volume failed due to %s" % e)980        # 13.Upload volume, provide wrong checksum981        try:982            self.upload_response = Volume.upload(983                self.userapiclient,984                zoneid=self.zone.id,985                url=self.extract_volume.url,986                services=self.testdata["configurableData"]["upload_volume"],987                checksome="123456")988            self.upload_response.wait_for_upload(self.userapiclient989                                                 )990            self.fail("volume got uploaded with wrong checksome")991        except Exception as e:992            self.debug("upload volume failed due to %s" % e)993        # 14.Upload a volume when maximum limit for the account is reached994        account_update = Resources.updateLimit(self.apiclient,995                                               resourcetype=2,996                                               account=self.account.name,997                                               domainid=self.account.domainid,998                                               max=1999                                               )1000        list_resource = Resources.list(self.apiclient,1001                                       account=self.account.name,1002                                       domainid=self.account.domainid,1003                                       resourcetype=21004                                       )1005        self.assertEqual(1006            validateList(list_resource)[0],1007            PASS,1008            "Check List resource response for volume %s" %1009            self.account.name)1010        self.assertEqual(1011            str(1012                list_resource[0].max),1013            '1',1014            "check list List resource response for account id:  %s" %1015            self.account.name)1016        self.debug(1017            "Max resources got updated successfully for account %s" %1018            self.account.name)1019        try:1020            self.upload_response = Volume.upload(1021                self.userapiclient,1022                zoneid=self.zone.id,1023                url=self.extract_volume.url,1024                services=self.testdata["configurableData"]["upload_volume"])1025            self.upload_response.wait_for_upload(self.userapiclient1026                                                 )1027            self.fail("volume got uploaded after account reached max limit for\1028                      volumes ")1029        except Exception as e:...
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.