本文整理汇总了Python中wok.xmlutils.utils.xpath_get_text函数的典型用法代码示例。如果您正苦于以下问题:Python xpath_get_text函数的具体用法?Python xpath_get_text怎么用?Python xpath_get_text使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了xpath_get_text函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_ip_xml
def test_ip_xml(self):
"""
Test network ip xml
"""
dhcp_range = {"start": "192.168.122.100", "end": "192.168.122.254"}
params = {}
dhcp = nxml._get_dhcp_elem(**params)
self.assertEquals(None, dhcp)
params["net"] = "192.168.122.0/255.255.255.0"
params["dhcp"] = {'range': dhcp_range}
xml = ET.tostring(nxml._get_ip_elem(**params))
start = xpath_get_text(xml, "/ip/dhcp/range/@start")[0]
end = xpath_get_text(xml, "/ip/dhcp/range/@end")[0]
self.assertEquals(dhcp_range['start'], start)
self.assertEquals(dhcp_range['end'], end)
address = xpath_get_text(xml, "/ip/@address")[0]
netmask = xpath_get_text(xml, "/ip/@netmask")[0]
self.assertEquals(address, params["net"].split("/")[0])
self.assertEquals(netmask, params["net"].split("/")[1])
# test _get_ip_xml can accepts strings: '192.168.122.0/24',
# which is same as "192.168.122.0/255.255.255.0"
params["net"] = "192.168.122.0/24"
xml = ET.tostring(nxml._get_ip_elem(**params))
netmask = xpath_get_text(xml, "/ip/@netmask")[0]
self.assertEquals(netmask,
str(ipaddr.IPNetwork(params["net"]).netmask))
开发者ID:madhawa,项目名称:kimchi,代码行数:30,代码来源:test_networkxml.py
示例2: test_dhcp_xml
def test_dhcp_xml(self):
"""
Test network dhcp xml
"""
dhcp_range = {'start': '192.168.122.100', 'end': '192.168.122.254'}
host1 = {
'mac': '00:16:3e:77:e2:ed',
'name': 'foo.example.com',
'ip': '192.168.122.10',
}
host2 = {
'mac': '00:16:3e:3e:a9:1a',
'name': 'bar.example.com',
'ip': '192.168.122.11',
}
params = {}
dhcp = nxml._get_dhcp_elem(**params)
self.assertEqual(None, dhcp)
params['range'] = dhcp_range
xml = ET.tostring(nxml._get_dhcp_elem(**params))
start = xpath_get_text(xml, '/dhcp/range/@start')
end = xpath_get_text(xml, '/dhcp/range/@end')
self.assertEqual(dhcp_range['start'], start[0])
self.assertEqual(dhcp_range['end'], end[0])
params['hosts'] = [host1, host2]
xml = ET.tostring(nxml._get_dhcp_elem(**params))
ip = xpath_get_text(xml, '/dhcp/host/@ip')
self.assertEqual(ip, [host1['ip'], host2['ip']])
开发者ID:alinefm,项目名称:kimchi,代码行数:31,代码来源:test_networkxml.py
示例3: test_ip_xml
def test_ip_xml(self):
"""
Test network ip xml
"""
dhcp_range = {'start': '192.168.122.100', 'end': '192.168.122.254'}
params = {}
dhcp = nxml._get_dhcp_elem(**params)
self.assertEqual(None, dhcp)
params['net'] = '192.168.122.0/255.255.255.0'
params['dhcp'] = {'range': dhcp_range}
xml = ET.tostring(nxml._get_ip_elem(**params))
start = xpath_get_text(xml, '/ip/dhcp/range/@start')[0]
end = xpath_get_text(xml, '/ip/dhcp/range/@end')[0]
self.assertEqual(dhcp_range['start'], start)
self.assertEqual(dhcp_range['end'], end)
address = xpath_get_text(xml, '/ip/@address')[0]
netmask = xpath_get_text(xml, '/ip/@netmask')[0]
self.assertEqual(address, params['net'].split('/')[0])
self.assertEqual(netmask, params['net'].split('/')[1])
# test _get_ip_xml can accepts strings: '192.168.122.0/24',
# which is same as "192.168.122.0/255.255.255.0"
params['net'] = '192.168.122.0/24'
xml = ET.tostring(nxml._get_ip_elem(**params))
netmask = xpath_get_text(xml, '/ip/@netmask')[0]
self.assertEqual(netmask, str(ipaddr.IPNetwork(params['net']).netmask))
开发者ID:alinefm,项目名称:kimchi,代码行数:29,代码来源:test_networkxml.py
示例4: test_dhcp_xml
def test_dhcp_xml(self):
"""
Test network dhcp xml
"""
dhcp_range = {"start": "192.168.122.100", "end": "192.168.122.254"}
host1 = {"mac": "00:16:3e:77:e2:ed",
"name": "foo.example.com",
"ip": "192.168.122.10"}
host2 = {"mac": "00:16:3e:3e:a9:1a",
"name": "bar.example.com",
"ip": "192.168.122.11"}
params = {}
dhcp = nxml._get_dhcp_elem(**params)
self.assertEquals(None, dhcp)
params["range"] = dhcp_range
xml = ET.tostring(nxml._get_dhcp_elem(**params))
start = xpath_get_text(xml, "/dhcp/range/@start")
end = xpath_get_text(xml, "/dhcp/range/@end")
self.assertEquals(dhcp_range['start'], start[0])
self.assertEquals(dhcp_range['end'], end[0])
params["hosts"] = [host1, host2]
xml = ET.tostring(nxml._get_dhcp_elem(**params))
ip = xpath_get_text(xml, "/dhcp/host/@ip")
self.assertEquals(ip, [host1['ip'], host2['ip']])
开发者ID:madhawa,项目名称:kimchi,代码行数:27,代码来源:test_networkxml.py
示例5: get_network_from_xml
def get_network_from_xml(xml):
address = xpath_get_text(xml, "/network/ip/@address")
address = address and address[0] or ""
netmask = xpath_get_text(xml, "/network/ip/@netmask")
netmask = netmask and netmask[0] or ""
net = address and netmask and "/".join([address, netmask]) or ""
dhcp_start = xpath_get_text(xml, "/network/ip/dhcp/range/@start")
dhcp_start = dhcp_start and dhcp_start[0] or ""
dhcp_end = xpath_get_text(xml, "/network/ip/dhcp/range/@end")
dhcp_end = dhcp_end and dhcp_end[0] or ""
dhcp = {"start": dhcp_start, "end": dhcp_end}
forward_mode = xpath_get_text(xml, "/network/forward/@mode")
forward_mode = forward_mode and forward_mode[0] or ""
forward_if = xpath_get_text(xml, "/network/forward/interface/@dev")
forward_pf = xpath_get_text(xml, "/network/forward/pf/@dev")
bridge = xpath_get_text(xml, "/network/bridge/@name")
bridge = bridge and bridge[0] or ""
return {
"subnet": net,
"dhcp": dhcp,
"bridge": bridge,
"forward": {"mode": forward_mode, "interface": forward_if, "pf": forward_pf},
}
开发者ID:bingger,项目名称:kimchi,代码行数:25,代码来源:networks.py
示例6: test_to_xml
def test_to_xml(self):
graphics = {'type': 'spice', 'listen': '127.0.0.1'}
vm_uuid = str(uuid.uuid4()).replace('-', '')
t = VMTemplate({'name': 'test-template', 'cdrom': self.iso})
xml = t.to_vm_xml('test-vm', vm_uuid, graphics=graphics)
self.assertEquals(vm_uuid, xpath_get_text(xml, "/domain/uuid")[0])
self.assertEquals('test-vm', xpath_get_text(xml, "/domain/name")[0])
expr = "/domain/devices/graphics/@type"
self.assertEquals(graphics['type'], xpath_get_text(xml, expr)[0])
expr = "/domain/devices/graphics/@listen"
self.assertEquals(graphics['listen'], xpath_get_text(xml, expr)[0])
expr = "/domain/maxMemory/@slots"
self.assertEquals('3', xpath_get_text(xml, expr)[0])
expr = "/domain/maxMemory"
self.assertEquals(str((1024 * 4) << 10), xpath_get_text(xml, expr)[0])
if hasattr(psutil, 'virtual_memory'):
host_memory = psutil.virtual_memory().total >> 10
else:
host_memory = psutil.TOTAL_PHYMEM >> 10
t = VMTemplate({'name': 'test-template', 'cdrom': self.iso,
'memory': (host_memory >> 10) - 512})
xml = t.to_vm_xml('test-vm', vm_uuid, graphics=graphics)
expr = "/domain/maxMemory"
self.assertEquals(str(host_memory), xpath_get_text(xml, expr)[0])
expr = "/domain/maxMemory/@slots"
self.assertEquals('1', xpath_get_text(xml, expr)[0])
开发者ID:nrigaud,项目名称:kimchi,代码行数:27,代码来源:test_vmtemplate.py
示例7: get_network_from_xml
def get_network_from_xml(xml):
address = xpath_get_text(xml, '/network/ip/@address')
address = address and address[0] or ''
netmask = xpath_get_text(xml, '/network/ip/@netmask')
netmask = netmask and netmask[0] or ''
net = address and netmask and '/'.join([address, netmask]) or ''
dhcp_start = xpath_get_text(xml, '/network/ip/dhcp/range/@start')
dhcp_start = dhcp_start and dhcp_start[0] or ''
dhcp_end = xpath_get_text(xml, '/network/ip/dhcp/range/@end')
dhcp_end = dhcp_end and dhcp_end[0] or ''
dhcp = {'start': dhcp_start, 'end': dhcp_end}
forward_mode = xpath_get_text(xml, '/network/forward/@mode')
forward_mode = forward_mode and forward_mode[0] or ''
forward_if = xpath_get_text(xml, '/network/forward/interface/@dev')
forward_pf = xpath_get_text(xml, '/network/forward/pf/@dev')
bridge = xpath_get_text(xml, '/network/bridge/@name')
bridge = bridge and bridge[0] or ''
return {
'subnet': net,
'dhcp': dhcp,
'bridge': bridge,
'forward': {
'mode': forward_mode,
'interface': forward_if,
'pf': forward_pf,
},
}
开发者ID:alinefm,项目名称:kimchi,代码行数:29,代码来源:networks.py
示例8: lookup
def lookup(self, name):
pool = self.get_storagepool(name, self.conn)
info = pool.info()
autostart = True if pool.autostart() else False
persistent = True if pool.isPersistent() else False
xml = pool.XMLDesc(0)
path = xpath_get_text(xml, '/pool/target/path')[0]
pool_type = xpath_get_text(xml, '/pool/@type')[0]
source = self._get_storage_source(pool_type, xml)
# FIXME: nfs workaround - prevent any libvirt operation
# for a nfs if the corresponding NFS server is down.
if pool_type == 'netfs' and not self._nfs_status_online(pool):
wok_log.debug(
'NFS pool %s is offline, reason: NFS ' 'server %s is unreachable.',
name,
source['addr'],
)
# Mark state as '4' => inaccessible.
info[0] = 4
# skip calculating volumes
nr_volumes = 0
else:
nr_volumes = self._get_storagepool_vols_num(pool)
res = {
'state': POOL_STATE_MAP[info[0]],
'path': path,
'source': source,
'type': pool_type,
'autostart': autostart,
'capacity': info[1],
'allocated': info[2],
'available': info[3],
'nr_volumes': nr_volumes,
'persistent': persistent,
'in_use': self._pool_used_by_template(name),
}
if not pool.isPersistent():
# Deal with deep scan generated pool
try:
with self.objstore as session:
task_id = session.get('scanning', name)
res['task_id'] = str(task_id)
res['type'] = 'kimchi-iso'
except NotFoundError:
# User created normal pool
pass
return res
开发者ID:alinefm,项目名称:kimchi,代码行数:49,代码来源:storagepools.py
示例9: _get_vms_attach_to_storagepool
def _get_vms_attach_to_storagepool(self, storagepool):
conn = self.conn.get()
# get storage pool path
pool = self.get_storagepool(storagepool, self.conn)
path = "".join(xpath_get_text(pool.XMLDesc(), "/pool/target/path"))
# activate and deactive quickly to get volumes
vms = []
for dom in conn.listAllDomains(0):
xml = dom.XMLDesc(0)
files = "/domain/devices/disk[@device='disk']/source/@file"
for file in xpath_get_text(xml, files):
if file not in vms and file.startswith(path):
vms.append(dom.name())
return vms
开发者ID:open-power-host-os,项目名称:kimchi,代码行数:16,代码来源:storagepools.py
示例10: test_forward_xml
def test_forward_xml(self):
"""
Test network forward xml
"""
params = {'mode': None}
forward = nxml._get_forward_elem(**params)
self.assertEqual(None, forward)
params['mode'] = 'nat'
params['dev'] = 'eth0'
xml = ET.tostring(nxml._get_forward_elem(**params))
mode = xpath_get_text(xml, '/forward/@mode')[0]
dev = xpath_get_text(xml, '/forward/@dev')[0]
self.assertEqual(params['mode'], mode)
self.assertEqual(params['dev'], dev)
开发者ID:alinefm,项目名称:kimchi,代码行数:16,代码来源:test_networkxml.py
示例11: deactivate
def deactivate(self, name):
if self._pool_used_by_template(name):
raise InvalidOperation('KCHPOOL0034E', {'name': name})
pool = self.get_storagepool(name, self.conn)
# FIXME: nfs workaround - do not try to deactivate a NFS pool
# if the NFS server is not reachable.
xml = pool.XMLDesc(0)
pool_type = xpath_get_text(xml, "/pool/@type")[0]
if pool_type == 'netfs' and not self._nfs_status_online(pool):
# block the user from dactivating the pool.
source = self._get_storage_source(pool_type, xml)
raise OperationFailed("KCHPOOL0033E",
{'name': name, 'server': source['addr']})
return
try:
persistent = pool.isPersistent()
pool.destroy()
except libvirt.libvirtError as e:
raise OperationFailed("KCHPOOL0010E",
{'name': name, 'err': e.get_error_message()})
# If pool was not persistent, then it was erased by destroy() and
# must return nothing here, to trigger _redirect() and avoid errors
if not persistent:
return ""
开发者ID:sdnnfv,项目名称:kimchi,代码行数:25,代码来源:storagepools.py
示例12: _get_storage_type
def _get_storage_type(self, pool_uri=None):
try:
pool = self._get_storage_pool(pool_uri)
except:
return ''
xml = pool.XMLDesc(0)
return xpath_get_text(xml, "/pool/@type")[0]
开发者ID:cclauss,项目名称:kimchi,代码行数:8,代码来源:templates.py
示例13: test_mem_dev_slots
def test_mem_dev_slots(self):
vm_uuid = str(uuid.uuid4()).replace('-', '')
t = VMTemplate({'name': 'test-template', 'cdrom': self.iso,
'memory': {'current': 2048, 'maxmemory': 3072}})
xml = t.to_vm_xml('test-vm', vm_uuid)
expr = "/domain/maxMemory/@slots"
slots = str(MEM_DEV_SLOTS[os.uname()[4]])
self.assertEquals(slots, xpath_get_text(xml, expr)[0])
开发者ID:aiminickwong,项目名称:kimchi,代码行数:8,代码来源:test_vmtemplate.py
示例14: _get_ipaddr_info
def _get_ipaddr_info(libvirt_interface_xml):
search_ip = \
xpath_get_text(libvirt_interface_xml,
"/interface/protocol[@family='ipv4']"
"/ip/@address")
if not len(search_ip):
return '', ''
search_prefix = \
xpath_get_text(libvirt_interface_xml,
"/interface/protocol[@family='ipv4']"
"/ip/@prefix")
ip_obj = ipaddr.IPv4Network('%s/%s' % (search_ip[0],
search_prefix[0]))
return str(ip_obj.ip), str(ip_obj.netmask)
开发者ID:MalleshKoti,项目名称:ginger,代码行数:17,代码来源:interfaces.py
示例15: test_to_xml
def test_to_xml(self):
graphics = {'type': 'spice', 'listen': '127.0.0.1'}
vm_uuid = str(uuid.uuid4()).replace('-', '')
if os.uname()[4] in ['ppc', 'ppc64', 'ppc64le']:
maxmem = 3328
else:
maxmem = 3072
t = VMTemplate({'name': 'test-template', 'cdrom': self.iso,
'max_memory': maxmem << 10})
xml = t.to_vm_xml('test-vm', vm_uuid, graphics=graphics)
self.assertEquals(vm_uuid, xpath_get_text(xml, "/domain/uuid")[0])
self.assertEquals('test-vm', xpath_get_text(xml, "/domain/name")[0])
expr = "/domain/devices/graphics/@type"
self.assertEquals(graphics['type'], xpath_get_text(xml, expr)[0])
expr = "/domain/devices/graphics/@listen"
self.assertEquals(graphics['listen'], xpath_get_text(xml, expr)[0])
expr = "/domain/maxMemory/@slots"
self.assertEquals('2', xpath_get_text(xml, expr)[0])
开发者ID:madhawa,项目名称:kimchi,代码行数:18,代码来源:test_vmtemplate.py
示例16: _get_storage_path
def _get_storage_path(self, pool_uri=None):
try:
pool = self._get_storage_pool(pool_uri)
except Exception:
return ''
xml = pool.XMLDesc(0)
return xpath_get_text(xml, '/pool/target/path')[0]
开发者ID:alinefm,项目名称:kimchi,代码行数:9,代码来源:templates.py
示例17: lookup
def lookup(self, name):
pool = self.get_storagepool(name, self.conn)
info = pool.info()
autostart = True if pool.autostart() else False
persistent = True if pool.isPersistent() else False
xml = pool.XMLDesc(0)
path = xpath_get_text(xml, "/pool/target/path")[0]
pool_type = xpath_get_text(xml, "/pool/@type")[0]
source = self._get_storage_source(pool_type, xml)
# FIXME: nfs workaround - prevent any libvirt operation
# for a nfs if the corresponding NFS server is down.
if pool_type == "netfs" and not self._nfs_status_online(pool):
wok_log.debug("NFS pool %s is offline, reason: NFS " "server %s is unreachable.", name, source["addr"])
# Mark state as '4' => inaccessible.
info[0] = 4
# skip calculating volumes
nr_volumes = 0
else:
nr_volumes = self._get_storagepool_vols_num(pool)
res = {
"state": POOL_STATE_MAP[info[0]],
"path": path,
"source": source,
"type": pool_type,
"autostart": autostart,
"capacity": info[1],
"allocated": info[2],
"available": info[3],
"nr_volumes": nr_volumes,
"persistent": persistent,
}
if not pool.isPersistent():
# Deal with deep scan generated pool
try:
with self.objstore as session:
task_id = session.get("scanning", name)
res["task_id"] = str(task_id)
res["type"] = "kimchi-iso"
except NotFoundError:
# User created normal pool
pass
return res
开发者ID:andreteodoro,项目名称:kimchi,代码行数:44,代码来源:storagepools.py
示例18: revert
def revert(self, vm_name, name):
try:
vir_dom = VMModel.get_vm(vm_name, self.conn)
vir_snap = self.get_vmsnapshot(vm_name, name)
vir_dom.revertToSnapshot(vir_snap, 0)
# get vm name recorded in the snapshot and return new uri params
vm_new_name = xpath_get_text(vir_snap.getXMLDesc(0), "domain/name")[0]
return [vm_new_name, name]
except libvirt.libvirtError, e:
raise OperationFailed("KCHSNAP0009E", {"name": name, "vm": vm_name, "err": e.message})
开发者ID:yashodhank,项目名称:kimchi,代码行数:11,代码来源:vmsnapshots.py
示例19: test_to_xml
def test_to_xml(self):
graphics = {'type': 'spice', 'listen': '127.0.0.1'}
vm_uuid = str(uuid.uuid4()).replace('-', '')
t = VMTemplate({'name': 'test-template', 'cdrom': self.iso})
xml = t.to_vm_xml('test-vm', vm_uuid, graphics=graphics)
self.assertEquals(vm_uuid, xpath_get_text(xml, "/domain/uuid")[0])
self.assertEquals('test-vm', xpath_get_text(xml, "/domain/name")[0])
expr = "/domain/devices/graphics/@type"
self.assertEquals(graphics['type'], xpath_get_text(xml, expr)[0])
expr = "/domain/devices/graphics/@listen"
self.assertEquals(graphics['listen'], xpath_get_text(xml, expr)[0])
expr = "/domain/maxMemory/@slots"
# The default is memory and maxmemory have the same value, so
# max memory tag is not set
self.assertEquals(0, len(xpath_get_text(xml, expr)))
expr = "/domain/memory"
self.assertEquals(str(1024), xpath_get_text(xml, expr)[0])
if hasattr(psutil, 'virtual_memory'):
host_memory = psutil.virtual_memory().total >> 10
else:
host_memory = psutil.TOTAL_PHYMEM >> 10
t = VMTemplate({'name': 'test-template', 'cdrom': self.iso,
'memory': {'current': (host_memory >> 10) - 512}})
try:
xml = t.to_vm_xml('test-vm', vm_uuid, graphics=graphics)
except Exception as e:
# Test current memory greater than maxmemory (1024/default)
self.assertTrue('KCHVM0041E' in e.message)
开发者ID:ShinJR,项目名称:kimchi,代码行数:29,代码来源:test_vmtemplate.py
示例20: test_to_xml
def test_to_xml(self):
graphics = {"type": "spice", "listen": "127.0.0.1"}
vm_uuid = str(uuid.uuid4()).replace("-", "")
t = VMTemplate({"name": "test-template", "cdrom": self.iso})
xml = t.to_vm_xml("test-vm", vm_uuid, graphics=graphics)
self.assertEquals(vm_uuid, xpath_get_text(xml, "/domain/uuid")[0])
self.assertEquals("test-vm", xpath_get_text(xml, "/domain/name")[0])
expr = "/domain/devices/graphics/@type"
self.assertEquals(graphics["type"], xpath_get_text(xml, expr)[0])
expr = "/domain/devices/graphics/@listen"
self.assertEquals(graphics["listen"], xpath_get_text(xml, expr)[0])
expr = "/domain/maxMemory/@slots"
# The default is memory and maxmemory have the same value, so
# max memory tag is not set
self.assertEquals(0, len(xpath_get_text(xml, expr)))
expr = "/domain/memory"
self.assertEquals(str(1024), xpath_get_text(xml, expr)[0])
if hasattr(psutil, "virtual_memory"):
host_memory = psutil.virtual_memory().total >> 10
else:
host_memory = psutil.TOTAL_PHYMEM >> 10
t = VMTemplate({"name": "test-template", "cdrom": self.iso, "memory": {"current": (host_memory >> 10) - 512}})
try:
xml = t.to_vm_xml("test-vm", vm_uuid, graphics=graphics)
except Exception as e:
# Test current memory greater than maxmemory (1024/default)
self.assertTrue("KCHVM0041E" in e.message)
开发者ID:carriercomm,项目名称:kimchi,代码行数:28,代码来源:test_vmtemplate.py
注:本文中的wok.xmlutils.utils.xpath_get_text函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论