IntenseWebs/freeipa
3
0
Fork 0
mirror of https://salsa.debian.org/freeipa-team/freeipa.git synced 2025-01-26 16:16:31 -06:00
Code Issues Packages Projects Releases Wiki Activity

rpcserver now uses xml_dumps() and xml_loads() functions

Browse Source
This commit is contained in:
Jason Gerard DeRose 2009-01-16 01:47:03 -07:00
parent 0227a12949
commit f2e479c33e
2 changed files with 24 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
ipaserver/rpcserver.py
View File

@ -18,13 +18,13 @@
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
"""
Execute an RPC request.
RPC server.
"""
from xmlrpclib import dumps, loads, Fault
from xmlrpclib import Fault
from ipalib import Backend
from ipalib.errors import HandledError, CommandError
from ipalib.rpc import xml_wrap, xml_unwrap
from ipalib.errors2 import PublicError, InternalError, CommandError
from ipalib.rpc import xml_dumps, xml_loads
def params_2_args_options(params):
@ -36,25 +36,28 @@ def params_2_args_options(params):
return (params, dict())
class xmlrpc(Backend):
class xmlserver(Backend):
"""
Execution backend for XML-RPC server.
"""
def dispatch(self, method, params):
assert type(method) is str
assert type(params) is tuple
self.info('Received RPC call to %r', method)
self.debug('Received RPC call to %r', method)
if method not in self.Command:
raise CommandError(name=method)
(args, options) = params_2_args_options(xml_unwrap(params))
(args, options) = params_2_args_options(params)
result = self.Command[method](*args, **options)
return (xml_wrap(result),)
return (result,) # Must wrap XML-RPC response in a tuple singleton
def execute(self, data, ccache=None, client_ip=None, locale=None):
def execute(self, data, ccache=None, client_ip=None, languages=None):
try:
(params, method) = loads(data)
(params, method) = xml_loads(data)
response = self.dispatch(method, params)
except Exception, e:
if not isinstance(e, HandledError):
e = UnknownError()
assert isinstance(e, HandledError)
response = Fault(e.code, e.message)
if not isinstance(e, PublicError):
e = InternalError()
assert isinstance(e, PublicError)
response = Fault(e.errno, e.strerror)
return dumps(response)

15
tests/test_ipaserver/test_rpcserver.py
View File

@ -23,7 +23,7 @@ Test the `ipaserver.rpc` module.
from tests.util import create_test_api, raises, PluginTester
from tests.data import unicode_str
from ipalib import errors, Command
from ipalib import errors2, Command
from ipaserver import rpcserver
@ -41,21 +41,20 @@ def test_params_2_args_options():
assert f((options,) + args) == ((options,) + args, dict())
class test_xmlrpc(PluginTester):
class test_xmlserver(PluginTester):
"""
Test the `ipaserver.rpcserver.xmlrpc` plugin.
Test the `ipaserver.rpcserver.xmlserver` plugin.
"""
_plugin = rpcserver.xmlrpc
_plugin = rpcserver.xmlserver
def test_dispatch(self):
"""
Test the `ipaserver.rpcserver.xmlrpc.dispatch` method.
Test the `ipaserver.rpcserver.xmlserver.dispatch` method.
"""
(o, api, home) = self.instance('Backend', in_server=True)
e = raises(errors.CommandError, o.dispatch, 'echo', tuple())
assert str(e) == "Unknown command 'echo'"
assert e.kw['name'] == 'echo'
e = raises(errors2.CommandError, o.dispatch, 'echo', tuple())
assert e.name == 'echo'
class echo(Command):
takes_args = ['arg1', 'arg2+']