本文整理汇总了Python中zstackwoodpecker.operations.resource_operations.query_resource函数的典型用法代码示例。如果您正苦于以下问题:Python query_resource函数的具体用法?Python query_resource怎么用?Python query_resource使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了query_resource函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test
def test():
h1_name = os.environ.get("hostName")
cond = res_ops.gen_query_conditions('name', '=', h1_name)
h1 = res_ops.query_resource(res_ops.HOST, cond)
ag1 = ag_ops.create_affinity_group(name="ag1", policy="antiHard")
vm1 = test_stub.create_ag_vm(affinitygroup_uuid=ag1.uuid, host_uuid=h1[0].uuid)
assert vm1.get_vm().hostUuid == h1[0].uuid
test_obj_dict.add_vm(vm1)
h2_name = os.environ.get("hostName2")
cond = res_ops.gen_query_conditions('name', '=', h2_name)
h2 = res_ops.query_resource(res_ops.HOST, cond)
vm2 = test_stub.create_ag_vm(affinitygroup_uuid=ag1.uuid, host_uuid=h2[0].uuid)
assert vm2.get_vm().hostUuid == h2[0].uuid
test_obj_dict.add_vm(vm2)
try:
vm3 = None
vm3 = test_stub.create_ag_vm(affinitygroup_uuid=ag1.uuid, host_uuid=h2[0].uuid)
except:
if not vm3:
test_util.test_logger("vm3 isn't created as expected")
finally:
if vm3:
test_util.test_fail("Test Fail, vm3 [uuid:%s] is not expected to be created" % vm3.get_vm().uuid)
test_lib.lib_error_cleanup(test_obj_dict)
ag_ops.delete_affinity_group(ag1.uuid)
test_util.test_pass("Affinity Group antiHard policy pass")
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:28,代码来源:test_shared_antihard_policy2.py
示例2: check
def check(self):
import json
import zstacklib.utils.jsonobject as jsonobject
super(zstack_vid_policy_db_checker, self).check()
try:
conditions = res_ops.gen_query_conditions('uuid', '=', self.test_obj.get_vid().uuid)
vid = res_ops.query_resource(res_ops.IAM2_VIRTUAL_ID, conditions)[0]
except Exception as e:
traceback.print_exc(file=sys.stdout)
test_util.test_logger('Check result: [vid Inventory uuid:] %s does not exist in database.' % self.test_obj.get_vid().uuid)
return self.judge(False)
conditions = res_ops.gen_query_conditions('virtualIDs.uuid', '=', self.test_obj.get_vid().uuid)
role_statements = res_ops.query_resource(res_ops.ROLE, conditions)[0].statements
for state_lst in self.test_obj.get_vid_statements()[0]['actions']:
test_result = False
for statement in role_statements:
atatement_dict = json.loads(jsonobject.dumps(statement))['statement']
for lst in atatement_dict['actions']:
if lst == state_lst:
test_result = True
if test_result == False:
test_util.test_logger('Check result: [vid Inventory statement:] does not exist in database.')
return self.judge(False)
return self.judge(True)
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:25,代码来源:zstack_db_checker.py
示例3: test
def test():
global ipsec
global vip1_uuid
global vpc_vr
cond = res_ops.gen_query_conditions('name', '=', 'public network')
public_network = res_ops.query_resource(res_ops.L3_NETWORK, cond)[0]
vip1 = test_stub.create_vip('vip_ipsec', public_network.uuid)
vip1_uuid = vip1.get_vip().uuid
test_util.test_dsc('Create vpc vr and attach networks')
vpc_vr = test_stub.create_vpc_vrouter()
cond = res_ops.gen_query_conditions('name', '=', 'l3VlanNetwork11')
l3_vlan_network11 = res_ops.query_resource(res_ops.L3_NETWORK, cond)[0]
vpc_vr.add_nic(l3_vlan_network11.uuid)
peer_address = '10.94.10.10'
try:
ipsec = ipsec_ops.create_ipsec_connection('ipsec', None, peer_address, '123456', vip1_uuid, None)
except:
test_util.test_fail('Failed to create vpc ipsec')
test_stub.delete_vip(vip1_uuid)
vpc_vr.destroy()
ipsec_ops.delete_ipsec_connection(ipsec.uuid)
test_util.test_pass('Create VPC Ipsec Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:26,代码来源:test_create_ipsec.py
示例4: add_storage_for_backup
def add_storage_for_backup(deployConfig):
print "try to add backup storage"
if xmlobject.has_element(deployConfig, 'backupStorages.imageStoreBackupStorage'):
print "find image store backup storage"
for bs in xmlobject.safe_list(deployConfig.backupStorages.imageStoreBackupStorage):
if hasattr(bs, 'local_backup_storage_'):
print "find local_backup_storage"
cond = res_ops.gen_query_conditions('tag', '=', "allowbackup")
tags = res_ops.query_resource(res_ops.SYSTEM_TAG, cond)
if len(tags) > 0:
print "local backup storage already exists"
break
cond = res_ops.gen_query_conditions('name', '=', bs.name_)
bss = res_ops.query_resource(res_ops.BACKUP_STORAGE, cond)
print bss
add_local_bs_tag = tag_ops.create_system_tag('ImageStoreBackupStorageVO', bss[0].uuid,'allowbackup')
if xmlobject.has_element(deployConfig, 'backupStorages.imageStoreBackupStorage'):
for bs in xmlobject.safe_list(deployConfig.backupStorages.imageStoreBackupStorage):
if hasattr(bs, 'remote_backup_storage_'):
print "find remote_backup_storage"
cond = res_ops.gen_query_conditions('tag', '=', "remotebackup")
tags = res_ops.query_resource(res_ops.SYSTEM_TAG, cond)
if len(tags) > 0:
print "remote backup storage already exists"
break
cond = res_ops.gen_query_conditions('name', '=', bs.name_)
bss = res_ops.query_resource(res_ops.BACKUP_STORAGE, cond)
print bss
add_local_bs_tag = tag_ops.create_system_tag('ImageStoreBackupStorageVO', bss[0].uuid,'remotebackup')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:29,代码来源:run_path.py
示例5: test
def test():
global test_obj_dict
volume_creation_option = test_util.VolumeOption()
ps_uuid = res_ops.query_resource(res_ops.PRIMARY_STORAGE, [])[0].uuid
test_util.test_dsc('Create volume and check')
disk_offering = test_lib.lib_get_disk_offering_by_name(os.environ.get('smallDiskOfferingName'))
volume_creation_option.set_disk_offering_uuid(disk_offering.uuid)
volume_creation_option.set_primary_storage_uuid(ps_uuid)
if res_ops.query_resource(res_ops.PRIMARY_STORAGE, [])[0].type == "LocalStorage":
host = test_lib.lib_find_random_host()
volume_creation_option.set_system_tags(["localStorage::hostUuid::%s" % host.uuid])
volume = test_stub.create_volume(volume_creation_option)
test_obj_dict.add_volume(volume)
volume.check()
volume_uuid = volume.volume.uuid
vol_size = volume.volume.size
set_size = 1024*1024*1024*5
vol_ops.resize_data_volume(volume_uuid, set_size)
cond = res_ops.gen_query_conditions('type', '=', "Data")
cond = res_ops.gen_query_conditions('status', '=', "Ready", cond)
vol_size_after = res_ops.query_resource(res_ops.VOLUME, cond)[0].size
if set_size != vol_size_after:
test_util.test_fail('Resize Data Volume failed, size = %s' % vol_size_after)
test_lib.lib_error_cleanup(test_obj_dict)
test_util.test_pass('Resize Data Volume Test Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:26,代码来源:test_resize_unattached_data_volume.py
示例6: test
def test():
global agent_url
global vm
imagestore = test_lib.lib_get_image_store_backup_storage()
if imagestore == None:
test_util.test_skip('Required imagestore to test')
image_uuid = test_stub.get_image_by_bs(imagestore.uuid)
cond = res_ops.gen_query_conditions('type', '=', 'SharedMountPoint')
pss = res_ops.query_resource(res_ops.PRIMARY_STORAGE, cond)
if len(pss) == 0:
test_util.test_skip('Required %s ps to test' % (ps_type))
ps_uuid = pss[0].uuid
vm = test_stub.create_vm(image_uuid=image_uuid, ps_uuid=ps_uuid)
vm.stop()
ha_ops.set_vm_instance_ha_level(vm.get_vm().uuid,'NeverStop')
cond = res_ops.gen_query_conditions('uuid', '=', vm.get_vm().uuid)
for i in range(5):
time.sleep(1)
try:
if res_ops.query_resource(res_ops.VM_INSTANCE, cond)[0].state == vm_header.RUNNING:
break
except:
test_util.test_logger('Retry until VM change to running')
if res_ops.query_resource(res_ops.VM_INSTANCE, cond)[0].state == vm_header.RUNNING:
test_util.test_pass('set HA after stopped VM test pass')
test_util.test_fail('set HA after stopped VM test fail')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:31,代码来源:test_set_ha_for_stopped_vm.py
示例7: umount_all_primary_storages_violently
def umount_all_primary_storages_violently():
session_uuid = acc_ops.login_as_admin()
zones = res_ops.query_resource(res_ops.ZONE)
for zone in zones:
conditions = res_ops.gen_query_conditions('zoneUuid', '=', zone.uuid)
conditions = res_ops.gen_query_conditions('state', '=', 'Enabled', conditions)
pss = res_ops.query_resource(res_ops.PRIMARY_STORAGE, conditions, session_uuid)
conditions = res_ops.gen_query_conditions('zoneUuid', '=', zone.uuid)
conditions = res_ops.gen_query_conditions('state', '=', host_header.ENABLED, conditions)
conditions = res_ops.gen_query_conditions('status', '=', host_header.CONNECTED, conditions)
conditions = res_ops.gen_query_conditions('hypervisorType', '=', inventory.KVM_HYPERVISOR_TYPE, conditions)
all_hosts = res_ops.query_resource(res_ops.HOST, conditions, session_uuid)
for host in all_hosts:
for ps in pss:
ps_url = ps.mountPath
thread = threading.Thread(\
target = umount_primary_storage_violently, \
args = (host.managementIp, ps_url))
thread.start()
while threading.active_count() > 1:
time.sleep(0.1)
acc_ops.logout(session_uuid)
delete_ps_ceph_pools()
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:25,代码来源:clean_util.py
示例8: test
def test():
global host
if test_lib.lib_get_active_host_number() < 2:
test_util.test_fail('Not available host to do maintenance, since there are not 2 hosts')
vm_cpu = 1
vm_memory = 1073741824 #1G
cond = res_ops.gen_query_conditions('name', '=', 'ttylinux')
image_uuid = res_ops.query_resource(res_ops.IMAGE, cond)[0].uuid
l3_network_uuid = res_ops.query_resource(res_ops.L3_NETWORK)[0].uuid
vm = test_stub.create_mini_vm([l3_network_uuid], image_uuid, cpu_num = vm_cpu, memory_size = vm_memory)
test_obj_dict.add_vm(vm)
host_uuid = test_lib.lib_get_vm_host(vm.vm).uuid
host_ops.change_host_state(host_uuid, 'maintain')
#need to update vm's inventory, since they will be changed by maintenace mode
vm.update()
vm.set_state(vm_header.STOPPED)
vm.check()
host_ops.change_host_state(host_uuid, 'enable')
if not linux.wait_callback_success(is_host_connected, host_uuid, 120):
test_util.test_fail('host status is not changed to connected, after changing its state to Enable')
vm.start()
vm.check()
vm.destroy()
test_obj_dict.rm_vm(vm)
test_util.test_pass('Maintain Host Test Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:31,代码来源:test_maintain_host.py
示例9: test
def test():
test_util.test_dsc("check all hosts chrony status")
host_uuid_list = []
host_ip_list = []
host_port_list = []
hosts = {}
for host_id in range(len(res_ops.query_resource(res_ops.HOST))):
managementIp = res_ops.query_resource(res_ops.HOST)[host_id].managementIp
sshPort = res_ops.query_resource(res_ops.HOST)[host_id].sshPort
uuid = res_ops.query_resource(res_ops.HOST)[host_id].uuid
host_ip_list.append(managementIp)
host_port_list.append(sshPort)
host_uuid_list.append(uuid)
hosts = dict(zip(host_ip_list, host_port_list))
print "hosts is %s" %(hosts)
for k, v in hosts.items():
check_chrony_status(k, v)
for host_uuid in host_uuid_list:
host_ops.reconnect_host(host_uuid)
time.sleep(5)
for k, v in hosts.items():
check_chrony_status(k, v)
host_ops.change_host_state(host_uuid_list[0], "disable")
for k, v in hosts.items():
check_chrony_status(k, v)
host_ops.change_host_state(host_uuid_list[0], "enable")
test_lib.lib_error_cleanup(test_obj_dict)
test_util.test_pass('Test chrony Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:31,代码来源:test_chrony.py
示例10: prepare_host_with_different_cpu_scenario
def prepare_host_with_different_cpu_scenario():
"""
Prepare vms in hosts
"""
global pre_vms
vm_creation_option = test_util.VmOption()
image_name = os.environ.get('imageName_s')
image_uuid = test_lib.lib_get_image_by_name(image_name).uuid
l3_name = os.environ.get('l3NoVlanNetworkName1')
#l3_name = os.environ.get('l3PublicNetworkName')
l3_net_uuid = test_lib.lib_get_l3_by_name(l3_name).uuid
conditions = res_ops.gen_query_conditions('type', '=', 'UserVm')
instance_offering_uuid = res_ops.query_resource(res_ops.INSTANCE_OFFERING, conditions)[0].uuid
#instance_offering_uuid = new_offering.uuid
vm_creation_option.set_l3_uuids([l3_net_uuid])
vm_creation_option.set_image_uuid(image_uuid)
vm_creation_option.set_instance_offering_uuid(instance_offering_uuid)
ps_uuid = res_ops.query_resource(res_ops.PRIMARY_STORAGE)[0].uuid
hosts = test_lib.lib_find_hosts_by_ps_uuid(ps_uuid)
host_id = 0
for host, max_vm_num in zip(hosts,[2,3,1,2]):
host_id +=1
for i in range(max_vm_num):
print "host_id=%s; i=%s" %(host_id, i)
vm_creation_option.set_name('pre-create-vm-%s-%s' %(host_id, i))
vm = test_vm_header.ZstackTestVm()
vm_creation_option.set_host_uuid(host.uuid)
vm.set_creation_option(vm_creation_option)
vm.create()
pre_vms.append(vm)
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:33,代码来源:test_allocator_strategy_crt_vm_cpu.py
示例11: test
def test():
h1_name = os.environ.get("hostName")
cond = res_ops.gen_query_conditions('name', '=', h1_name)
h1 = res_ops.query_resource(res_ops.HOST, cond)
ag1 = ag_ops.create_affinity_group(name="ag1", policy="antiHard")
vm1 = test_stub.create_ag_vm(affinitygroup_uuid=ag1.uuid, host_uuid=h1[0].uuid)
assert vm1.get_vm().hostUuid == h1[0].uuid
test_obj_dict.add_vm(vm1)
h2_name = os.environ.get("hostName2")
cond = res_ops.gen_query_conditions('name', '=', h2_name)
h2 = res_ops.query_resource(res_ops.HOST, cond)
vm2 = test_stub.create_ag_vm(affinitygroup_uuid=ag1.uuid, host_uuid=h2[0].uuid)
assert vm2.get_vm().hostUuid == h2[0].uuid
test_obj_dict.add_vm(vm2)
try:
vm1.migrate(vm2.get_vm().hostUuid)
except:
test_util.test_logger("vm1 is not expected to migrate to host2 [uuid: %s]" % vm2.get_vm().hostUuid)
h3_name = os.environ.get("hostName3")
cond = res_ops.gen_query_conditions('name', '=', h3_name)
h3 = res_ops.query_resource(res_ops.HOST, cond)
vm1.migrate(h3[0].uuid)
vm1.migrate(h1[0].uuid)
test_lib.lib_error_cleanup(test_obj_dict)
ag_ops.delete_affinity_group(ag1.uuid)
test_util.test_pass("Affinity Group antiHard policy pass")
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:31,代码来源:test_shared_antihard_policy5.py
示例12: test
def test():
global test_obj_dict
global ps_uuid
global host_uuid
global vr_uuid
test_util.test_dsc('Create test vm and check')
l3_1_name = os.environ.get('l3VlanNetworkName1')
vm = test_stub.create_vlan_vm(l3_name=l3_1_name)
#l3_1 = test_lib.lib_get_l3_by_name(l3_1_name)
#vr = test_lib.lib_find_vr_by_l3_uuid(l3_1.uuid)[0]
#vr_uuid = vr.uuid
host = test_lib.lib_get_vm_host(vm.get_vm())
host_uuid = host.uuid
test_obj_dict.add_vm(vm)
vm.check()
test_util.test_dsc('Add ISO Image')
cond = res_ops.gen_query_conditions("status", '=', "Connected")
bs_uuid = res_ops.query_resource(res_ops.BACKUP_STORAGE, cond)[0].uuid
img_option = test_util.ImageOption()
img_option.set_name('iso')
img_option.set_backup_storage_uuid_list([bs_uuid])
os.system("echo fake iso for test only > %s/apache-tomcat/webapps/zstack/static/test.iso" % (os.environ.get('zstackInstallPath')))
img_option.set_url('http://%s:8080/zstack/static/test.iso' % (os.environ.get('node1Ip')))
image_inv = img_ops.add_iso_template(img_option)
image = test_image.ZstackTestImage()
image.set_image(image_inv)
image.set_creation_option(img_option)
test_obj_dict.add_image(image)
ps = test_lib.lib_get_primary_storage_by_vm(vm.get_vm())
ps_uuid = ps.uuid
ps_ops.change_primary_storage_state(ps_uuid, 'maintain')
if not test_lib.lib_wait_target_down(vm.get_vm().vmNics[0].ip, '22', 90):
test_util.test_fail('VM is expected to stop when PS change to maintain state')
vm.set_state(vm_header.STOPPED)
vm.check()
test_util.test_dsc('Attach ISO to VM')
cond = res_ops.gen_query_conditions('name', '=', 'iso')
iso_uuid = res_ops.query_resource(res_ops.IMAGE, cond)[0].uuid
img_ops.attach_iso(iso_uuid, vm.vm.uuid)
ps_ops.change_primary_storage_state(ps_uuid, 'enable')
host_ops.reconnect_host(host_uuid)
#vm_ops.reconnect_vr(vr_uuid)
vrs = test_lib.lib_get_all_vrs()
for vr in vrs:
vm_ops.start_vm(vr.uuid)
vm.start()
vm.check()
vm.destroy()
vm.check()
#vm.expunge()
#vm.check()
test_util.test_pass('PS maintain mode Test Success')
开发者ID:mrwangxc,项目名称:zstack-woodpecker,代码行数:60,代码来源:test_maintain_ps_attach_iso.py
示例13: test
def test():
global test_obj_dict, bs, ps
#judge whether BS is imagestore
bs = res_ops.query_resource(res_ops.BACKUP_STORAGE)
for i in bs:
if i.type == inventory.IMAGE_STORE_BACKUP_STORAGE_TYPE:
break
else:
test_util.test_skip('Skip test on non-imagestore')
#judge whether PS is SharedBlock
ps = res_ops.query_resource(res_ops.PRIMARY_STORAGE)
for i in ps:
if i.type in ['SharedBlock', 'AliyunNAS']:
test_util.test_skip('Skip test on SharedBlock and PS')
image_name = os.environ.get('imageName_s')
l3_name = os.environ.get('l3PublicNetworkName')
vm = test_stub.create_vm("test_vm", image_name, l3_name)
#vm.check()
test_obj_dict.add_vm(vm)
new_vm = vm.clone(['test_vm_clone_with_on_data_volume'], full=True)[0]
test_obj_dict.add_vm(new_vm)
volumes_number = len(test_lib.lib_get_all_volumes(new_vm.vm))
if volumes_number != 1:
test_util.test_fail('Did not find 1 volumes for [vm:] %s. But we assigned 1 data volume when create the vm. We only catch %s volumes' % (new_vm.vm.uuid, volumes_number))
else:
test_util.test_logger('Find 1 volumes for [vm:] %s.' % new_vm.vm.uuid)
test_lib.lib_error_cleanup(test_obj_dict)
test_util.test_pass('Test clone vm with one data volume Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:33,代码来源:test_online_local_full_clone_no_data_volume.py
示例14: get_bs
def get_bs(self):
if self.ceph_bs_name:
conditions = res_ops.gen_query_conditions('name', '=', self.ceph_bs_name)
self.ceph_bs = res_ops.query_resource(res_ops.BACKUP_STORAGE, conditions)[0]
if self.ceph_bs_name_2:
conditions = res_ops.gen_query_conditions('name', '=', self.ceph_bs_name_2)
self.ceph_bs_2 = res_ops.query_resource(res_ops.BACKUP_STORAGE, conditions)[0]
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:7,代码来源:test_stub.py
示例15: path
def path():
cond = res_ops.gen_query_conditions('state', '=', "Enabled")
cond = res_ops.gen_query_conditions('status', '=', "Connected", cond)
ps_inv = res_ops.query_resource(res_ops.PRIMARY_STORAGE, cond)
cond_imagestore = res_ops.gen_query_conditions('type', '=', "ImageStoreBackupStorage", cond)
cond_ceph = res_ops.gen_query_conditions('type', '=', "Ceph", cond)
imagestore = res_ops.query_resource(res_ops.BACKUP_STORAGE, cond_imagestore)
ceph_bs = res_ops.query_resource(res_ops.BACKUP_STORAGE, cond_ceph)
san_ps = [ps.uuid for ps in ps_inv if ps.type == 'SharedBlock']
ceph_ps = [ps.uuid for ps in ps_inv if ps.type == 'Ceph']
san_vms = ['utility_vm_for_robot_test' + '-' + ps.name for ps in ps_inv if ps.type == 'SharedBlock']
ceph_vms = ['utility_vm_for_robot_test' + '-' + ps.name for ps in ps_inv if ps.type == 'Ceph']
if san_ps and ceph_ps:
return dict(initial_formation="template3",
path_list=[[TestAction.create_volume, "volume1", "=ps_uuid::%s" % san_ps[0]],
[TestAction.attach_volume, san_vms[-1], "volume1"],
[TestAction.resize_volume, san_vms[-1], 5*1024*1024],
[TestAction.clone_vm, san_vms[-1], "vm2", "=full"],
[TestAction.detach_volume, "volume1"],
[TestAction.attach_volume, ceph_vms[0], "volume1"],
[TestAction.stop_vm, san_vms[-1]],
[TestAction.reinit_vm, san_vms[-1]],
[TestAction.start_vm, san_vms[-1]],
[TestAction.clone_vm, ceph_vms[0], "vm3"],
[TestAction.resize_volume, ceph_vms[0], 5*1024*1024],
[TestAction.detach_volume, "volume1"],
[TestAction.create_image_from_volume, ceph_vms[0], 'image_created_from_%s' % ceph_vms[0], "=bs_uuid::%s" % imagestore[0].uuid],
[TestAction.create_vm_by_image, 'image_created_from_%s' % ceph_vms[0], 'qcow2', 'vm4', '=ps_uuid::%s' % random.choice(san_ps)],
[TestAction.stop_vm, 'vm4'],
[TestAction.ps_migrage_vm, 'vm4'],
[TestAction.attach_volume, 'vm4', "volume1"]])
else:
return dict(initial_formation="template3", path_list=[])
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:34,代码来源:ceph_san_path1.py
示例16: test
def test():
#skip ceph in c74
cmd = "cat /etc/redhat-release | grep '7.4'"
mn_ip = res_ops.query_resource(res_ops.MANAGEMENT_NODE)[0].hostName
rsp = test_lib.lib_execute_ssh_cmd(mn_ip, 'root', 'password', cmd, 180)
if rsp != False:
ps = res_ops.query_resource(res_ops.PRIMARY_STORAGE)
for i in ps:
if i.type == 'Ceph':
test_util.test_skip('cannot hotplug iso to the vm in ceph,it is a libvirt bug:https://bugzilla.redhat.com/show_bug.cgi?id=1541702.')
multi_iso.add_iso_image()
multi_iso.create_windows_vm()
test_obj_dict.add_vm(multi_iso.vm1)
multi_iso.get_all_iso_uuids()
multi_iso.attach_iso(multi_iso.iso_uuids[0])
multi_iso.attach_iso(multi_iso.iso_uuids[1])
multi_iso.attach_iso(multi_iso.iso_uuids[2])
multi_iso.check_windows_vm_cdrom(3)
multi_iso.detach_iso(multi_iso.iso_uuids[1])
multi_iso.check_windows_vm_cdrom(2)
# multi_iso.vm1.reboot()
multi_iso.detach_iso(multi_iso.iso_uuids[0])
multi_iso.check_windows_vm_cdrom(1)
multi_iso.detach_iso(multi_iso.iso_uuids[2])
multi_iso.check_windows_vm_cdrom(0)
test_lib.lib_robot_cleanup(test_obj_dict)
test_util.test_pass('Attach 3 ISO Test Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:31,代码来源:test_attach_3_iso_windows.py
示例17: check
def check(self):
super(zstack_kvm_sg_db_exist_checker, self).check()
sg_list = self.test_obj.get_sg_list_by_nic(self.nic_uuid)
if not sg_list:
conditions = res_ops.gen_query_conditions('vmNicUuid', '=', self.nic_uuid)
nic_sg = res_ops.query_resource(res_ops.VM_SECURITY_GROUP, conditions)
if not nic_sg:
test_util.test_logger('Check result: No [Security Group] is found in database for [nic:] %s.' % self.nic_uuid)
return self.judge(False)
else:
test_util.test_warn('Check result: [Security Group] is found in database for [nic:] %s. It is not consistent with test_sg record.' % self.nic_uuid)
return self.judge(True)
for test_sg in sg_list:
try:
conditions = res_ops.gen_query_conditions('uuid', '=', test_sg.security_group.uuid)
sg = res_ops.query_resource(res_ops.SECURITY_GROUP, conditions)[0]
except Exception as e:
traceback.print_exc(file=sys.stdout)
test_util.test_logger('Check result: [Security Group Inventory uuid:] %s does not exist in database.' % test_sg.security_group.uuid)
return self.judge(False)
test_util.test_logger('Check result: [SecurityGroup Inventory uuid:] %s exist in database.' % sg.uuid)
return self.judge(True)
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:25,代码来源:zstack_kvm_security_group_checker.py
示例18: check
def check(self):
super(zstack_kvm_lbl_checker, self).check()
self.vm_nic_uuids = self.lbl.get_vm_nics_uuid()
self.algorithm = self.lbl.get_algorithm()
self.vm_list = []
self.vm_ip_test_dict = {}
for vm_nic_uuid in self.vm_nic_uuids:
vm = test_lib.lib_get_vm_by_nic(vm_nic_uuid)
if vm.state == 'Running':
nic_ip = test_lib.lib_get_nic_by_uuid(vm_nic_uuid).ip
self.vm_ip_test_dict[nic_ip] = 0
self.vm_list.append(vm)
if not self.vm_list:
test_util.test_logger('There is not living vm for load balancer test')
return self.judge(False)
cond = res_ops.gen_query_conditions('listeners.uuid', '=', self.lbl_uuid)
vip_uuid = res_ops.query_resource(res_ops.LOAD_BALANCER, cond)[0].vipUuid
cond = res_ops.gen_query_conditions('uuid', '=', vip_uuid)
self.vip_ip = res_ops.query_resource(res_ops.VIP, cond)[0].ip
if not len(self.vm_list) > 1:
self.do_so_check()
return
if self.algorithm == lb_header.LB_ALGORITHM_RR:
self.do_rr_check()
elif self.algorithm == lb_header.LB_ALGORITHM_LC:
#self.do_lc_check()
#If not consider long connection, leastconn is same as round robin.
self.do_rr_check()
elif self.algorithm == lb_header.LB_ALGORITHM_SO:
self.do_so_check()
开发者ID:yuyanshuo,项目名称:zstack-woodpecker,代码行数:35,代码来源:zstack_kvm_load_balancer_checker.py
示例19: compare
def compare(ps, vm, dvol, backup):
test_util.test_logger("-----------------compare----------------")
# find vm_host
host = test_lib.lib_find_host_by_vm(vm.vm)
bs = res_ops.query_resource(res_ops.BACKUP_STORAGE)[0]
cond = res_ops.gen_query_conditions("uuid", '=', dvol.volume.uuid)
current_volume = res_ops.query_resource(res_ops.VOLUME, cond)[0]
vol_path = current_volume.installPath
if ps.type == "SharedBlock":
vol_path = "/dev/" + current_volume.installPath.split("/")[2] + "/" + current_volume.installPath.split("/")[3]
test_util.test_logger(vol_path)
name = backup.backupStorageRefs[0].installPath.split("/")[2]
id = backup.backupStorageRefs[0].installPath.split("/")[3]
# compare vm_root_volume & image
cmd = "mkdir /root/%s;" \
"/usr/local/zstack/imagestore/bin/zstcli " \
"-rootca=/var/lib/zstack/imagestorebackupstorage/package/certs/ca.pem " \
"-url=%s:8000 " \
"pull -installpath /root/%s/old.qcow2 %s:%s;" \
"qemu-img compare %s /root/%s/old.qcow2;" % (id, bs.hostname, id, name, id, vol_path, id)
# clean image
result = test_lib.lib_execute_ssh_cmd(host.managementIp, "root", "password", cmd, timeout=300)
if result != "Images are identical.\n":
test_util.test_fail("compare vm_root_volume & image created by backup")
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:27,代码来源:test_backup_replay_volume.py
示例20: add_image_config
def add_image_config(root_xml, original_images_setting, session_uuid = None):
images_xml = etree.SubElement(root_xml, "images")
cond = res_ops.gen_query_conditions('state', '=', 'Enabled')
images = res_ops.query_resource(res_ops.IMAGE, cond, session_uuid)
pre_images = {}
if original_images_setting:
pre_images_list = \
original_images_setting.get_child_node_as_list('image')
for pre_image in pre_images_list:
pre_images[pre_image.url_] = pre_image
for image in images:
image_xml = etree.SubElement(images_xml, "image")
set_xml_item_attr(image_xml, 'name', image.name)
set_xml_item_attr(image_xml, 'description', image.description)
set_xml_item_attr(image_xml, 'url', image.url)
set_xml_item_attr(image_xml, 'format', image.format)
set_xml_item_attr(image_xml, 'mediaType', image.mediaType)
set_xml_item_attr(image_xml, 'guestOsType', image.guestOsType)
set_xml_item_attr(image_xml, 'hypervisorType', image.hypervisorType)
set_xml_item_attr(image_xml, 'bits', image.bits)
if pre_images.has_key(image.url):
set_xml_item_attr(image_xml, 'username', \
pre_images[image.url].username__)
set_xml_item_attr(image_xml, 'password', \
pre_images[image.url].password__)
for bs in image.backupStorageRefs:
cond = res_ops.gen_query_conditions('uuid', '=', \
bs.backupStorageUuid)
bs = res_ops.query_resource(res_ops.BACKUP_STORAGE, cond, \
session_uuid)[0]
add_xml_item_value(image_xml, 'backupStorageRef', bs.name)
开发者ID:chancelq,项目名称:zstack-woodpecker,代码行数:33,代码来源:export_operations.py
注:本文中的zstackwoodpecker.operations.resource_operations.query_resource函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论