此内容没有您所选择的语言版本。

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
返回顶部
Red Hat logoGithubredditYoutubeTwitter

学习

尝试、购买和销售

社区

关于红帽文档

通过我们的产品和服务,以及可以信赖的内容,帮助红帽用户创新并实现他们的目标。 了解我们当前的更新.

让开源更具包容性

红帽致力于替换我们的代码、文档和 Web 属性中存在问题的语言。欲了解更多详情,请参阅红帽博客.

關於紅帽

我们提供强化的解决方案,使企业能够更轻松地跨平台和环境(从核心数据中心到网络边缘)工作。

Theme

© 2025 Red Hat