Dieser Inhalt ist in der von Ihnen ausgewählten Sprache nicht verfügbar.

Appendix A. SDK Examples


A.1. Common Functions and Wrappers

Header
__test__ = False

 import config
 import states

 from time import sleep
 import logging
 import ovirtsdk.api
 from ovirtsdk.xml import params
 from ovirtsdk.infrastructure import errors
 from ovirtsdk.infrastructure import contextmanager
 from functools import wraps

 MB = 1024*1024
 GB = 1024*MB

 logging.basicConfig(filename='messages.log',level=logging.DEBUG)
 LOGGER = logging.getLogger(__name__)
 _major = int(config.OVIRT_VERSION[0])
 _minor = int(config.OVIRT_VERSION[2:])
 VERSION = params.Version(major=_major, minor=_minor)
Copy to Clipboard Toggle word wrap
disconnect
def disconnect():
     proxy = contextmanager.get('proxy')
     persistent_auth = contextmanager.get('persistent_auth')
     filter_header = contextmanager.get('filter')

     if proxy and persistent_auth:
         try:
             proxy.request(method='GET',
                         url='/api',
                         headers={'Filter': filter_header},
                         last=True)
         except Exception:
             pass
     contextmanager._clear(force=True)
Copy to Clipboard Toggle word wrap
getApi
def _getApi():
     """ Return ovirtsdk api.

     Will not create another API instance when reloading this module in
     ipython (when common.API is already defined).
     Works around problem when reloading, which would
     otherwise cause the error `ImmutableError: [ERROR]::'proxy' is immutable.`.
     """
     try:
         return API
     except NameError:
         disconnect()
         return ovirtsdk.api.API(
                     url=config.OVIRT_URL, insecure=True,
                     username=config.OVIRT_USERNAME+'@'+config.OVIRT_DOMAIN,
                     password=config.OVIRT_PASSWORD)

 API = _getApi()
Copy to Clipboard Toggle word wrap
waitForState
def waitForState(obj, desiredStates, failStates=None, timeout=config.TIMEOUT,
                     sampling=1, restoreState=None):
     """ Waits for oVirt object to change state using :py:func:`time.sleep`.

     :param obj:             the oVirt object (host, VM, ...) for which to wait
     :param desiredStates:   the desired oVirt object states, accepts both a
                             list of states or a single state
     :param failStates:      fail if the object reaches one of these states
     :param timeout:         (int) time in seconds to wait for desired state
     :param sampling:        (int) how often to check state, in seconds
     :param restoreState     state, tryies to maintentce->up host, when it is non_operational

     :raises AssertionError: when timeout is exceeded and the object still isn't
         in the desired state or if failState was reached

     .. seealso:: :mod:`tests.states`
     """
     # 120 seconds is not enougn to wait for start
     if type(obj).__name__ == 'VM' and desiredStates == states.vm.up:
         timeout = 240

     global res

     if obj is None:
         return
     if type(desiredStates) is not list:
         desiredStates = [desiredStates]
     if type(failStates) is not list and failStates is not None:
         failStates = [failStates]
     elif failStates is None:
         failStates = []

     assert type(obj) is not str, "Bad use of 'waitForState()'"
     t = 0
     state = newState(obj)
     while state not in desiredStates and t <= timeout:
         sleep(sampling)
         t += sampling
         state = newState(obj)
         assert state not in failStates, \
         "Failed to get %s into state '%s' because it reached the fail \
         state '%s'" % (objectDescr(obj), desiredStates, state)
     if t > timeout:
         LOGGER.error("%s didn't reach one of states %s in timout \
                     of %i sec, current state is '%s'"
                     % (objectDescr(obj), str(desiredStates), timeout, state))
     assert state in desiredStates, \
         "Failed to get %s into desired state" % objectDescr(obj)

     if type(obj).__name__ == 'VM':
         disks = obj.get_disks().list()
         if disks is not None:
             for disk in disks:
                 waitForState(disk, states.disk.ok)
Copy to Clipboard Toggle word wrap
newState
def (obj):
     """ Obtain new state of an oVirt object.

     :param obj:         oVirt object (host, VM, storage, ...)
     :return:            (string) the new state of the object

     .. seealso:: :func:`updateObject`, :mod:`tests.states`
     """
     assert type(obj) is not str, "Bad use of 'newState()'"
     updatedObject = updateObject(obj)
     if updatedObject is None:
         LOGGER.warning("Object %s has no status" % (objectDescr(obj)))
         return None

     if type(updatedObject).__name__ == 'VMSnapshot':
         return updatedObject.snapshot_status

     status = updatedObject.status
     if status is None:
         LOGGER.warning("Object %s has no status" % (objectDescr(obj)))
         return None
     return status.state
Copy to Clipboard Toggle word wrap
objectDescr
""" Return ovirt object description

param obj:  ovirt object (glustervolumes, glusterbrick)
:return: (string) ...

typeName = type(obj).__name__
if 'GlusterVolume' in typeName:
  typeName = "Gluster Volume"

return "%s '%s'" % (typeName, obj.name)
Copy to Clipboard Toggle word wrap
22632%2C+Console+Developer+Guide-322-09-2014+17%3A11%3A35Report a bug
Nach oben
Red Hat logoGithubredditYoutubeTwitter

Lernen

Testen, kaufen und verkaufen

Communitys

Über Red Hat Dokumentation

Wir helfen Red Hat Benutzern, mit unseren Produkten und Diensten innovativ zu sein und ihre Ziele zu erreichen – mit Inhalten, denen sie vertrauen können. Entdecken Sie unsere neuesten Updates.

Mehr Inklusion in Open Source

Red Hat hat sich verpflichtet, problematische Sprache in unserem Code, unserer Dokumentation und unseren Web-Eigenschaften zu ersetzen. Weitere Einzelheiten finden Sie in Red Hat Blog.

Über Red Hat

Wir liefern gehärtete Lösungen, die es Unternehmen leichter machen, plattform- und umgebungsübergreifend zu arbeiten, vom zentralen Rechenzentrum bis zum Netzwerkrand.

Theme

© 2025 Red Hat