Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_call(self):
method = 'my_rpc_method'
for value in self.standard_data:
value = (value, "other arg")
enc = xmlrpcplus.dumps(value, methodname=method)
_enc = xmlrpc_client.dumps(value, methodname=method, allow_none=1)
self.assertEqual(enc, _enc)
params, method = xmlrpc_client.loads(enc)
self.assertEqual(params, value)
self.assertEqual(method, method)
def test_generator(self):
value = (self.gendata(),)
enc = xmlrpcplus.dumps(value, methodresponse=1)
params, method = xmlrpc_client.loads(enc)
expect = (list(self.gendata()),)
self.assertEqual(params, expect)
self.assertEqual(method, None)
def test_no_i8(self):
# we shouldn't use i8 if we don't have to
data = [
23,
42,
-1024,
2 ** 31 - 1,
-2 ** 31,
[2**31 -1],
{"a": -2 ** 31, "b": 3.14},
]
for value in data:
value = (value,)
enc = xmlrpcplus.dumps(value, methodresponse=1, encoding='us-ascii')
_enc = xmlrpc_client.dumps(value, methodresponse=1, allow_none=1, encoding='us-ascii')
if 'i8' in enc or 'I8' in enc:
raise Exception('i8 used unnecessarily')
self.assertEqual(enc, _enc)
params, method = xmlrpc_client.loads(enc)
self.assertEqual(params, value)
self.assertEqual(method, None)
def test_i8(self):
for value in self.long_data:
value = (value,)
enc = xmlrpcplus.dumps(value, methodresponse=1)
params, method = xmlrpc_client.loads(enc)
self.assertEqual(params, value)
self.assertEqual(method, None)
# and as a call
method = "foomethod"
value = tuple(self.long_data)
enc = xmlrpcplus.dumps(value, methodname=method)
params, method = xmlrpc_client.loads(enc)
self.assertEqual(params, value)
self.assertEqual(method, method)
def test_badargs(self):
wrong_type = ["a", 0, 0.1, [], {}, True]
for value in wrong_type:
with self.assertRaises(TypeError):
xmlrpcplus.dumps(value, methodname="foo")
# responses much be singletons
value = (1, 2, 3)
with self.assertRaises(ValueError):
xmlrpcplus.dumps(value, methodresponse=1)
def runTask(self, handler):
try:
response = (handler.run(),)
# note that we wrap response in a singleton tuple
response = koji.xmlrpcplus.dumps(response, methodresponse=1, allow_none=1)
self.logger.info("RESPONSE: %r" % response)
self.session.host.closeTask(handler.id, response)
return
except koji.xmlrpcplus.Fault as fault:
response = koji.xmlrpcplus.dumps(fault)
tb = ''.join(traceback.format_exception(*sys.exc_info())).replace(r"\n", "\n")
self.logger.warn("FAULT:\n%s" % tb)
except (SystemExit, koji.tasks.ServerExit, KeyboardInterrupt):
#we do not trap these
raise
except koji.tasks.ServerRestart:
#freeing this task will allow the pending restart to take effect
self.session.host.freeTasks([handler.id])
return
except:
tb = ''.join(traceback.format_exception(*sys.exc_info()))
def offline_reply(start_response, msg=None):
"""Send a ServerOffline reply"""
faultCode = koji.ServerOffline.faultCode
if msg is None:
faultString = "server is offline"
else:
faultString = msg
response = dumps(Fault(faultCode, faultString))
if six.PY3:
response = response.encode()
headers = [
('Content-Length', str(len(response))),
('Content-Type', "text/xml"),
]
start_response('200 OK', headers)
return [response]
#we do not trap these
raise
except koji.tasks.ServerRestart:
#freeing this task will allow the pending restart to take effect
self.session.host.freeTasks([handler.id])
return
except:
tb = ''.join(traceback.format_exception(*sys.exc_info()))
self.logger.warn("TRACEBACK: %s" % tb)
# report exception back to server
e_class, e = sys.exc_info()[:2]
faultCode = getattr(e_class, 'faultCode', 1)
if issubclass(e_class, koji.GenericError):
#just pass it through
tb = str(e)
response = koji.xmlrpcplus.dumps(koji.xmlrpcplus.Fault(faultCode, tb))
# if we get here, then we're handling an exception, so fail the task
self.session.host.failTask(handler.id, response)
def _wrap_handler(self, handler, environ):
"""Catch exceptions and encode response of handler"""
# generate response
try:
response = handler(environ)
# wrap response in a singleton tuple
response = (response,)
response = dumps(response, methodresponse=1, marshaller=Marshaller)
except Fault as fault:
self.traceback = True
response = dumps(fault, marshaller=Marshaller)
except:
self.traceback = True
# report exception back to server
e_class, e = sys.exc_info()[:2]
faultCode = getattr(e_class, 'faultCode', 1)
tb_type = context.opts.get('KojiTraceback', None)
tb_str = ''.join(traceback.format_exception(*sys.exc_info()))
if issubclass(e_class, koji.GenericError):
if context.opts.get('KojiDebug'):
if tb_type == "extended":
faultString = koji.format_exc_plus()
else:
faultString = tb_str
def _wrap_handler(self, handler, environ):
"""Catch exceptions and encode response of handler"""
# generate response
try:
response = handler(environ)
# wrap response in a singleton tuple
response = (response,)
response = dumps(response, methodresponse=1, marshaller=Marshaller)
except Fault as fault:
self.traceback = True
response = dumps(fault, marshaller=Marshaller)
except:
self.traceback = True
# report exception back to server
e_class, e = sys.exc_info()[:2]
faultCode = getattr(e_class, 'faultCode', 1)
tb_type = context.opts.get('KojiTraceback', None)
tb_str = ''.join(traceback.format_exception(*sys.exc_info()))
if issubclass(e_class, koji.GenericError):
if context.opts.get('KojiDebug'):
if tb_type == "extended":
faultString = koji.format_exc_plus()
else:
faultString = tb_str
else:
faultString = str(e)
else: