Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use stage_image_file method in tempest

Best Python code snippet using tempest_python

test_images.py

Source:test_images.py

...67        image = self._create_image()68        # Stage image data69        file_content = data_utils.random_bytes()70        image_file = io.BytesIO(file_content)71        self.client.stage_image_file(image['id'], image_file)72        # Check image status is 'uploading'73        body = self.client.show_image(image['id'])74        self.assertEqual(image['id'], body['id'])75        self.assertEqual('uploading', body['status'])76        return image['id']77    @decorators.idempotent_id('32ca0c20-e16f-44ac-8590-07869c9b4cc2')78    def test_image_glance_direct_import(self):79        """Test 'glance-direct' import functionalities80        Create image, stage image data, import image and verify81        that import succeeded.82        """83        self._require_import_method('glance-direct')84        image_id = self._stage_and_check()85        # import image from staging to backend86        resp = self.client.image_import(image_id, method='glance-direct')87        waiters.wait_for_image_imported_to_stores(self.client, image_id)88        if not self.versions_client.has_version('2.12'):89            # API is not new enough to support image/tasks API90            LOG.info('Glance does not support v2.12, so I am unable to '91                     'validate the image/tasks API.')92            return93        tasks = waiters.wait_for_image_tasks_status(94            self.client, image_id, 'success')95        self.assertEqual(1, len(tasks))96        task = tasks[0]97        self.assertEqual(resp.response['x-openstack-request-id'],98                         task['request_id'])99        self.assertEqual('glance-direct',100                         task['input']['import_req']['method']['name'])101    @decorators.idempotent_id('f6feb7a4-b04f-4706-a011-206129f83e62')102    def test_image_web_download_import(self):103        """Test 'web-download' import functionalities104        Create image, import image and verify that import105        succeeded.106        """107        self._require_import_method('web-download')108        image = self._create_image()109        # Now try to get image details110        body = self.client.show_image(image['id'])111        self.assertEqual(image['id'], body['id'])112        self.assertEqual('queued', body['status'])113        # import image from web to backend114        image_uri = CONF.image.http_image115        self.client.image_import(image['id'], method='web-download',116                                 import_params={'uri': image_uri})117        waiters.wait_for_image_imported_to_stores(self.client, image['id'])118    @decorators.idempotent_id('e04761a1-22af-42c2-b8bc-a34a3f12b585')119    def test_remote_import(self):120        """Test image import against a different worker than stage.121        This creates and stages an image against the primary API worker,122        but then calls import on a secondary worker (if available) to123        test that distributed image import works (i.e. proxies the import124        request to the proper worker).125        """126        self._require_import_method('glance-direct')127        if not CONF.image.alternate_image_endpoint:128            raise self.skipException('No image_remote service to test '129                                     'against')130        image_id = self._stage_and_check()131        # import image from staging to backend, but on the alternate worker132        self.os_primary.image_client_remote.image_import(133            image_id, method='glance-direct')134        waiters.wait_for_image_imported_to_stores(self.client, image_id)135    @decorators.idempotent_id('44d60544-1524-42f7-8899-315301105dd8')136    def test_remote_delete(self):137        """Test image delete against a different worker than stage.138        This creates and stages an image against the primary API worker,139        but then calls delete on a secondary worker (if available) to140        test that distributed image import works (i.e. proxies the delete141        request to the proper worker).142        """143        self._require_import_method('glance-direct')144        if not CONF.image.alternate_image_endpoint:145            raise self.skipException('No image_remote service to test '146                                     'against')147        image_id = self._stage_and_check()148        # delete image from staging to backend, but on the alternate worker149        self.os_primary.image_client_remote.delete_image(image_id)150        self.client.wait_for_resource_deletion(image_id)151class MultiStoresImportImagesTest(base.BaseV2ImageTest):152    """Test importing image in multiple stores"""153    @classmethod154    def skip_checks(cls):155        super(MultiStoresImportImagesTest, cls).skip_checks()156        if not CONF.image_feature_enabled.import_image:157            skip_msg = (158                "%s skipped as image import is not available" % cls.__name__)159            raise cls.skipException(skip_msg)160    @classmethod161    def resource_setup(cls):162        super(MultiStoresImportImagesTest, cls).resource_setup()163        cls.available_import_methods = cls.client.info_import()[164            'import-methods']['value']165        if not cls.available_import_methods:166            raise cls.skipException('Server does not support '167                                    'any import method')168        # NOTE(pdeore): Skip if glance-direct import method and mutlistore169        # are not enabled/configured, or only one store is configured in170        # multiple stores setup.171        cls.available_stores = cls.get_available_stores()172        if ('glance-direct' not in cls.available_import_methods or173                not len(cls.available_stores) > 1):174            raise cls.skipException(175                'Either glance-direct import method not present in %s or '176                'None or only one store is '177                'configured %s' % (cls.available_import_methods,178                                   cls.available_stores))179    def _create_and_stage_image(self, all_stores=False):180        """Create Image & stage image file for glance-direct import method."""181        image_name = data_utils.rand_name('test-image')182        container_format = CONF.image.container_formats[0]183        disk_format = CONF.image.disk_formats[0]184        image = self.create_image(name=image_name,185                                  container_format=container_format,186                                  disk_format=disk_format,187                                  visibility='private')188        self.assertEqual('queued', image['status'])189        self.client.stage_image_file(190            image['id'],191            io.BytesIO(data_utils.random_bytes()))192        # Check image status is 'uploading'193        body = self.client.show_image(image['id'])194        self.assertEqual(image['id'], body['id'])195        self.assertEqual('uploading', body['status'])196        if all_stores:197            stores_list = ','.join([store['id']198                                    for store in self.available_stores])199        else:200            stores = [store['id'] for store in self.available_stores]201            stores_list = stores[::len(stores) - 1]202        return body, stores_list203    @decorators.idempotent_id('bf04ff00-3182-47cb-833a-f1c6767b47fd')...

test_unified_limits.py

Source:test_unified_limits.py

...170        image1 = self.create_image(name='first',171                                   container_format='bare',172                                   disk_format='raw',173                                   visibility='private')174        self.image_client.stage_image_file(image1['id'],175                                           io.BytesIO(file_content))176        # Check that we can not stage another177        image2 = self.create_image(name='second',178                                   container_format='bare',179                                   disk_format='raw',180                                   visibility='private')181        self.assertRaises(lib_exc.OverLimit,182                          self.image_client.stage_image_file,183                          image2['id'], io.BytesIO(file_content))184        # ... nor upload directly185        image3 = self.create_image(name='third',186                                   container_format='bare',187                                   disk_format='raw',188                                   visibility='private')189        self.assertRaises(lib_exc.OverLimit,190                          self.image_client.store_image_file,191                          image3['id'],192                          io.BytesIO(file_content))193        # Update our quota to make room194        self._update_limit('image_count_uploading', 2)195        # Now our upload should work196        self.image_client.store_image_file(image3['id'],197                                           io.BytesIO(file_content))198        # ...and because that is no longer in uploading state, we should be199        # able to stage our second image from above.200        self.image_client.stage_image_file(image2['id'],201                                           io.BytesIO(file_content))202        # Finish our import of image2203        self.image_client.image_import(image2['id'], method='glance-direct')204        waiters.wait_for_image_imported_to_stores(self.image_client,205                                                  image2['id'])206        # Set our quota back to one207        self._update_limit('image_count_uploading', 1)208        # Since image1 is still staged, we should not be able to upload209        # an image.210        image4 = self.create_image(name='fourth',211                                   container_format='bare',212                                   disk_format='raw',213                                   visibility='private')214        self.assertRaises(lib_exc.OverLimit,215                          self.image_client.store_image_file,216                          image4['id'],217                          io.BytesIO(file_content))218        # Finish our import of image1 to make space in our uploading quota.219        self.image_client.image_import(image1['id'], method='glance-direct')220        waiters.wait_for_image_imported_to_stores(self.image_client,221                                                  image1['id'])222        # Make sure that freed up the one upload quota to complete our upload223        self.image_client.store_image_file(image4['id'],224                                           io.BytesIO(file_content))225        # Delete all the images we created before the next test runs,226        # so that it starts with full quota.227        self._cleanup_images()228    @decorators.idempotent_id('05e8d064-c39a-4801-8c6a-465df375ec5b')229    @utils.services('image', 'identity')230    def test_image_size_quota(self):231        self.check_quotas_enabled()232        # Set a quota on the image size for our tenant to 1MiB, and allow ten233        # images.234        self._update_limit('image_size_total', 1)235        self._update_limit('image_count_total', 10)236        self._update_limit('image_count_uploading', 10)237        file_content = data_utils.random_bytes(1 * units.Mi)238        # Create and upload a 1MiB image.239        image1 = self.create_image(name='first',240                                   container_format='bare',241                                   disk_format='raw',242                                   visibility='private')243        self.image_client.store_image_file(image1['id'],244                                           io.BytesIO(file_content))245        # Create and upload a second 1MiB image. This succeeds, but246        # after completion, we are over quota. Despite us being at247        # quota above, the initial quota check for the second248        # operation has no idea what the image size will be, and thus249        # uses delta=0. This will succeed because we're not250        # technically over-quota and have not asked for any more (this251        # is oslo.limit behavior). After the second operation,252        # however, we will be over-quota regardless of the delta and253        # subsequent attempts will fail. Because glance goes not254        # require an image size to be declared before upload, this is255        # really the best it can do without an API change.256        image2 = self.create_image(name='second',257                                   container_format='bare',258                                   disk_format='raw',259                                   visibility='private')260        self.image_client.store_image_file(image2['id'],261                                           io.BytesIO(file_content))262        # Create and attempt to upload a third 1MiB image. This should fail to263        # upload (but not create) because we are over quota.264        image3 = self.create_image(name='third',265                                   container_format='bare',266                                   disk_format='raw',267                                   visibility='private')268        self.assertRaises(lib_exc.OverLimit,269                          self.image_client.store_image_file,270                          image3['id'], io.BytesIO(file_content))271        # Increase our size quota to 2MiB.272        self._update_limit('image_size_total', 2)273        # Now the upload of the already-created image is allowed, but274        # after completion, we are over quota again.275        self.image_client.store_image_file(image3['id'],276                                           io.BytesIO(file_content))277        # Create and attempt to upload a fourth 1MiB image. This should278        # fail to upload (but not create) because we are over quota.279        image4 = self.create_image(name='fourth',280                                   container_format='bare',281                                   disk_format='raw',282                                   visibility='private')283        self.assertRaises(lib_exc.OverLimit,284                          self.image_client.store_image_file,285                          image4['id'], io.BytesIO(file_content))286        # Delete our first image to make space in our existing 2MiB quota.287        self.image_client.delete_image(image1['id'])288        # Now the upload of the already-created image is allowed.289        self.image_client.store_image_file(image4['id'],290                                           io.BytesIO(file_content))291        # Delete all the images we created before the next test runs,292        # so that it starts with full quota.293        self._cleanup_images()294    @decorators.idempotent_id('fc76b8d9-aae5-46fb-9285-099e37f311f7')295    @utils.services('image', 'identity')296    def test_image_stage_quota(self):297        if not CONF.image_feature_enabled.import_image:298            skip_msg = (299                "%s skipped as image import is not available" % __name__)300            raise self.skipException(skip_msg)301        self.check_quotas_enabled()302        # Create a staging quota of 1MiB, allow 10MiB of active303        # images, and a total of ten images.304        self._update_limit('image_stage_total', 1)305        self._update_limit('image_size_total', 10)306        self._update_limit('image_count_total', 10)307        self._update_limit('image_count_uploading', 10)308        file_content = data_utils.random_bytes(1 * units.Mi)309        # Create and stage a 1MiB image.310        image1 = self.create_image(name='first',311                                   container_format='bare',312                                   disk_format='raw',313                                   visibility='private')314        self.image_client.stage_image_file(image1['id'],315                                           io.BytesIO(file_content))316        # Create and stage a second 1MiB image. This succeeds, but317        # after completion, we are over quota.318        image2 = self.create_image(name='second',319                                   container_format='bare',320                                   disk_format='raw',321                                   visibility='private')322        self.image_client.stage_image_file(image2['id'],323                                           io.BytesIO(file_content))324        # Create and attempt to stage a third 1MiB image. This should fail to325        # stage (but not create) because we are over quota.326        image3 = self.create_image(name='third',327                                   container_format='bare',328                                   disk_format='raw',329                                   visibility='private')330        self.assertRaises(lib_exc.OverLimit,331                          self.image_client.stage_image_file,332                          image3['id'], io.BytesIO(file_content))333        # Make sure that even though we are over our stage quota, we334        # can still create and upload an image the regular way.335        image_upload = self.create_image(name='uploaded',336                                         container_format='bare',337                                         disk_format='raw',338                                         visibility='private')339        self.image_client.store_image_file(image_upload['id'],340                                           io.BytesIO(file_content))341        # Increase our stage quota to two MiB.342        self._update_limit('image_stage_total', 2)343        # Now the upload of the already-created image is allowed, but344        # after completion, we are over quota again.345        self.image_client.stage_image_file(image3['id'],346                                           io.BytesIO(file_content))347        # Create and attempt to stage a fourth 1MiB image. This should348        # fail to stage (but not create) because we are over quota.349        image4 = self.create_image(name='fourth',350                                   container_format='bare',351                                   disk_format='raw',352                                   visibility='private')353        self.assertRaises(lib_exc.OverLimit,354                          self.image_client.stage_image_file,355                          image4['id'], io.BytesIO(file_content))356        # Finish our import of image1 to make space in our stage quota.357        self.image_client.image_import(image1['id'], method='glance-direct')358        waiters.wait_for_image_imported_to_stores(self.image_client,359                                                  image1['id'])360        # Now the upload of the already-created image is allowed.361        self.image_client.stage_image_file(image4['id'],362                                           io.BytesIO(file_content))363        # Delete all the images we created before the next test runs,364        # so that it starts with full quota....

images_client.py

Source:images_client.py

...137        resp, body = self.request('PUT', url, headers=headers,138                                  body=data, chunked=True)139        self.expected_success(204, resp.status)140        return rest_client.ResponseBody(resp, body)141    def stage_image_file(self, image_id, data):142        """Upload binary image data to staging area.143        For a full list of available parameters, please refer to the official144        API reference (stage API:145        https://docs.openstack.org/api-ref/image/v2/#interoperable-image-import146        """147        url = 'images/%s/stage' % image_id148        # We are going to do chunked transfer, so split the input data149        # info fixed-sized chunks.150        headers = {'Content-Type': 'application/octet-stream'}151        data = iter(functools.partial(data.read, CHUNKSIZE), b'')152        resp, body = self.request('PUT', url, headers=headers,153                                  body=data, chunked=True)154        self.expected_success(204, resp.status)155        return rest_client.ResponseBody(resp, body)...

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.