本文整理汇总了Python中zstacklib.utils.http.json_dump_post函数的典型用法代码示例。如果您正苦于以下问题:Python json_dump_post函数的具体用法?Python json_dump_post怎么用?Python json_dump_post使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了json_dump_post函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test
def test():
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.
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()
deploy_operations.deploy_initial_database(test_lib.deploy_config)
if os.path.exists(EXTRA_SUITE_SETUP_SCRIPT):
os.system("bash %s" % EXTRA_SUITE_SETUP_SCRIPT)
test_util.test_pass('Suite Setup Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:29,代码来源:suite_setup.py
示例2: test
def test():
#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.
#linux.create_vlan_eth("eth0", 10, "10.0.0.200", "255.255.255.0")
#linux.create_vlan_eth("eth0", 11, "10.0.1.200", "255.255.255.0")
#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("eth0", 10)
linux.create_vlan_eth("eth0", 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()
hosts = test_lib.lib_get_all_hosts_from_plan()
if type(hosts) != type([]):
hosts = [hosts]
for host in hosts:
cmd.ethname = 'eth0'
cmd.vlan = 10
http.json_dump_post(testagent.build_http_path(host.managementIp_, host_plugin.CREATE_VLAN_DEVICE_PATH), cmd)
cmd.vlan = 11
http.json_dump_post(testagent.build_http_path(host.managementIp_, host_plugin.CREATE_VLAN_DEVICE_PATH), cmd)
test_lib.setup_plan.execute_plan_without_deploy_test_agent()
deploy_operations.deploy_initial_database(test_lib.deploy_config)
if os.path.exists(EXTRA_SUITE_SETUP_SCRIPT):
os.system("bash %s" % EXTRA_SUITE_SETUP_SCRIPT)
test_util.test_pass('Suite Setup Success')
开发者ID:tianshangjun,项目名称:zstack-woodpecker,代码行数:28,代码来源:suite_setup.py
示例3: testName
def testName(self):
server = sftpbackupstorage.SftpBackupStorageAgent()
server.http_server.register_sync_uri('/testcallback', self.callback)
server.http_server.start_in_thread()
time.sleep(2)
cmd = sftpbackupstorage.ConnectCmd()
cmd.storagePath = "/tmp"
#url = sftpbackupstorage._build_url_for_test([sftpbackupstorage.SftpBackupStorageAgent.CONNECT_PATH])
url = 'http://localhost:7171%s' % sftpbackupstorage.SftpBackupStorageAgent.CONNECT_PATH
print url
rsp = http.json_dump_post(url, cmd)
cmd = sftpbackupstorage.DownloadCmd()
cmd.accountUuid = uuidhelper.uuid()
cmd.bits = 64
cmd.description = "Test"
cmd.format = sftpbackupstorage.SftpBackupStorageAgent.IMAGE_TEMPLATE
cmd.guestOsType = "rpm"
cmd.hypervisorType = "KVM"
cmd.imageUuid = uuidhelper.uuid()
cmd.name = "test"
cmd.timeout = 60
cmd.url = "http://yum.puppetlabs.com/el/6/products/i386/puppetlabs-release-6-6.noarch.rpm"
cmd.urlScheme = "http"
url = 'http://localhost:7171%s' % sftpbackupstorage.SftpBackupStorageAgent.DOWNLOAD_IMAGE_PATH
print url
rsp = http.json_dump_post(url, cmd, headers={http.TASK_UUID:uuidhelper.uuid(), http.CALLBACK_URI:self.CALLBACK_URL})
print "post back"
time.sleep(20)
server.http_server.stop()
开发者ID:QiRaining,项目名称:zstack-utility,代码行数:32,代码来源:test_download.py
示例4: test
def test():
# Clear ECS instance remained in Aliyun
hybrid.add_datacenter_iz()
hybrid.tear_down()
clean_util.cleanup_all_vms_violently()
clean_util.cleanup_none_vm_volumes_violently()
clean_util.umount_all_primary_storages_violently()
clean_util.cleanup_backup_storage()
#linux.remove_vlan_eth("eth0", 10)
#linux.remove_vlan_eth("eth0", 11)
cmd = host_plugin.DeleteVlanDeviceCmd()
cmd.vlan_ethname = 'eth0.10'
hosts = test_lib.lib_get_all_hosts_from_plan()
if type(hosts) != type([]):
hosts = [hosts]
for host in hosts:
http.json_dump_post(testagent.build_http_path(host.managementIp_, host_plugin.DELETE_VLAN_DEVICE_PATH), cmd)
cmd.vlan_ethname = 'eth0.11'
for host in hosts:
http.json_dump_post(testagent.build_http_path(host.managementIp_, host_plugin.DELETE_VLAN_DEVICE_PATH), cmd)
test_lib.setup_plan.stop_node()
test_lib.lib_cleanup_host_ip_dict()
test_util.test_pass('Hybrid Teardown Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:26,代码来源:suite_teardown.py
示例5: report
def report(self):
cmd = ProgressReportCmd()
cmd.serverUuid = Report.serverUuid
cmd.processType = self.processType
cmd.progress = self.progress
cmd.resourceUuid = self.resourceUuid
logger.debug("url: %s, progress: %s, header: %s", Report.url, cmd.progress, self.header)
http.json_dump_post(Report.url, cmd, self.header)
开发者ID:ShaofeiWang,项目名称:zstack-utility,代码行数:8,代码来源:report.py
示例6: _rm_folder_contents_violently
def _rm_folder_contents_violently(host_ip, path):
cmd = host_plugin.HostShellCmd()
cmd.command = "rm -rf %s/*" % path
try:
http.json_dump_post(testagent.build_http_path(host_ip, host_plugin.HOST_SHELL_CMD_PATH), cmd)
except Exception as e:
err = linux.get_exception_stacktrace()
test_util.test_logger("Fail to delete contents in folder: %s in Host: %s" % (path, host_ip))
test_util.test_logger(err)
test_util.test_logger("Successfully delete contents in folder: %s in Host: %s" % (path, host_ip))
开发者ID:tianshangjun,项目名称:zstack-woodpecker,代码行数:11,代码来源:clean_util.py
示例7: _umount_folder_violently
def _umount_folder_violently(host_ip, path):
cmd = host_plugin.HostShellCmd()
cmd.command = "umount %s" % path
try:
http.json_dump_post(testagent.build_http_path(host_ip, host_plugin.HOST_SHELL_CMD_PATH), cmd)
except Exception as e:
err = linux.get_exception_stacktrace()
test_util.test_logger("Fail to umount folder: %s in Host: %s" % (path, host_ip))
test_util.test_logger(err)
test_util.test_logger("Umount folder: %s in Host: %s" % (path, host_ip))
开发者ID:tianshangjun,项目名称:zstack-woodpecker,代码行数:11,代码来源:clean_util.py
示例8: report_to_management_node
def report_to_management_node():
cmd = ReportPsStatusCmd()
cmd.psUuids = ps_uuids
cmd.hostUuid = host_uuid
cmd.psStatus = ps_status
cmd.reason = reason
logger.debug(
'primary storage[psList:%s] has new connection status[%s], report it to %s' % (
ps_uuids, ps_status, url))
http.json_dump_post(url, cmd, {'commandpath': '/kvm/reportstoragestatus'})
开发者ID:zstackorg,项目名称:zstack-utility,代码行数:11,代码来源:ha_plugin.py
示例9: 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'" % (EXTRA_SUITE_SETUP_SCRIPT, mn_ips))
elif os.path.exists(EXTRA_SUITE_SETUP_SCRIPT):
os.system("bash %s" % (EXTRA_SUITE_SETUP_SCRIPT))
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_))
mn_ip = res_ops.query_resource(res_ops.MANAGEMENT_NODE)[0].hostName
if test_lib.ver_ge_zstack_2_0(mn_ip):
test_lib.lib_set_allow_live_migration_local_storage('true')
test_lib.lib_set_primary_storage_imagecache_gc_interval(1)
test_lib.ensure_recover_script_l2_correct()
if test_lib.lib_is_storage_network_separate():
add_ps_network_gateway_sys_tag()
test_util.test_pass('Suite Setup Success')
开发者ID:zstackorg,项目名称:zstack-woodpecker,代码行数:54,代码来源:suite_setup.py
示例10: testName
def testName(self):
cmd = dnsmasq.AddDhcpEntryCmd()
nic = DhcpInfo()
nic.ip = "10.0.0.10"
nic.mac = "50:e5:49:c9:65:a3"
nic.gateway = "10.0.0.1"
nic.netmask = "255.255.255.0"
nic.hostname = 'host1'
nic.dns = ['8.8.8.8', '75.75.75.75']
nic1 = DhcpInfo()
nic1.ip = "10.0.0.13"
nic1.mac = "50:e5:49:c9:65:b0"
nic1.netmask = "255.255.255.0"
nic1.hostname = 'host2'
nic1.dns = ['8.8.8.8', '75.75.75.75']
nic2 = DhcpInfo()
nic2.ip = "10.0.0.12"
nic2.mac = "50:e5:49:c9:65:b1"
nic2.netmask = "255.255.255.0"
nic2.hostname = 'host3'
cmd.dhcpEntries = [nic, nic1, nic2]
cmd.rebuild = True
rsp = http.json_dump_post('http://localhost:7272/adddhcp', cmd, headers={http.TASK_UUID:uuidhelper.uuid(), http.CALLBACK_URI:self.CALLBACK_URL})
time.sleep(10)
self.service.stop()
开发者ID:ShaofeiWang,项目名称:zstack-utility,代码行数:30,代码来源:test_configure_dnsmasq.py
示例11: async_call_wait_for_complete
def async_call_wait_for_complete(self, apicmd, exception_on_error=True, interval=500, fail_soon=False):
self._check_not_none_field(apicmd)
timeout = apicmd.timeout
if not timeout:
timeout = 1800000
cmd = {apicmd.FULL_NAME: apicmd}
logger.debug("async call[url: %s, request: %s]" % (self.api_url, jsonobject.dumps(cmd)))
jstr = http.json_dump_post(self.api_url, cmd, fail_soon=fail_soon)
rsp = jsonobject.loads(jstr)
if rsp.state == 'Done':
logger.debug("async call[url: %s, response: %s]" % (self.api_url, rsp.result))
reply = jsonobject.loads(rsp.result)
(name, event) = (reply.__dict__.items()[0])
if exception_on_error and not event.success:
raise ApiError('API call[%s] failed because %s' % (name, self._error_code_to_string(event.error)))
return name, event
curr = 0
finterval = float(float(interval) / float(1000))
ret_uuid = rsp.uuid
while rsp.state != 'Done' and curr < timeout:
time.sleep(finterval)
rsp = self._get_response(ret_uuid)
curr += interval
if curr >= timeout:
raise ApiError('API call[%s] timeout after %dms' % (apicmd.FULL_NAME, curr))
logger.debug("async call[url: %s, response: %s] after %dms" % (self.api_url, rsp.result, curr))
reply = jsonobject.loads(rsp.result)
(name, event) = (reply.__dict__.items()[0])
if exception_on_error and not event.success:
raise ApiError('API call[%s] failed because %s' % (name, self._error_code_to_string(event.error)))
return name, event
开发者ID:zstackorg,项目名称:zstack-utility,代码行数:34,代码来源:api.py
示例12: check
def check(self):
super(zstack_kvm_volume_file_checker, self).check()
volume = self.test_obj.volume
volume_installPath = volume.installPath
if not volume_installPath:
test_util.test_logger('Check result: [installPath] is Null for [volume uuid: ] %s. Can not check volume file existence' % volume.uuid)
return self.judge(False)
host = test_lib.lib_get_volume_host(volume)
if not host:
test_util.test_logger('Check result: can not find Host, who is belonged to same Zone Uuid of [volume uuid: ] %s. Can not check volume file existence' % volume.uuid)
return self.judge(False)
cmd = host_plugin.HostShellCmd()
file_exist = "file_exist"
cmd.command = '[ -f %s ] && echo %s' % (volume_installPath, file_exist)
rspstr = http.json_dump_post(testagent.build_http_path(host.managementIp, host_plugin.HOST_SHELL_CMD_PATH), cmd)
rsp = jsonobject.loads(rspstr)
output = jsonobject.dumps(rsp.stdout)
if file_exist in output:
test_util.test_logger('Check result: [volume file: ] %s exist on [host name:] %s .' % (volume.uuid, host.name))
return self.judge(True)
else:
test_util.test_logger('Check result: [volume file: ] %s does not exist on [host name:] %s .' % (volume.uuid, host.name))
return self.judge(False)
开发者ID:chancelq,项目名称:zstack-woodpecker,代码行数:25,代码来源:zstack_kvm_volume_checker.py
示例13: _wait_echo
def _wait_echo(target_ip):
try:
rspstr = http.json_dump_post(testagent.build_http_path(target_ip, host_plugin.ECHO_PATH))
except:
print('zstack-testagent does not startup, will try again ...')
return False
return True
开发者ID:dennis-sun-chao,项目名称:zstack-woodpecker,代码行数:7,代码来源:setup_actions.py
示例14: testName
def testName(self):
self.mount()
cmd = nfs_primarystorage_plugin.GetCapacityCmd()
url = kvmagent._build_url_for_test([nfs_primarystorage_plugin.GET_CAPACITY_PATH])
rsp = http.json_dump_post(url, cmd, headers={http.TASK_UUID:uuidhelper.uuid(), http.CALLBACK_URI:self.CALLBACK_URL})
time.sleep(5)
self.service.stop()
linux.umount_by_url(self.NFS_URL)
开发者ID:QiRaining,项目名称:zstack-utility,代码行数:8,代码来源:test_nfs_primary_storage_get_capacity.py
示例15: test_mount_failure
def test_mount_failure(self):
cmd = nfs_primarystorage_plugin.MountCmd()
cmd.url = 'this_is_a_wrong_path'
cmd.mountPath = os.path.join('/mnt', uuidhelper.uuid())
callurl = kvmagent._build_url_for_test([nfs_primarystorage_plugin.MOUNT_PATH])
ret = http.json_dump_post(callurl, cmd)
rsp = jsonobject.loads(ret)
self.assertFalse(rsp.success, rsp.error)
开发者ID:ShaofeiWang,项目名称:zstack-utility,代码行数:8,代码来源:test_nfs_primarystorage_plugin.py
示例16: test_check_physical_network_interface
def test_check_physical_network_interface(self):
url = kvmagent._build_url_for_test([network_plugin.CHECK_PHYSICAL_NETWORK_INTERFACE_PATH])
logger.debug('calling %s' % url)
cmd = network_plugin.CheckPhysicalNetworkInterfaceCmd()
cmd.interfaceNames = ['p5p1']
ret = http.json_dump_post(url, body=cmd)
rsp = jsonobject.loads(ret)
self.assertTrue(rsp.success)
开发者ID:QiRaining,项目名称:zstack-utility,代码行数:8,代码来源:test_network_plugin.py
示例17: testName
def testName(self):
url = kvmagent._build_url_for_test([network_plugin.KVM_REALIZE_L2NOVLAN_NETWORK_PATH])
logger.debug('calling %s' % url)
cmd = network_plugin.CreateBridgeCmd()
cmd.physicalInterfaceName = 'eth0'
cmd.bridgeName = 'br_eth0'
rsp = http.json_dump_post(url, cmd, headers={http.TASK_UUID:uuidhelper.uuid(), http.CALLBACK_URI:self.CALLBACK_URL})
time.sleep(2)
self.service.stop()
开发者ID:QiRaining,项目名称:zstack-utility,代码行数:9,代码来源:test_network_plugin_create_bridge.py
示例18: test_hostfact
def test_hostfact(self):
url = kvmagent._build_url_for_test([host_plugin.HOSTFACT_PATH])
cmd = HostFactCmd()
ret = http.json_dump_post(url, body=cmd)
rsp = jsonobject.loads(ret)
self.assertTrue(rsp.success)
self.assertEqual(host_plugin._get_cpu_num(), rsp.cpuNum)
self.assertEqual(host_plugin._get_cpu_speed(), rsp.cpuSpeed)
self.assertEqual(host_plugin._get_total_memory(), rsp.totalMemory)
开发者ID:QiRaining,项目名称:zstack-utility,代码行数:9,代码来源:test_host_plugin.py
示例19: callback
def callback(self, req):
rsp = jsonobject.loads(req[http.REQUEST_BODY])
print jsonobject.dumps(rsp)
cmd = vm_plugin.RebootVmCmd()
cmd.uuid = self.uuid
cmd.timeout = 30
url = kvmagent._build_url_for_test([vm_plugin.KVM_REBOOT_VM_PATH])
rsp = http.json_dump_post(url, cmd, headers={http.TASK_UUID:uuidhelper.uuid(), http.CALLBACK_URI:self.CALLBACK_URL2})
开发者ID:QiRaining,项目名称:zstack-utility,代码行数:9,代码来源:test_vm_plugin_reboot_vm.py
示例20: mount
def mount(self):
cmd = nfs_primarystorage_plugin.MountCmd()
cmd.url = self.NFS_URL
cmd.mountPath = os.path.join('/mnt', uuidhelper.uuid())
callurl = kvmagent._build_url_for_test([nfs_primarystorage_plugin.MOUNT_PATH])
ret = http.json_dump_post(callurl, cmd)
rsp = jsonobject.loads(ret)
self.assertTrue(rsp.success, rsp.error)
self.assertTrue(linux.is_mounted(cmd.url, cmd.mountPath))
开发者ID:QiRaining,项目名称:zstack-utility,代码行数:9,代码来源:test_nfs_primary_storage_get_capacity.py
注:本文中的zstacklib.utils.http.json_dump_post函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论