本文整理汇总了Python中zstackwoodpecker.test_lib.lib_execute_ssh_cmd函数的典型用法代码示例。如果您正苦于以下问题:Python lib_execute_ssh_cmd函数的具体用法?Python lib_execute_ssh_cmd怎么用?Python lib_execute_ssh_cmd使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了lib_execute_ssh_cmd函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: check_host_prometheus_conf
def check_host_prometheus_conf():
global host_ip
global host_uuid
global mn_ip
mn_ip = res_ops.query_resource(res_ops.MANAGEMENT_NODE)[0].hostName
# host_list = test_stub.get_sce_hosts(test_lib.all_scenario_config, test_lib.scenario_file)
host_list = test_lib.lib_get_all_hosts_from_plan()
cmd = 'yum install -y jq --nogpgcheck'
test_lib.lib_execute_ssh_cmd(mn_ip, 'root', 'password', cmd, 180)
for host in host_list:
host_ip = host.managementIp_
host_hostname = host_ip.replace('.', '-')
host_uuid = test_lib.lib_get_host_by_ip(host_ip).uuid
if host_ip == mn_ip:
cmd = "jq -r '.[].targets[]' /usr/local/zstacktest/prometheus/discovery/management-node/management-server-exporter.json"
cmd_out = test_lib.lib_execute_ssh_cmd(mn_ip, 'root', 'password', cmd, 180)
expect_result = mn_ip + ':' + str(8081)
if cmd_out.strip() != expect_result.strip():
test_util.test_fail('targets in management-node-exporter.json is not right.')
else:
cmd = "jq -r '.[].targets[]' /usr/local/zstacktest/prometheus/discovery/hosts/%s-%s.json" % (host_uuid, host_hostname)
cmd_out = test_lib.lib_execute_ssh_cmd(mn_ip, 'root', 'password', cmd, 180)
expect_result1 = host_ip + ':' + str(9103)
expect_result2 = host_ip + ':' + str(9100)
expect_result3 = host_ip + ':' + str(7069)
if expect_result1 not in cmd_out or expect_result2 not in cmd_out or expect_result3 not in cmd_out:
test_util.test_fail('targets in hosts/%s-%s.json is not right.') % (host_uuid, host_hostname)
cmd = "jq -r '.[].labels[]' /usr/local/zstacktest/prometheus/discovery/hosts/%s-%s.json" % (host_uuid, host_hostname)
cmd_out = test_lib.lib_execute_ssh_cmd(mn_ip, 'root', 'password', cmd, 180)
if host_uuid.strip() != cmd_out.strip():
test_util.test_fail('labels in hosts/%s-%s.json is not right.') % (host_uuid, host_hostname)
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:33,代码来源:test_prometheus.py
示例2: test
def test():
global vm
cmd = "init 0"
host_username = os.environ.get('nodeUserName')
host_password = os.environ.get('nodePassword')
zstack_ha_vip = os.environ.get('zstackHaVip')
node1_ip = os.environ.get('node1Ip')
test_util.test_logger("shutdown node: %s" % (node1_ip))
rsp = test_lib.lib_execute_ssh_cmd(node1_ip, host_username, host_password, cmd, 180)
node2_ip = os.environ.get('node2Ip')
test_util.test_logger("shutdown node: %s" % (node2_ip))
rsp = test_lib.lib_execute_ssh_cmd(node2_ip, host_username, host_password, cmd, 180)
test_util.test_logger("recover node: %s" % (node1_ip))
os.system('bash -ex %s %s' % (os.environ.get('nodeRecoverScript'), node1_ip))
test_util.test_logger("recover node: %s" % (node2_ip))
os.system('bash -ex %s %s' % (os.environ.get('nodeRecoverScript'), node2_ip))
cmd = "zstack-ctl install_ha --host1-info %s:%[email protected]%s --host2-info %s:%[email protected]%s --vip %s --recovery-from-this-host" % \
(host_username, host_password, node1_ip, host_username, host_password, node2_ip, zstack_ha_vip)
rsp = test_lib.lib_execute_ssh_cmd(node1_ip, host_username, host_password, cmd, 180)
if not rsp:
rsp = test_lib.lib_execute_ssh_cmd(node2_ip, host_username, host_password, cmd, 180)
time.sleep(180)
test_stub.exercise_connection(600)
vm = test_stub.create_basic_vm()
vm.check()
vm.destroy()
test_util.test_pass('Create VM Test Success')
开发者ID:TinaL3,项目名称:zstack-woodpecker,代码行数:31,代码来源:test_all_nodes_recovery_create_vm.py
示例3: get_s_vm_cfg_lst_vip_bind
def get_s_vm_cfg_lst_vip_bind(scenarioConfig, scenarioFile, retry_times=1):
"""
It gets host with vip bound, while returned a s_vm config
"""
mha_s_vm_list = get_mha_s_vm_list_from_scenario_file(scenarioConfig, scenarioFile)
if len(mha_s_vm_list) < 1:
return []
test_util.test_logger("@@[email protected]@: mha_s_vm_list=<%s>" %(str(mha_s_vm_list)))
host_list = []
vip = os.environ['zstackHaVip']
for host in mha_s_vm_list:
host_config = sce_ops.get_scenario_config_vm(host.name_, scenarioConfig)
cmd = "ip a|grep " + vip
try:
for retry_cnt in range(retry_times):
if sce_is_sep_pub():
vm_list = test_lib.lib_execute_ssh_cmd(host.managementIp_, host_config.imageUsername_, host_config.imagePassword_,cmd)
else:
vm_list = test_lib.lib_execute_ssh_cmd(host.ip_, host_config.imageUsername_, host_config.imagePassword_,cmd)
if vm_list:
host_list.append(host)
break
time.sleep(1)
except Exception, e:
test_util.test_logger("@@get host [email protected]@:%s" %(str(e)))
continue
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:28,代码来源:test_stub.py
示例4: test
def test():
test_util.test_dsc("create vr vm and vpc vrouter")
vm = test_stub.create_vr_vm('vr_vm','imageName_s','l3NoVlanNetworkName2')
test_obj_dict.add_vm(vm)
if not test_lib.lib_find_vr_by_vm(vm.vm) or not test_lib.lib_find_vr_by_vm(vm.vm)[0]:
test_lib.lib_error_cleanup(test_obj_dict)
test_util.test_skip("skip the test for no vr found in the env.")
vpc_vr = test_stub.create_vpc_vrouter()
check_host_prometheus_conf()
check_vrouter_prometheus_conf()
check_prometheus_data()
hosts = test_lib.lib_get_all_hosts_from_plan()
for host in hosts:
host_ops.reconnect_host(host_uuid)
check_host_prometheus_conf()
check_vrouter_prometheus_conf()
check_prometheus_data()
test_lib.lib_execute_ssh_cmd(mn_ip,"root","password","zstack-ctl restart_node",timeout=300)
check_host_prometheus_conf()
check_vrouter_prometheus_conf()
check_prometheus_data()
test_lib.lib_error_cleanup(test_obj_dict)
test_util.test_pass('Test prometheus Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:27,代码来源:test_prometheus.py
示例5: test
def test():
global vm1
global vm2
global node_ip
vm1 = test_stub.create_basic_vm()
node_ip = os.environ.get('node1Ip')
test_util.test_logger("stop rabbitmq on node: %s" % (node_ip))
cmd = "rabbitmqctl stop_app"
host_username = os.environ.get('nodeUserName')
host_password = os.environ.get('nodePassword')
rsp = test_lib.lib_execute_ssh_cmd(node_ip, host_username, host_password, cmd, 180)
time.sleep(120)
test_stub.exercise_connection(1200)
test_util.test_logger("create vm to check if it still works")
vm1.check()
vm1.destroy()
vm2 = test_stub.create_basic_vm()
vm2.check()
vm2.destroy()
cmd = "rabbitmqctl start_app"
rsp = test_lib.lib_execute_ssh_cmd(node_ip, host_username, host_password, cmd, 180)
time.sleep(60)
test_stub.exercise_connection(1200)
test_util.test_pass('Create VM Test Rabbitmq Stop on one node Success')
开发者ID:rynetang,项目名称:zstack-woodpecker,代码行数:25,代码来源:test_rabbitmq_stop_create_vm.py
示例6: test
def test():
imagestore_backup_storage = test_lib.lib_get_image_store_backup_storage()
if not imagestore_backup_storage:
test_util.test_skip('Not find image store type backup storage.')
img_option = test_util.ImageOption()
img_option.set_name('iso')
root_disk_uuid = test_lib.lib_get_disk_offering_by_name(os.environ.get('rootDiskOfferingName')).uuid
bs_uuid = imagestore_backup_storage.uuid
img_option.set_backup_storage_uuid_list([bs_uuid])
command = "command -v genisoimage"
result = test_lib.lib_execute_ssh_cmd(os.environ['ZSTACK_BUILT_IN_HTTP_SERVER_IP'], 'root', 'password', command)
if not result:
command = "yum -y install genisoimage --disablerepo=* --enablerepo=zstack-local"
test_lib.lib_execute_ssh_cmd(os.environ['ZSTACK_BUILT_IN_HTTP_SERVER_IP'], 'root', 'password', command)
command = "genisoimage -o %s/apache-tomcat/webapps/zstack/static/zstack-repo/7/x86_64/os/test.iso /tmp/" % os.environ.get('zstackInstallPath')
test_lib.lib_execute_ssh_cmd(os.environ['ZSTACK_BUILT_IN_HTTP_SERVER_IP'], 'root', 'password', command)
img_option.set_url('http://%s:8080/zstack/static/zstack-repo/7/x86_64/os/test.iso' % (os.environ['ZSTACK_BUILT_IN_HTTP_SERVER_IP']))
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)
image_url = image.export()
image.delete_exported_image()
test_lib.lib_robot_cleanup(test_obj_dict)
if image_url.endswith('.iso'):
test_util.test_pass('Export ISO Image Test Success')
else:
test_util.test_fail('Export ISO Image Test Fail, exported ISO image ends with %s' % (image_url.split('.')[-1]))
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:31,代码来源:test_export_iso.py
示例7: test
def test():
global test_host
global mn_ip
global host_list
allow_ps_list = [inventory.LOCAL_STORAGE_TYPE]
test_lib.skip_test_when_ps_type_not_in_list(allow_ps_list)
mn_ip = res_ops.query_resource(res_ops.MANAGEMENT_NODE)[0].hostName
host_list = test_stub.get_sce_hosts(test_lib.all_scenario_config, test_lib.scenario_file)
for host in host_list:
if host.ip_ != mn_ip:
test_host = host
break
if not test_host:
test_util.test_fail('there is no host with ip excluding mn_ip: %s in scenario file.' %(mn_ip))
host_username = os.environ.get('hostUsername')
host_password = os.environ.get('hostPassword')
cmd = "zstack-ctl stop"
if not test_lib.lib_execute_ssh_cmd(mn_ip, host_username, host_password, cmd, timeout = 120):
test_util.test_fail("CMD:%s execute failed on %s" %(cmd, mn_ip))
t = threading.Thread(target=async_ifconfig_br_eth0_down_up_wrapper,
args=(240, test_host.ip_, host_username, host_password))
t.start()
cmd = "nohup zstack-ctl start &"
if not test_lib.lib_execute_ssh_cmd(mn_ip, host_username, host_password, cmd, timeout = 120):
test_util.test_fail("CMD:%s execute failed on %s" %(cmd, mn_ip))
t.join()
test_util.test_pass('Test zstack-ctl start when host status is not conformance with zstack db Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:34,代码来源:test_ifdown_br_eth0_zstack_ctl_start.py
示例8: test
def test():
bs = res_ops.query_resource(res_ops.BACKUP_STORAGE)[0]
if bs.type == "Ceph":
test_util.test_skip('bs: %s is ceph backup storage. Will skip test.' % bs.uuid)
command = 'dd if=/dev/zero of=%s bs=1M count=1 seek=300' % test_image
test_lib.lib_execute_ssh_cmd(bs.hostname, 'root', 'password', command)
image_name = 'test-image-%s' % time.time()
image_option = test_util.ImageOption()
image_option.set_name(image_name)
image_option.set_description('test image which is upload from local filesystem.')
image_option.set_url('file://%s' % test_image)
image_option.set_backup_storage_uuid_list([bs.uuid])
image_option.set_format('raw')
image_option.set_mediaType('RootVolumeTemplate')
image_inv = img_ops.add_root_volume_template(image_option)
time.sleep(10)
image = zstack_image_header.ZstackTestImage()
image.set_creation_option(image_option)
image.set_image(image_inv)
test_obj_dict.add_image(image)
image.check()
vm = test_stub.create_vm(image_name = image_name)
vm.destroy()
image.delete()
command = 'ls %s' % test_image
if not test_lib.lib_execute_ssh_cmd(bs.hostname, 'root', 'password', command):
test_util.test_fail('test image disappeared, after add image.')
command = 'rm -f %s' % test_image
result = test_lib.lib_execute_ssh_cmd(bs.hostname, 'root', 'password', command)
test_util.test_pass('Test adding image from local storage pass.')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:32,代码来源:test_add_local_image.py
示例9: test
def test():
ps = res_ops.query_resource(res_ops.PRIMARY_STORAGE)
for i in ps:
if i.type == 'AliyunEBS':
test_util.test_skip('Skip vm reinit test on AliyunEBS')
vm = test_stub.create_user_vlan_vm()
test_obj_dict.add_vm(vm)
vm.check()
vm_inv = vm.get_vm()
vm_ip = vm_inv.vmNics[0].ip
cmd = 'touch /root/test-file-for-reinit'
rsp = test_lib.lib_execute_ssh_cmd(vm_ip, 'root', 'password', cmd, 180)
if rsp == False:
test_util.test_fail('Fail to create file in VM')
vm.stop()
vm.reinit()
vm.update()
vm.check()
vm.start()
cmd = '[ -e /root/test-file-for-reinit ] && echo yes || echo no'
rsp = test_lib.lib_execute_ssh_cmd(vm_ip, 'root', 'password', cmd, 180)
if rsp == 'yes':
test_util.test_fail('VM does not be reverted to image used for creating the VM, the later file still exists')
vm.destroy()
test_util.test_pass('Re-init VM Test Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:30,代码来源:test_reinit_vm.py
示例10: test
def test():
vm = test_stub.create_user_vlan_vm()
test_obj_dict.add_vm(vm)
vm.check()
vm_inv = vm.get_vm()
vm_ip = vm_inv.vmNics[0].ip
cmd = 'touch /root/test-file-for-reinit'
rsp = test_lib.lib_execute_ssh_cmd(vm_ip, 'root', 'password', cmd, 180)
if rsp == False:
test_util.test_fail('Fail to create file in VM')
vm.stop()
vm.reinit()
vm.update()
vm.check()
vm.start()
cmd = '[ -e /root/test-file-for-reinit ] && echo yes || echo no'
rsp = test_lib.lib_execute_ssh_cmd(vm_ip, 'root', 'password', cmd, 180)
if rsp == 'yes':
test_util.test_fail('VM does not be reverted to image used for creating the VM, the later file still exists')
vm.destroy()
test_util.test_pass('Re-init VM Test Success')
开发者ID:mrwangxc,项目名称:zstack-woodpecker,代码行数:25,代码来源:test_reinit_vm.py
示例11: test
def test():
if test_lib.scenario_config != None and test_lib.scenario_file != None and not os.path.exists(test_lib.scenario_file):
scenario_operations.deploy_scenario(test_lib.all_scenario_config, test_lib.scenario_file, test_lib.deploy_config)
test_util.test_skip('Suite Setup Success')
if test_lib.scenario_config != None and test_lib.scenario_destroy != None:
scenario_operations.destroy_scenario(test_lib.all_scenario_config, test_lib.scenario_destroy)
nic_name = "eth0"
if test_lib.scenario_config != None and test_lib.scenario_file != None and os.path.exists(test_lib.scenario_file):
nic_name = "zsn0"
#This vlan creation is not a must, if testing is under nested virt env. But it is required on physical host without enough physcial network devices and your test execution machine is not the same one as Host machine.
#no matter if current host is a ZStest host, we need to create 2 vlan devs for future testing connection for novlan test cases.
#linux.create_vlan_eth(nic_name, 10)
#linux.create_vlan_eth(nic_name, 11)
#If test execution machine is not the same one as Host machine, deploy work is needed to separated to 2 steps(deploy_test_agent, execute_plan_without_deploy_test_agent). And it can not directly call SetupAction.run()
test_lib.setup_plan.deploy_test_agent()
#cmd = host_plugin.CreateVlanDeviceCmd()
#cmd.ethname = nic_name
#cmd.vlan = 10
#cmd2 = host_plugin.CreateVlanDeviceCmd()
#cmd2.ethname = nic_name
#cmd2.vlan = 11
testHosts = test_lib.lib_get_all_hosts_from_plan()
if type(testHosts) != type([]):
testHosts = [testHosts]
#for host in testHosts:
# http.json_dump_post(testagent.build_http_path(host.managementIp_, host_plugin.CREATE_VLAN_DEVICE_PATH), cmd)
# http.json_dump_post(testagent.build_http_path(host.managementIp_, host_plugin.CREATE_VLAN_DEVICE_PATH), cmd2)
test_lib.setup_plan.execute_plan_without_deploy_test_agent()
if test_lib.scenario_config != None and test_lib.scenario_file != None and os.path.exists(test_lib.scenario_file):
mn_ips = deploy_operations.get_nodes_from_scenario_file(test_lib.all_scenario_config, test_lib.scenario_file, test_lib.deploy_config)
if os.path.exists(EXTRA_SUITE_SETUP_SCRIPT):
os.system("bash %s '%s' %s" % (EXTRA_SUITE_SETUP_SCRIPT, mn_ips, 'disaster-recovery'))
elif os.path.exists(EXTRA_SUITE_SETUP_SCRIPT):
os.system("bash %s '' '%s'" % (EXTRA_SUITE_SETUP_SCRIPT, 'disaster-recovery'))
dicts = {}
for host in testHosts:
cmd = 'virsh --version'
dicts[host.managementIp_] = test_lib.lib_execute_ssh_cmd(host.managementIp_, host.username_, host.password_, cmd, 180)
deploy_operations.deploy_initial_database(test_lib.deploy_config, test_lib.all_scenario_config, test_lib.scenario_file)
for host in testHosts:
os.system("bash %s %s" % (EXTRA_HOST_SETUP_SCRIPT, host.managementIp_))
version = test_lib.lib_execute_ssh_cmd(host.managementIp_, host.username_, host.password_, cmd, 180)
# Some host are installed in expert mode without virsh
if dicts[host.managementIp_]:
assert version == dicts[host.managementIp_]
if test_lib.lib_get_ha_selffencer_maxattempts() != None:
test_lib.lib_set_ha_selffencer_maxattempts('60')
test_lib.lib_set_ha_selffencer_storagechecker_timeout('60')
test_lib.lib_set_primary_storage_imagecache_gc_interval(1)
test_util.test_pass('Suite Setup Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:56,代码来源:suite_setup.py
示例12: check_prometheus_data
def check_prometheus_data():
test_util.test_dsc("Check prometheus data.")
cmd = "ps -ef|grep [t]ools/prometheus |awk '{for(i=7+1;i<=NF;i++)printf $i \" \";printf\"\\n\"}'"
cmd_out = test_lib.lib_execute_ssh_cmd(mn_ip, 'root', 'password', cmd, 180)
if cmd_out == None:
test_util.test_fail('prometheus agent is not enabled.')
cmd = 'ls -A /var/lib/zstack/prometheus/data/ |wc -w'
cmd_out = test_lib.lib_execute_ssh_cmd(mn_ip, 'root', 'password', cmd, 180)
if cmd_out <= 10:
test_util.test_fail('prometheus data is null.')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:10,代码来源:test_prometheus.py
示例13: verify_config
def verify_config(management_ip):
tmp_file = '/tmp/%s' % uuid.uuid1().get_hex()
test_util.test_logger("%s" %(tmp_file))
test_lib.lib_execute_ssh_cmd(management_ip, 'root', 'password', 'zstack-cli LogInByAccount accountName=admin password=password', 180)
test_lib.lib_execute_ssh_cmd(management_ip,'root','password','zstack-cli QueryGlobalConfig sortBy=name > %s' %(tmp_file), 180)
default_path=environment(management_ip)
test_util.test_logger("default global config path:%s" %(default_path))
test_util.test_logger("check two file diff:%s, %s" %(default_path,tmp_file))
check(default_path,tmp_file)
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:10,代码来源:test_global_config.py
示例14: test
def test():
global vm, kvm_host
ps = res_ops.query_resource(res_ops.PRIMARY_STORAGE)
for i in ps:
if i.type == inventory.LOCAL_STORAGE_TYPE:
break
else:
test_util.test_skip('Skip test on non-localstoreate PS')
#ps = res_ops.query_resource(res_ops.PRIMARY_STORAGE)[0]
#if ps.type != inventory.LOCAL_STORAGE_TYPE:
# test_util.test_skip('Skip test on non-localstorage')
#if "test-config-local-ps.xml" != os.path.basename(os.environ.get('WOODPECKER_TEST_CONFIG_FILE')).strip():
# test_util.test_skip('Skip test on non-localstoreage')
#query all hosts and change password
conditions = res_ops.gen_query_conditions('state', '=', 'Enabled')
if res_ops.query_resource(res_ops.HOST, conditions):
kvm_host = res_ops.query_resource(res_ops.HOST, conditions)
for i in kvm_host:
host_ops.update_kvm_host(i.uuid, 'password', 'password*()')
cmd = 'echo "password*()"| passwd --stdin root'
test_lib.lib_execute_ssh_cmd(i.managementIp,"root","password",cmd)
host_ops.reconnect_host(i.uuid)
else:
test_util.test_skip("There is no host. Skip test")
test_util.test_dsc('Test KVM Host Infomation: password')
#create vm and stop and migrate
vm = test_stub.create_vr_vm('migrate_stopped_vm', 'imageName_s', 'l3VlanNetwork2')
vm.check()
target_host = test_lib.lib_find_random_host(vm.vm)
vm.stop()
vol_ops.migrate_volume(vm.get_vm().allVolumes[0].uuid, target_host.uuid)
vm.check()
vm.start()
vm.check()
vm.destroy()
vm.expunge()
################################# recover KVM HOST Password #################################
conditions = res_ops.gen_query_conditions('state', '=', 'Enabled')
if res_ops.query_resource(res_ops.HOST, conditions):
kvm_host = res_ops.query_resource(res_ops.HOST, conditions)
for i in kvm_host:
host_ops.update_kvm_host(i.uuid, 'password', 'password')
cmd = 'echo "password"| passwd --stdin root'
test_lib.lib_execute_ssh_cmd(i.managementIp,"root","password*()",cmd)
host_ops.reconnect_host(i.uuid)
else:
test_util.test_skip("There is no host. Skip test")
test_util.test_dsc('Test KVM Host Infomation: password')
test_util.test_pass('Migrate Stopped VM with special_password Test Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:55,代码来源:test_cold_migrate_change_host_special_password.py
示例15: test
def test():
global test_obj_dict
global origin_ip
global test_config_des
global deploy_tmpt_src
global deploy_tmpt_des
origin_ip = os.environ['ZSTACK_BUILT_IN_HTTP_SERVER_IP']
test_util.test_dsc('Create test vm to test zstack all installation in CentOS7.')
# image_name = os.environ.get('imageName_i_c7')
# image_name = "zstack_iso_centos7_141"
image_name = os.environ.get('imageName_i_offline')
vm = test_stub.create_vlan_vm(image_name)
test_obj_dict.add_vm(vm)
if os.environ.get('zstackManagementIp') == None:
vm.check()
else:
time.sleep(60)
vm_inv = vm.get_vm()
vm_ip = vm_inv.vmNics[0].ip
target_file = '/root/zstack-all-in-one.tgz'
test_stub.prepare_test_env(vm_inv, target_file)
ssh_cmd = 'ssh -oStrictHostKeyChecking=no -oCheckHostIP=no -oUserKnownHostsFile=/dev/null %s' % vm_ip
test_stub.execute_all_install(ssh_cmd, target_file, tmp_file)
# test_stub.check_installation(ssh_cmd, tmp_file, vm_inv)
test_util.test_dsc("Prepare Config Files")
cmd = "cp %s %s" % (deploy_tmpt_src, deploy_tmpt_des)
os.system(cmd)
cmd = "sed -i \"s/templateIP/%s/g\" %s" % (vm_ip, deploy_tmpt_des)
os.system(cmd)
test_config_obj = test_util.TestConfig(test_config_des)
test_config = test_config_obj.get_test_config()
all_config = test_config_obj.get_deploy_config()
deploy_config = all_config.deployerConfig
cmd = "vconfig add eth0 213"
rsp = test_lib.lib_execute_ssh_cmd(vm_ip, 'root', 'password', cmd, 180)
cmd = "vconfig add eth0 212"
rsp = test_lib.lib_execute_ssh_cmd(vm_ip, 'root', 'password', cmd, 180)
os.environ['ZSTACK_BUILT_IN_HTTP_SERVER_IP'] = vm_ip
deploy_operations.deploy_initial_database(deploy_config)
test_util.test_dsc("Clean up")
os.environ['ZSTACK_BUILT_IN_HTTP_SERVER_IP'] = origin_ip
os.system('rm -f %s' % deploy_tmpt_des)
os.system('rm -f %s' % tmp_file)
vm.destroy()
test_obj_dict.rm_vm(vm)
test_util.test_pass('ZStack installation Test Success')
开发者ID:mrwangxc,项目名称:zstack-woodpecker,代码行数:54,代码来源:test_zs_inst_and_deploy_on_cos7.py
示例16: set_host_cpu_model
def set_host_cpu_model(ip,model=None):
i = 0
_model = get_host_cpu_model(ip)
test_util.test_logger(_model)
while _model != model:
i += 1
if i > 5:
test_util.test_fail("set host cpu model faild")
cmd = '''sed -i "s/'{}'/'{}'/g" /usr/share/libvirt/cpu_map.xml'''.format(_model,model)
test_lib.lib_execute_ssh_cmd(ip, 'root', 'password',cmd)
cmd = "systemctl restart libvirtd"
test_lib.lib_execute_ssh_cmd(ip, 'root', 'password',cmd)
_model = get_host_cpu_model(ip)
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:13,代码来源:test_stub.py
示例17: test
def test():
global test_obj_dict
global test_file_src
global test_file_des
global ct_original
if con_ops.get_global_config_value('vm', 'cleanTraffic') == 'false' :
ct_original='false'
con_ops.change_global_config('vm', 'cleanTraffic', 'true')
else:
ct_original='true'
vm = test_stub.create_basic_vm()
test_obj_dict.add_vm(vm)
vm.check()
vm_inv = vm.get_vm()
vm_ip = vm_inv.vmNics[0].ip
new_vm_ip = '172.20.1.1'
if new_vm_ip == vm_ip:
new_vm_ip = '172.20.1.2'
test_util.test_dsc("Prepare Test File")
cmd = "cp %s %s" % (test_file_src, test_file_des)
os.system(cmd)
cmd = "sed -i \"s/TemplateNodeIP/%s/g\" %s" % (node_ip, test_file_des)
os.system(cmd)
cmd = "sed -i \"s/TemplateOriginalIP/%s/g\" %s" % (vm_ip, test_file_des)
os.system(cmd)
cmd = "sed -i \"s/TemplateTestIP/%s/g\" %s" % (new_vm_ip, test_file_des)
os.system(cmd)
target_file = "/home/change_ip_test.sh"
test_stub.scp_file_to_vm(vm_inv, test_file_des, target_file)
cmd = "chmod +x %s" % target_file
rsp = test_lib.lib_execute_ssh_cmd(vm_ip, 'root', 'password', cmd, 180)
rsp = test_lib.lib_execute_ssh_cmd(vm_ip, 'root', 'password', target_file, 180)
time.sleep(60)
cmd = "cat /home/ip_spoofing_result"
rsp = test_lib.lib_execute_ssh_cmd(vm_ip, 'root', 'password', cmd, 180)
if rsp[0] != "1":
test_util.test_fail(rsp)
vm.destroy()
test_obj_dict.rm_vm(vm)
con_ops.change_global_config('vm', 'cleanTraffic', ct_original)
os.system('rm -f %s' % test_file_des)
test_util.test_pass('IP Spoofing Test Success')
开发者ID:mrwangxc,项目名称:zstack-woodpecker,代码行数:51,代码来源:test_ip_spoofing.py
示例18: recover_host_vlan
def recover_host_vlan(host_vm, scenarioConfig, deploy_config):
host_ip = host_vm.ip_
test_lib.lib_wait_target_up(host_ip, '22', 120)
host_config = sce_ops.get_scenario_config_vm(host_vm.name_,scenarioConfig)
for l3network in xmlobject.safe_list(host_config.l3Networks.l3Network):
if hasattr(l3network, 'l2NetworkRef'):
for l2networkref in xmlobject.safe_list(l3network.l2NetworkRef):
nic_name = sce_ops.get_ref_l2_nic_name(l2networkref.text_, deploy_config)
if nic_name.find('.') >= 0 :
vlan = nic_name.split('.')[1]
test_util.test_logger('[vm:] %s %s is created.' % (host_ip, nic_name.replace("eth", "zsn")))
cmd = 'vconfig add %s %s' % (nic_name.split('.')[0].replace("eth", "zsn"), vlan)
test_lib.lib_execute_ssh_cmd(host_ip, host_config.imageUsername_, host_config.imagePassword_, cmd)
return True
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:14,代码来源:test_stub.py
示例19: compare
def compare(ps, vm, backup):
test_util.test_logger("-----------------compare----------------")
# find vm_host
host = test_lib.lib_find_host_by_vm(vm.vm)
cond = res_ops.gen_query_conditions("type", '=', "ImageStoreBackupStorage")
bs = res_ops.query_resource(res_ops.BACKUP_STORAGE, cond)[0]
root_volume = test_lib.lib_get_root_volume(vm.get_vm())
for i in test_lib.lib_get_data_volumes(vm.get_vm()):
if i.name == "DATA-for-test_vm":
data_volume = i
vm_path = root_volume.installPath
data_path = data_volume.installPath
if ps.type == "SharedBlock":
vm_path = "/dev/" + root_volume.installPath.split("/")[2] + "/" + root_volume.installPath.split("/")[3]
data_path = "/dev/" + data_volume.installPath.split("/")[2] + "/" + data_volume.installPath.split("/")[3]
test_util.test_logger(vm_path)
test_util.test_logger(data_path)
for i in backup:
if i.type == "Root":
name = i.backupStorageRefs[0].installPath.split("/")[2]
id = i.backupStorageRefs[0].installPath.split("/")[3]
if i.type == "Data" and "DATA-for-test_vm" in i.metadata:
name1 = i.backupStorageRefs[0].installPath.split("/")[2]
id1 = i.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, vm_path, id)
cmd1 = "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;" % (id1, bs.hostname, id1, name1, id1, data_path, id1)
# 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")
result = test_lib.lib_execute_ssh_cmd(host.managementIp, "root", "password", cmd1, timeout=300)
if result != "Images are identical.\n":
test_util.test_fail("compare vm_data_volume & image created by backup")
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:48,代码来源:test_del_backup_full.py
示例20: get_qcow_backing_file_by_ip
def get_qcow_backing_file_by_ip(installPath, hostIp):
'''
get backing file with given install Path
'''
cmd = 'qemu-img info %s | grep "backing file:" | awk \'{print $3}\'' % installPath
return test_lib.lib_execute_ssh_cmd(hostIp, 'root', 'password', cmd)
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:7,代码来源:zstack_kvm_snapshot_checker.py
注:本文中的zstackwoodpecker.test_lib.lib_execute_ssh_cmd函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论