KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use wait_for_caching method in tempest

Best Python code snippet using tempest_python

utils.py

Source:utils.py

...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:...

test_cache_api.py

Source:test_cache_api.py

...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']))...

test_image_caching.py

Source:test_image_caching.py

...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...

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.