Best Python code snippet using tempest_python
test_volumes_extend.py
Source:test_volumes_extend.py
...81 for event in action['events']:82 if (event['event'] == 'compute_extend_volume' and83 event['finish_time']):84 return event85 def _test_extend_attached_volume(self, volume):86 """This is a happy path test which does the following:87 * Create a server instance.88 * Attach the volume to the server.89 * Wait for the volume status to be "in-use".90 * Extend the size of the volume and wait for the volume status to go91 back to "in-use".92 * Assert the volume size change is reflected in the volume API.93 * Wait for the "compute_extend_volume" instance action event to show94 up in the compute API with the success or failure status. We fail95 if we timeout waiting for the instance action event to show up, or96 if the action on the server fails.97 """98 # Create a test server. Will be automatically cleaned up on teardown.99 server = self.create_server()100 # Attach the volume to the server and wait for the volume status to be101 # "in-use".102 self.attach_volume(server['id'], volume['id'])103 # Extend the size of the volume. If this is successful, the volume API104 # will change the status on the volume to "extending" before doing an105 # RPC cast to the volume manager on the backend. Note that we multiply106 # the size of the volume since certain Cinder backends, e.g. ScaleIO,107 # require multiples of 8GB.108 extend_size = volume['size'] * 2109 self.volumes_client.extend_volume(volume['id'], new_size=extend_size)110 # The volume status should go back to in-use since it is still attached111 # to the server instance.112 waiters.wait_for_volume_resource_status(self.volumes_client,113 volume['id'], 'in-use')114 # Assert that the volume size has changed in the volume API.115 volume = self.volumes_client.show_volume(volume['id'])['volume']116 self.assertEqual(extend_size, volume['size'])117 # Now we wait for the "compute_extend_volume" instance action event118 # to show up for the server instance. This is our indication that the119 # asynchronous operation is complete on the compute side.120 start_time = int(time.time())121 timeout = self.servers_client.build_timeout122 action = self._find_extend_volume_instance_action(server['id'])123 while action is None and int(time.time()) - start_time < timeout:124 time.sleep(self.servers_client.build_interval)125 action = self._find_extend_volume_instance_action(server['id'])126 if action is None:127 msg = ("Timed out waiting to get 'extend_volume' instance action "128 "record for server %(server)s after %(timeout)s seconds." %129 {'server': server['id'], 'timeout': timeout})130 raise lib_exc.TimeoutException(msg)131 # Now that we found the extend_volume instance action, we can wait for132 # the compute_extend_volume instance action event to show up to133 # indicate the operation is complete.134 start_time = int(time.time())135 event = self._find_extend_volume_instance_action_finish_event(action)136 while event is None and int(time.time()) - start_time < timeout:137 time.sleep(self.servers_client.build_interval)138 event = self._find_extend_volume_instance_action_finish_event(139 action)140 if event is None:141 msg = ("Timed out waiting to get 'compute_extend_volume' instance "142 "action event record for server %(server)s and request "143 "%(request_id)s after %(timeout)s seconds." %144 {'server': server['id'],145 'request_id': action['request_id'],146 'timeout': timeout})147 raise lib_exc.TimeoutException(msg)148 # Finally, assert that the action completed successfully.149 self.assertTrue(150 event['result'].lower() == 'success',151 "Unexpected compute_extend_volume result '%(result)s' for request "152 "%(request_id)s." %153 {'result': event['result'],154 'request_id': action['request_id']})155class VolumesExtendAttachedTest(BaseVolumesExtendAttachedTest):156 @decorators.idempotent_id('301f5a30-1c6f-4ea0-be1a-91fd28d44354')157 @testtools.skipUnless(CONF.volume_feature_enabled.extend_attached_volume,158 "Attached volume extend is disabled.")159 @utils.services('compute')160 def test_extend_attached_volume(self):161 volume = self.create_volume()...
test_encrypted_volumes_extend.py
Source:test_encrypted_volumes_extend.py
...26 @utils.services('compute')27 def test_extend_attached_encrypted_volume_luksv1(self):28 """LUKs v1 decrypts and extends through libvirt."""29 volume = self.create_encrypted_volume(encryption_provider="luks")30 self._test_extend_attached_volume(volume)31 @decorators.idempotent_id('381a2a3a-b2f4-4631-a910-720881f2cc2f')32 @testtools.skipUnless(33 CONF.volume_feature_enabled.extend_attached_encrypted_volume,34 "Attached encrypted volume extend is disabled.")35 @testtools.skipIf(CONF.volume.storage_protocol == 'ceph',36 'Ceph only supports LUKSv2 if doing host attach.')37 @utils.services('compute')38 def test_extend_attached_encrypted_volume_luksv2(self):39 """LUKs v2 decrypts and extends through os-brick."""40 volume = self.create_encrypted_volume(encryption_provider="luks2")...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!