How to use wait_for_caching method in tempest

Best Python code snippet using tempest_python

utils.py

Source:utils.py Github

copy

Full Screen

...354 else:355 print_dict(image)356CACHE_STATUS_CACHED = "Cached"357CACHE_STATUS_ERROR = "Error"358def wait_for_caching(interval, gc, id):359 """Wait until image caching has finished."""360 timeout = time.time() + float(interval)361 while True:362 # Obtain image metadata.363 image = gc.images.get(id)364 # If there is no caching information, return.365 cache_status = getattr(image, "cache_raw_status", "")366 if not cache_status:367 print('INFO: There is no information available about the '368 'caching status.')369 break370 # If the caching status is 'Cached' or 'Error', return.371 if cache_status == CACHE_STATUS_CACHED or \372 cache_status == CACHE_STATUS_ERROR:...

Full Screen

Full Screen

test_cache_api.py

Source:test_cache_api.py Github

copy

Full Screen

...86 if target not in ('', 'cache', 'queue'):87 self.assertEqual(expected_code, response.status_code)88 else:89 self.assertEqual(expected_code, response.status_code)90 def wait_for_caching(self, image_id, max_sec=10, delay_sec=0.2,91 start_delay_sec=None):92 start_time = time.time()93 done_time = start_time + max_sec94 if start_delay_sec:95 time.sleep(start_delay_sec)96 while time.time() <= done_time:97 output = self.list_cache()['cached_images']98 output = [image['image_id'] for image in output]99 if output and image_id in output:100 return101 time.sleep(delay_sec)102 msg = "Image {0} failed to cached within {1} sec"103 raise Exception(msg.format(image_id, max_sec))104 def test_cache_list(self):105 self.start_server(enable_cache=True)106 images = self.load_data()107 # Ensure that nothing is cached and nothing is queued for caching108 output = self.list_cache()109 self.assertEqual(0, len(output['queued_images']))110 self.assertEqual(0, len(output['cached_images']))111 # Queue 1 image for caching112 self.cache_queue(images['public'])113 output = self.list_cache()114 self.assertEqual(1, len(output['queued_images']))115 self.assertEqual(0, len(output['cached_images']))116 def test_cache_queue(self):117 self.start_server(enable_cache=True)118 images = self.load_data()119 # Ensure that nothing is cached and nothing is queued for caching120 output = self.list_cache()121 self.assertEqual(0, len(output['queued_images']))122 self.assertEqual(0, len(output['cached_images']))123 # Queue 1 image for caching124 self.cache_queue(images['public'])125 # NOTE(abhishekk): As queue call will immediately start caching126 # the image, lets wait for completion.127 self.wait_for_caching(images['public'])128 # Now verify that we have 1 cached image129 output = self.list_cache()130 self.assertEqual(1, len(output['cached_images']))131 # Verify same image is cached132 self.assertIn(images['public'], output['cached_images'][0]['image_id'])133 def test_cache_delete(self):134 self.start_server(enable_cache=True)135 images = self.load_data()136 # Queue 1 image for caching137 self.cache_queue(images['public'])138 self.wait_for_caching(images['public'])139 # Now verify that we have 1 cached image140 output = self.list_cache()141 self.assertEqual(1, len(output['cached_images']))142 # Verify same image is cached143 self.assertIn(images['public'], output['cached_images'][0]['image_id'])144 # Delete cached image145 self.cache_delete(images['public'])146 # Now verify that we have 0 cached image147 output = self.list_cache()148 self.assertEqual(0, len(output['cached_images']))149 def test_cache_clear_queued_images(self):150 self.start_server(enable_cache=True)151 images = self.load_data()152 # Queue 2 images for caching153 self.cache_queue(images['public'])154 self.cache_queue(images['private'])155 # Now verify that we have 2 queued images156 # NOTE(abhishekk): We might fail with race here as queue call157 # will immediately start caching of an image, so we may not find158 # all images in queued state.159 output = self.list_cache()160 self.assertEqual(2, len(output['queued_images']))161 self.assertEqual(0, len(output['cached_images']))162 # Clear all images from cache163 self.cache_clear(target='queue')164 # Now verify that we have 0 queued images165 output = self.list_cache()166 self.assertEqual(0, len(output['queued_images']))167 self.assertEqual(0, len(output['cached_images']))168 def test_cache_clear_cached_images(self):169 self.start_server(enable_cache=True)170 images = self.load_data()171 # Queue 2 images for caching172 self.cache_queue(images['public'])173 self.cache_queue(images['private'])174 self.wait_for_caching(images['public'])175 self.wait_for_caching(images['private'])176 # Now verify that we have 2 cached images177 output = self.list_cache()178 self.assertEqual(0, len(output['queued_images']))179 self.assertEqual(2, len(output['cached_images']))180 # Clear all images from cache181 self.cache_clear(target='cache')182 # Now verify that we have 0 cached images183 output = self.list_cache()184 self.assertEqual(0, len(output['queued_images']))185 self.assertEqual(0, len(output['cached_images']))186 def test_cache_clear(self):187 self.start_server(enable_cache=True)188 images = self.load_data()189 # Queue 2 images for caching190 self.cache_queue(images['public'])191 self.wait_for_caching(images['public'])192 # Now verify that we have 1 cached images193 output = self.list_cache()194 self.assertEqual(1, len(output['cached_images']))195 self.cache_queue(images['private'])196 # Now verify that we have 1 queued and 1 cached images197 output = self.list_cache()198 # NOTE(abhishekk): We might fail with race here as queue call199 # will immediately start caching of an image, so we may not find200 # image in queued state.201 self.assertEqual(1, len(output['queued_images']))202 self.assertEqual(1, len(output['cached_images']))203 # Clear all images from cache204 self.cache_clear()205 # Now verify that we have 0 queued and cached images206 output = self.list_cache()207 self.assertEqual(0, len(output['queued_images']))208 self.assertEqual(0, len(output['cached_images']))209 def test_cache_api_negative_scenarios(self):210 self.start_server(enable_cache=True)211 images = self.load_data()212 # Try non-existing image to queue for caching213 self.cache_queue('non-existing-image-id', expected_code=404)214 # Verify that you can not queue non-active image215 self.cache_queue(images['queued'], expected_code=400)216 # Try to delete non-existing image from cache217 self.cache_delete('non-existing-image-id', expected_code=404)218 # Verify clearing cache fails with 400 if invalid header is passed219 self.cache_clear(target='both', expected_code=400)220 def test_cache_image_queue_delete(self):221 # This test verifies that if image is queued for caching222 # and user deletes the original image, but it is still223 # present in queued list and deleted with cache-delete API.224 self.start_server(enable_cache=True)225 images = self.load_data()226 # Ensure that nothing is cached and nothing is queued for caching227 output = self.list_cache()228 self.assertEqual(0, len(output['queued_images']))229 self.assertEqual(0, len(output['cached_images']))230 self.cache_queue(images['public'])231 # Now verify that we have 1 image queued for caching and 0232 # cached images233 output = self.list_cache()234 self.assertEqual(1, len(output['queued_images']))235 self.assertEqual(0, len(output['cached_images']))236 # Verify same image is queued for caching237 self.assertIn(images['public'], output['queued_images'])238 # Delete image and verify that it is still present239 # in queued list240 path = '/v2/images/%s' % images['public']241 response = self.api_delete(path)242 self.assertEqual(204, response.status_code)243 output = self.list_cache()244 self.assertEqual(1, len(output['queued_images']))245 self.assertEqual(0, len(output['cached_images']))246 self.assertIn(images['public'], output['queued_images'])247 # Deleted the image from queued list248 self.cache_delete(images['public'])249 output = self.list_cache()250 self.assertEqual(0, len(output['queued_images']))251 self.assertEqual(0, len(output['cached_images']))252 def test_cache_image_cache_delete(self):253 # This test verifies that if image is queued for caching254 # and user deletes the original image, but it is still255 # present in queued list and deleted with cache-delete API.256 self.start_server(enable_cache=True)257 images = self.load_data()258 # Ensure that nothing is cached and nothing is queued for caching259 output = self.list_cache()260 self.assertEqual(0, len(output['queued_images']))261 self.assertEqual(0, len(output['cached_images']))262 self.cache_queue(images['public'])263 # wait for caching the image264 self.wait_for_caching(images['public'])265 # Now verify that we have 0 queued image and 1 cached image266 output = self.list_cache()267 self.assertEqual(0, len(output['queued_images']))268 self.assertEqual(1, len(output['cached_images']))269 # Verify same image cached270 self.assertIn(images['public'], output['cached_images'][0]['image_id'])271 # Delete image and verify that it is deleted from272 # cache as well273 path = '/v2/images/%s' % images['public']274 response = self.api_delete(path)275 self.assertEqual(204, response.status_code)276 output = self.list_cache()277 self.assertEqual(0, len(output['queued_images']))278 self.assertEqual(0, len(output['cached_images']))...

Full Screen

Full Screen

test_image_caching.py

Source:test_image_caching.py Github

copy

Full Screen

...104 self._assertCheckQueues(output['queued_images'])105 self._assertCheckCache(output['cached_images'])106 # Wait for image caching107 LOG.info("Waiting for image %s to get cached", image['id'])108 caching = waiters.wait_for_caching(109 self.client,110 self.os_admin.image_cache_client,111 image['id'])112 self.cached_info[image['id']] = 'cached'113 # verify that we have image in cache and not in queued114 self._assertCheckQueues(caching['queued_images'])115 self._assertCheckCache(caching['cached_images'])116 # Verify that we can delete images from caching and queueing with117 # api call.118 self.os_admin.image_cache_client.cache_clear()119 output = self.os_admin.image_cache_client.list_cache()120 self.assertEqual(0, len(output['queued_images']))121 self.assertEqual(0, len(output['cached_images']))122 # Verify that invalid header value for target returns 400 response...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run tempest automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful