本文整理汇总了Python中xmlrpclib.loads函数的典型用法代码示例。如果您正苦于以下问题:Python loads函数的具体用法?Python loads怎么用?Python loads使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了loads函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_xml_dumps
def test_xml_dumps():
"""
Test the `ipalib.rpc.xml_dumps` function.
"""
f = rpc.xml_dumps
params = (binary_bytes, utf8_bytes, unicode_str, None)
# Test serializing an RPC request:
data = f(params, API_VERSION, b'the_method')
(p, m) = loads(data)
assert_equal(m, u'the_method')
assert type(p) is tuple
assert rpc.xml_unwrap(p) == params
# Test serializing an RPC response:
data = f((params,), API_VERSION, methodresponse=True)
(tup, m) = loads(data)
assert m is None
assert len(tup) == 1
assert type(tup) is tuple
assert rpc.xml_unwrap(tup[0]) == params
# Test serializing an RPC response containing a Fault:
fault = Fault(69, unicode_str)
data = f(fault, API_VERSION, methodresponse=True)
e = raises(Fault, loads, data)
assert e.faultCode == 69
assert_equal(e.faultString, unicode_str)
开发者ID:andygabby,项目名称:freeipa,代码行数:28,代码来源:test_rpc.py
示例2: makedict
def makedict(xmlnodes):
nodelist = []
for node in xmlnodes[0].childNodes:
#print job.childNodes
newdict = {}
for attr in node.childNodes:
if not attr.hasChildNodes():
element_data = ''
else:
element_data = attr.childNodes[0].data
# print attr.nodeName, element_data
attr_type = attr.getAttribute('type')
if attr_type == 'bool':
element_data = bool(element_data)
elif attr_type == 'int':
element_data = int(element_data)
elif attr_type == 'float':
element_data = float(element_data)
elif attr_type == 'dict':
attrdict = {}
for x,y in xmlrpclib.loads(element_data)[0]:
attrdict.update({x:y})
element_data = attrdict
elif attr_type == 'list':
# load list data using xmlrpc marshalling
element_data = list(xmlrpclib.loads(element_data)[0])
newdict.update({attr.nodeName:element_data})
nodelist.append(newdict)
return nodelist
开发者ID:benmcclelland,项目名称:cobalt-orcm,代码行数:29,代码来源:cdump.py
示例3: test_xmlrequests
def test_xmlrequests(self):
data = (
'<?xml version="1.0"?><methodCall><methodName>system.listMethods</methodName><params></params></methodCall>'
)
response = self.client.post(self.rpc_path, data, "text/xml")
self.assertEqual(response.status_code, 200)
xmlrpclib.loads(response.content) # this will throw an exception with bad data
开发者ID:HalasNet,项目名称:felix,代码行数:7,代码来源:test_rpcviews.py
示例4: test_kwargs
def test_kwargs(self):
xml = dumps((1, 2), 'kwargstest')
ret = self.dispatcher.dispatch(xml)
out, name = loads(ret)
self.assertFalse(out[0])
ret = self.dispatcher.dispatch(xml, c=1)
out, name = loads(ret)
self.assertTrue(out[0])
xml = dumps((1,),'requestargtest')
ret = self.dispatcher.dispatch(xml, request=True)
out, name = loads(ret)
self.assertTrue(out[0])
xml = """<?xml version='1.0'?>
<methodCall>
<methodName>withoutargstest</methodName>
<params>
</params>
</methodCall>
"""
ret = self.dispatcher.dispatch(xml, request='fakerequest')
out, name = loads(ret)
self.assertTrue(out[0])
开发者ID:ngaranko,项目名称:rpc4django,代码行数:25,代码来源:test_xmlrpcdispatcher.py
示例5: test_kwargs
def test_kwargs(self):
xml = dumps((1, 2), 'kwargstest')
ret = self.dispatcher.dispatch(xml)
out, name = loads(ret)
self.assertFalse(out[0])
ret = self.dispatcher.dispatch(xml, c=1)
out, name = loads(ret)
self.assertTrue(out[0])
开发者ID:mksh,项目名称:rpc4django,代码行数:9,代码来源:test_xmlrpcdispatcher.py
示例6: test_bad_xml_input
def test_bad_xml_input(self):
request = Request.blank("/", method="POST", body="<foo")
response = request.get_response(self.app)
self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, response.body)
try:
xmlrpclib.loads(response.body)
except xmlrpclib.Fault, e:
self.assertEquals(repr(e), "<Fault -32700: 'Not well formed.'>")
开发者ID:Hazer,项目名称:WebCore,代码行数:9,代码来源:test_rpc_xml.py
示例7: wrap
def wrap(self, func, *p, **kw):
try:
try:
parms, method = xmlrpclib.loads(request.body)
except:
parms, method = xmlrpclib.loads(urllib.unquote_plus(request.body))
rpcresponse = xmlrpclib.dumps((func(p[0], *parms),), methodresponse=1)
except xmlrpclib.Fault, fault:
rpcresponse = xmlrpclib.dumps(fault)
开发者ID:TurboGears,项目名称:tgext.xmlrpc,代码行数:9,代码来源:controllers.py
示例8: test_datetime_before_1900
def test_datetime_before_1900(self):
# same as before but with a date before 1900
dt = datetime.datetime(1, 02, 10, 11, 41, 23)
s = xmlrpclib.dumps((dt,))
(newdt,), m = xmlrpclib.loads(s, use_datetime=1)
self.assertEqual(newdt, dt)
self.assertEqual(m, None)
(newdt,), m = xmlrpclib.loads(s, use_datetime=0)
self.assertEqual(newdt, xmlrpclib.DateTime('00010210T11:41:23'))
开发者ID:89sos98,项目名称:main,代码行数:10,代码来源:test_xmlrpc.py
示例9: main
def main(argv):
output_python=False
if argv[0] == '-p':
output_python=True
argv.pop(0)
host, methodname = argv[:2]
respxml = do_scgi_xmlrpc_request(host, methodname, convert_params_to_native(argv[2:]))
if not output_python:
print respxml
else:
print xmlrpclib.loads(respxml)[0][0]
开发者ID:WanderingStar,项目名称:arch-rutorrentvpn,代码行数:11,代码来源:xmlrpc2scgi.py
示例10: test_dump_bare_date
def test_dump_bare_date(self):
# This checks that an unwrapped datetime.date object can be handled
# by the marshalling code. This can't be done via test_dump_load()
# since the unmarshaller produces a datetime object
d = datetime.datetime(2005, 02, 10, 11, 41, 23).date()
s = xmlrpclib.dumps((d,))
(newd,), m = xmlrpclib.loads(s, use_datetime=1)
self.assertEquals(newd.date(), d)
self.assertEquals(newd.time(), datetime.time(0, 0, 0))
self.assertEquals(m, None)
(newdt,), m = xmlrpclib.loads(s, use_datetime=0)
self.assertEquals(newdt, xmlrpclib.DateTime('20050210T00:00:00'))
开发者ID:Alex-CS,项目名称:sonify,代码行数:13,代码来源:test_xmlrpc.py
示例11: test_dump_bare_datetime
def test_dump_bare_datetime(self):
# This checks that an unwrapped datetime.date object can be handled
# by the marshalling code. This can't be done via test_dump_load()
# since with use_datetime set to 1 the unmarshaller would create
# datetime objects for the 'datetime[123]' keys as well
dt = datetime.datetime(2005, 02, 10, 11, 41, 23)
s = xmlrpclib.dumps((dt,))
(newdt,), m = xmlrpclib.loads(s, use_datetime=1)
self.assertEqual(newdt, dt)
self.assertEqual(m, None)
(newdt,), m = xmlrpclib.loads(s, use_datetime=0)
self.assertEqual(newdt, xmlrpclib.DateTime('20050210T11:41:23'))
开发者ID:89sos98,项目名称:main,代码行数:13,代码来源:test_xmlrpc.py
示例12: render_POST
def render_POST(self, request):
request.content.seek(0, 0)
request.setHeader("content-type", "text/xml")
try:
if self.useDateTime:
args, functionPath = xmlrpclib.loads(request.content.read(),
use_datetime=True)
else:
# Maintain backwards compatibility with Python < 2.5
args, functionPath = xmlrpclib.loads(request.content.read())
except Exception, e:
f = Fault(self.FAILURE, "Can't deserialize input: %s" % (e,))
self._cbRender(f, request)
开发者ID:jdb,项目名称:twisted,代码行数:13,代码来源:xmlrpc.py
示例13: test_dump_bare_time
def test_dump_bare_time(self):
# This checks that an unwrapped datetime.time object can be handled
# by the marshalling code. This can't be done via test_dump_load()
# since the unmarshaller produces a datetime object
t = datetime.datetime(2005, 02, 10, 11, 41, 23).time()
s = xmlrpclib.dumps((t,))
(newt,), m = xmlrpclib.loads(s, use_datetime=1)
today = datetime.datetime.now().date().strftime("%Y%m%d")
self.assertEquals(newt.time(), t)
self.assertEquals(newt.date(), datetime.datetime.now().date())
self.assertEquals(m, None)
(newdt,), m = xmlrpclib.loads(s, use_datetime=0)
self.assertEquals(newdt, xmlrpclib.DateTime('%sT11:41:23'%today))
开发者ID:Alex-CS,项目名称:sonify,代码行数:14,代码来源:test_xmlrpc.py
示例14: test_it_with_invalid_body
def test_it_with_invalid_body(self):
config = self.config
config.include('pyramid_rpc.xmlrpc')
config.add_xmlrpc_endpoint('rpc', '/api/xmlrpc')
app = config.make_wsgi_app()
app = TestApp(app)
resp = app.post('/api/xmlrpc', content_type='text/xml',
params='<')
try:
xmlrpclib.loads(resp.body)
except xmlrpclib.Fault:
exc = sys.exc_info()[1] # 2.5 compat
self.assertEqual(exc.faultCode, -32700)
else: # pragma: no cover
raise AssertionError
开发者ID:llacroix,项目名称:pyramid_rpc,代码行数:15,代码来源:test_xmlrpc.py
示例15: render
def render(self, request):
request.content.seek(0, 0)
args, functionPath = xmlrpclib.loads(request.content.read())
try:
function = self._getFunction(functionPath)
except Fault, f:
self._cbRender(f, request)
开发者ID:3rdandUrban-dev,项目名称:hadoop-20,代码行数:7,代码来源:socketServers.py
示例16: render
def render(self, request):
"""
override method of xmlrpc python twisted framework
@param request: raw request xmlrpc
@type request: xml str
@return: interpreted request
"""
args, functionPath = xmlrpclib.loads(request.content.read())
s = request.getSession()
try:
s.loggedin
except AttributeError:
s.loggedin = False
# Set session expire timeout
s.sessionTimeout = self.config.sessiontimeout
# Check authorization using HTTP Basic
cleartext_token = self.config.login + ":" + self.config.password
token = request.getUser() + ":" + request.getPassword()
if token != cleartext_token:
logger.error("Invalid login / password for HTTP basic authentication")
request.setResponseCode(http.UNAUTHORIZED)
self._cbRender(
xmlrpc.Fault(http.UNAUTHORIZED, "Unauthorized: invalid credentials to connect to the MMC agent, basic HTTP authentication is required"),
request
)
return server.NOT_DONE_YET
if not s.loggedin:
logger.debug("RPC method call from unauthenticated user: %s" % functionPath + str(args))
# Save the first sent HTTP headers, as they contain some
# informations
s.http_headers = request.received_headers.copy()
else:
logger.debug("RPC method call from user %s: %s" % (s.userid, functionPath + str(args)))
try:
if not s.loggedin and self._needAuth(functionPath):
msg = "Authentication needed: %s" % functionPath
logger.error(msg)
raise Fault(8003, msg)
else:
if not s.loggedin and not self._needAuth(functionPath):
# Provide a security context when a method which doesn't
# require a user authentication is called
s = request.getSession()
s.userid = 'root'
try:
self._associateContext(request, s, s.userid)
except Exception, e:
s.loggedin = False
logger.exception(e)
f = Fault(8004, "MMC agent can't provide a security context")
self._cbRender(f, request)
return server.NOT_DONE_YET
function = self._getFunction(functionPath, request)
except Fault, f:
self._cbRender(f, request)
开发者ID:allgi,项目名称:mmc,代码行数:60,代码来源:agent.py
示例17: test_encode
def test_encode(self):
for type, val in self._data:
try:
v = xrpcrequest._encoder[type](val)
A = [xrpcrequest._values_start,
xrpcrequest._values_head%{'name':'pvx','type':42,'count':43},
xrpcrequest._sample_head%{'stat':1, 'sevr':2, 'secs':3, 'nano':4},
xrpcrequest._sample_start,
v,
xrpcrequest._sample_foot,
xrpcrequest._values_foot,
xrpcrequest._values_end,
]
A = ''.join(A)
X = loads(A)[0][0]
self.assertEqual(X, [{'count': 43,
'meta': {'alarm_high': 0.0,
'alarm_low': 0.0,
'disp_high': 0.0,
'disp_low': 0.0,
'prec': 0,
'type': 1,
'units': '',
'warn_high': 0.0,
'warn_low': 0.0},
'name': 'pvx',
'type': 42,
'values': [{'nano': 4, 'secs': 3, 'sevr': 2, 'stat': 1, 'value': [val]}],
}])
except:
print 'Error in',type,val
raise
开发者ID:aderbenev,项目名称:carchivetools,代码行数:35,代码来源:test_xrpc.py
示例18: do_POST
def do_POST(self):
""" Access point from HTTP handler """
def onParent(pid, fromchild, tochild):
self._server._registerChild(pid, fromchild)
tochild.write("done\n")
tochild.flush()
def onChild(pid, fromparent, toparent):
try:
response = self._server._marshaled_dispatch(data)
self._sendResponse(response)
line = fromparent.readline()
toparent.write("done\n")
toparent.flush()
except:
logger(name="pathos.xmlrpc", level=30).error(print_exc_info())
os._exit(0)
try:
data = self.rfile.read(int(self.headers["content-length"]))
params, method = xmlrpclib.loads(data)
if method == "run":
return spawn2(onParent, onChild)
else:
response = self._server._marshaled_dispatch(data)
self._sendResponse(response)
return
except:
self._debug.error(print_exc_info())
self.send_response(500)
self.end_headers()
return
开发者ID:mash0304,项目名称:pathos,代码行数:33,代码来源:server.py
示例19: make_api_call
def make_api_call(uri, method, args=None, headers=None,
http_headers=None, timeout=None, verbose=False):
if args is None:
args = tuple()
try:
largs = list(args)
largs.insert(0, {'headers': headers})
payload = xmlrpclib.dumps(tuple(largs), methodname=method,
allow_none=True)
response = requests.post(uri, data=payload,
headers=http_headers,
timeout=timeout)
response.raise_for_status()
result = xmlrpclib.loads(response.content,)[0][0]
return result
except xmlrpclib.Fault, e:
# These exceptions are formed from the XML-RPC spec
# http://xmlrpc-epi.sourceforge.net/specs/rfc.fault_codes.php
error_mapping = {
'-32700': NotWellFormed,
'-32701': UnsupportedEncoding,
'-32702': InvalidCharacter,
'-32600': SpecViolation,
'-32601': MethodNotFound,
'-32602': InvalidMethodParameters,
'-32603': InternalError,
'-32500': ApplicationError,
'-32400': RemoteSystemError,
'-32300': TransportError,
}
raise error_mapping.get(e.faultCode, SoftLayerAPIError)(
e.faultCode, e.faultString)
开发者ID:quiteliderally,项目名称:softlayer-api-python-client,代码行数:35,代码来源:requests_transport.py
示例20: continue_request
def continue_request (self, data, request):
params, method = xmlrpclib.loads (data)
try:
# generate response
try:
response = self.call (method, params)
#if type(response) == type(()):
response = (response,)
except:
# report exception back to server
response = xmlrpclib.dumps (
xmlrpclib.Fault (1, "%s:%s" % (sys.exc_type, sys.exc_value))
)
else:
response = xmlrpclib.dumps (response, methodresponse=1)
except:
import traceback
traceback.print_exc()
# internal error, report as HTTP server error
request.error (500)
else:
# got a valid XML RPC response
request['Content-Type'] = 'text/xml'
request.push (response)
request.done()
开发者ID:Lukc,项目名称:ospace-lukc,代码行数:25,代码来源:xmlrpc_handler.py
注:本文中的xmlrpclib.loads函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论