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

How to use glance method in tempest

Best Python code snippet using tempest_python

glance.py

Source:glance.py

...119        image_metas = self._get_images(context, **params)120        images = []121        for image_meta in image_metas:122            if self._is_image_available(context, image_meta):123                base_image_meta = self._translate_from_glance(image_meta)124                images.append(base_image_meta)125        return images126    def _extract_query_params(self, params):127        _params = {}128        accepted_params = ('filters', 'marker', 'limit',129                           'sort_key', 'sort_dir')130        for param in accepted_params:131            if param in params:132                _params[param] = params.get(param)133        return _params134    def _get_images(self, context, **kwargs):135        """Get image entitites from images service"""136        # ensure filters is a dict137        kwargs['filters'] = kwargs.get('filters') or {}138        # NOTE(vish): don't filter out private images139        kwargs['filters'].setdefault('is_public', 'none')140        client = self._get_client(context)141        return self._fetch_images(client.get_images_detailed, **kwargs)142    def _fetch_images(self, fetch_func, **kwargs):143        """Paginate through results from glance server"""144        images = fetch_func(**kwargs)145        if not images:146            # break out of recursive loop to end pagination147            return148        for image in images:149            yield image150        try:151            # attempt to advance the marker in order to fetch next page152            kwargs['marker'] = images[-1]['id']153        except KeyError:154            raise exception.ImagePaginationFailed()155        try:156            kwargs['limit'] = kwargs['limit'] - len(images)157            # break if we have reached a provided limit158            if kwargs['limit'] <= 0:159                return160        except KeyError:161            # ignore missing limit, just proceed without it162            pass163        for image in self._fetch_images(fetch_func, **kwargs):164            yield image165    def show(self, context, image_id):166        """Returns a dict with image data for the given opaque image id."""167        try:168            image_meta = self._get_client(context).get_image_meta(image_id)169        except glance_exception.NotFound:170            raise exception.ImageNotFound(image_id=image_id)171        if not self._is_image_available(context, image_meta):172            raise exception.ImageNotFound(image_id=image_id)173        base_image_meta = self._translate_from_glance(image_meta)174        return base_image_meta175    def show_by_name(self, context, name):176        """Returns a dict containing image data for the given name."""177        # TODO(vish): replace this with more efficient call when glance178        #             supports it.179        image_metas = self.detail(context)180        for image_meta in image_metas:181            if name == image_meta.get('name'):182                return image_meta183        raise exception.ImageNotFound(image_id=name)184    def get(self, context, image_id, data):185        """Calls out to Glance for metadata and data and writes data."""186        num_retries = FLAGS.glance_num_retries187        for count in xrange(1 + num_retries):188            client = self._get_client(context)189            try:190                image_meta, image_chunks = client.get_image(image_id)191                break192            except glance_exception.NotFound:193                raise exception.ImageNotFound(image_id=image_id)194            except Exception:195                if count == num_retries:196                    raise197            time.sleep(1)198        for chunk in image_chunks:199            data.write(chunk)200        base_image_meta = self._translate_from_glance(image_meta)201        return base_image_meta202    def create(self, context, image_meta, data=None):203        """Store the image data and return the new image id.204        :raises: AlreadyExists if the image already exist.205        """206        # Translate Base -> Service207        LOG.debug(_('Creating image in Glance. Metadata passed in %s'),208                  image_meta)209        sent_service_image_meta = self._translate_to_glance(image_meta)210        LOG.debug(_('Metadata after formatting for Glance %s'),211                  sent_service_image_meta)212        recv_service_image_meta = self._get_client(context).add_image(213            sent_service_image_meta, data)214        # Translate Service -> Base215        base_image_meta = self._translate_from_glance(recv_service_image_meta)216        LOG.debug(_('Metadata returned from Glance formatted for Base %s'),217                  base_image_meta)218        return base_image_meta219    def update(self, context, image_id, image_meta, data=None):220        """Replace the contents of the given image with the new data.221        :raises: ImageNotFound if the image does not exist.222        """223        # NOTE(vish): show is to check if image is available224        self.show(context, image_id)225        image_meta = self._translate_to_glance(image_meta)226        try:227            client = self._get_client(context)228            image_meta = client.update_image(image_id, image_meta, data)229        except glance_exception.NotFound:230            raise exception.ImageNotFound(image_id=image_id)231        base_image_meta = self._translate_from_glance(image_meta)232        return base_image_meta233    def delete(self, context, image_id):234        """Delete the given image.235        :raises: ImageNotFound if the image does not exist.236        :raises: NotAuthorized if the user is not an owner.237        """238        # NOTE(vish): show is to check if image is available239        image_meta = self.show(context, image_id)240        if FLAGS.use_deprecated_auth:241            # NOTE(parthi): only allow image deletions if the user242            # is a member of the project owning the image, in case of243            # setup without keystone244            # TODO Currently this access control breaks if245            # 1. Image is not owned by a project246            # 2. Deleting user is not bound a project247            properties = image_meta['properties']248            if (context.project_id and ('project_id' in properties)249                and (context.project_id != properties['project_id'])):250                raise exception.NotAuthorized(_("Not the image owner"))251            if (context.project_id and ('owner_id' in properties)252                and (context.project_id != properties['owner_id'])):253                raise exception.NotAuthorized(_("Not the image owner"))254        try:255            result = self._get_client(context).delete_image(image_id)256        except glance_exception.NotFound:257            raise exception.ImageNotFound(image_id=image_id)258        return result259    def delete_all(self):260        """Clears out all images."""261        pass262    @classmethod263    def _translate_to_glance(cls, image_meta):264        image_meta = _convert_to_string(image_meta)265        image_meta = _remove_read_only(image_meta)266        return image_meta267    @classmethod268    def _translate_from_glance(cls, image_meta):269        image_meta = _limit_attributes(image_meta)270        image_meta = _convert_timestamps_to_datetimes(image_meta)271        image_meta = _convert_from_string(image_meta)272        return image_meta273    @staticmethod274    def _is_image_available(context, image_meta):275        """Check image availability.276        Under Glance, images are always available if the context has277        an auth_token.278        """279        if hasattr(context, 'auth_token') and context.auth_token:280            return True281        if image_meta['is_public'] or context.is_admin:282            return True...

vmware_images.py

Source:vmware_images.py

1# vim: tabstop=4 shiftwidth=4 softtabstop=423# Copyright (c) 2011 Citrix Systems, Inc.4# Copyright 2011 OpenStack LLC.5#6#    Licensed under the Apache License, Version 2.0 (the "License"); you may7#    not use this file except in compliance with the License. You may obtain8#    a copy of the License at9#10#         http://www.apache.org/licenses/LICENSE-2.011#12#    Unless required by applicable law or agreed to in writing, software13#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT14#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the15#    License for the specific language governing permissions and limitations16#    under the License.17"""18Utility functions for Image transfer.19"""2021from nova import exception22from nova.image import glance23from nova import log as logging24from nova.virt.vmwareapi import io_util25from nova.virt.vmwareapi import read_write_util2627LOG = logging.getLogger("nova.virt.vmwareapi.vmware_images")2829QUEUE_BUFFER_SIZE = 10303132def start_transfer(read_file_handle, data_size, write_file_handle=None,33                   glance_client=None, image_id=None, image_meta=None):34    """Start the data transfer from the reader to the writer.35    Reader writes to the pipe and the writer reads from the pipe. This means36    that the total transfer time boils down to the slower of the read/write37    and not the addition of the two times."""3839    if not image_meta:40        image_meta = {}4142    # The pipe that acts as an intermediate store of data for reader to write43    # to and writer to grab from.44    thread_safe_pipe = io_util.ThreadSafePipe(QUEUE_BUFFER_SIZE, data_size)45    # The read thread. In case of glance it is the instance of the46    # GlanceFileRead class. The glance client read returns an iterator47    # and this class wraps that iterator to provide datachunks in calls48    # to read.49    read_thread = io_util.IOThread(read_file_handle, thread_safe_pipe)5051    # In case of Glance - VMWare transfer, we just need a handle to the52    # HTTP Connection that is to send transfer data to the VMWare datastore.53    if write_file_handle:54        write_thread = io_util.IOThread(thread_safe_pipe, write_file_handle)55    # In case of VMWare - Glance transfer, we relinquish VMWare HTTP file read56    # handle to Glance Client instance, but to be sure of the transfer we need57    # to be sure of the status of the image on glnace changing to active.58    # The GlanceWriteThread handles the same for us.59    elif glance_client and image_id:60        write_thread = io_util.GlanceWriteThread(thread_safe_pipe,61                                         glance_client, image_id, image_meta)62    # Start the read and write threads.63    read_event = read_thread.start()64    write_event = write_thread.start()65    try:66        # Wait on the read and write events to signal their end67        read_event.wait()68        write_event.wait()69    except Exception, exc:70        # In case of any of the reads or writes raising an exception,71        # stop the threads so that we un-necessarily don't keep the other one72        # waiting.73        read_thread.stop()74        write_thread.stop()7576        # Log and raise the exception.77        LOG.exception(exc)78        raise exception.Error(exc)79    finally:80        # No matter what, try closing the read and write handles, if it so81        # applies.82        read_file_handle.close()83        if write_file_handle:84            write_file_handle.close()858687def fetch_image(context, image, instance, **kwargs):88    """Download image from the glance image server."""89    LOG.debug(_("Downloading image %s from glance image server") % image)90    (glance_client, image_id) = glance.get_glance_client(context, image)91    metadata, read_iter = glance_client.get_image(image_id)92    read_file_handle = read_write_util.GlanceFileRead(read_iter)93    file_size = int(metadata['size'])94    write_file_handle = read_write_util.VMWareHTTPWriteFile(95                                kwargs.get("host"),96                                kwargs.get("data_center_name"),97                                kwargs.get("datastore_name"),98                                kwargs.get("cookies"),99                                kwargs.get("file_path"),100                                file_size)101    start_transfer(read_file_handle, file_size,102                   write_file_handle=write_file_handle)103    LOG.debug(_("Downloaded image %s from glance image server") % image)104105106def upload_image(context, image, instance, **kwargs):107    """Upload the snapshotted vm disk file to Glance image server."""108    LOG.debug(_("Uploading image %s to the Glance image server") % image)109    read_file_handle = read_write_util.VmWareHTTPReadFile(110                                kwargs.get("host"),111                                kwargs.get("data_center_name"),112                                kwargs.get("datastore_name"),113                                kwargs.get("cookies"),114                                kwargs.get("file_path"))115    file_size = read_file_handle.get_size()116    (glance_client, image_id) = glance.get_glance_client(context, image)117    # The properties and other fields that we need to set for the image.118    image_metadata = {"is_public": True,119                      "disk_format": "vmdk",120                      "container_format": "bare",121                      "type": "vmdk",122                      "properties": {"vmware_adaptertype":123                                            kwargs.get("adapter_type"),124                                     "vmware_ostype": kwargs.get("os_type"),125                                     "vmware_image_version":126                                            kwargs.get("image_version")}}127    start_transfer(read_file_handle, file_size, glance_client=glance_client,128                        image_id=image_id, image_meta=image_metadata)129    LOG.debug(_("Uploaded image %s to the Glance image server") % image)130131132def get_vmdk_size_and_properties(context, image, instance):133    """134    Get size of the vmdk file that is to be downloaded for attach in spawn.135    Need this to create the dummy virtual disk for the meta-data file. The136    geometry of the disk created depends on the size.137    """138139    LOG.debug(_("Getting image size for the image %s") % image)140    (glance_client, image_id) = glance.get_glance_client(context, image)141    meta_data = glance_client.get_image_meta(image_id)142    size, properties = meta_data["size"], meta_data["properties"]143    LOG.debug(_("Got image size of %(size)s for the image %(image)s") %144              locals())
...

test.py

Source:test.py

1# import os2# os.makedirs('glance/api')3# os.makedirs('glance/cmd')4# os.makedirs('glance/db')5# l = []6# l.append(open('glance/__init__.py','w'))7# l.append(open('glance/api/__init__.py','w'))8# l.append(open('glance/api/policy.py','w'))9# l.append(open('glance/api/versions.py','w'))10# l.append(open('glance/cmd/__init__.py','w'))11# l.append(open('glance/cmd/manage.py','w'))12# l.append(open('glance/db/__init__.py','w'))13# l.append(open('glance/db/models.py','w'))14# map(lambda f:f.close() ,l)15# å¯ä»¥å¯¼å¥glanceè¿ä¸ªå16# import glance   # å¨å¯¼å¥ä¸ä¸ªåçæ¶å. é»è®¤æ§è¡çæ¯è¿ä¸ªåä¸ç__init__.py17# import master  # master.py18# glance.api.policy.get()  # ä¸å¯ä»¥è¿æ ·ç´æ¥ä½¿ç¨19# import glance.api.policy20#21# glance.api.policy.get()22# è¿ç§æ¹æ¡æ¯æä»¬ä½¿ç¨åä½¿ç¨æå¤ç23# from urllib.request import Request24from glance.api import policy25policy.func2()26# # éè¯¯ç¤ºè       å¨ä½¿ç¨fromè¿ç§å½¢å¼çå¯¼å. importåé¢ä¸åè®¸åºç°.27# from glance import api.policy28# api.policy.get()29# from glance.api import policy30#31# policy.get()...

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.