From a04857a23947d5a520204480ed344e69d0c8cf6a Mon Sep 17 00:00:00 2001 From: Jason Gerard DeRose Date: Fri, 16 Jan 2009 00:00:15 -0700 Subject: Renamed ipaserver.rpc to ipaserver.rpcserver --- ipaserver/rpc.py | 60 ------------------------- ipaserver/rpcserver.py | 60 +++++++++++++++++++++++++ tests/test_ipaserver/test_rpc.py | 80 ---------------------------------- tests/test_ipaserver/test_rpcserver.py | 80 ++++++++++++++++++++++++++++++++++ 4 files changed, 140 insertions(+), 140 deletions(-) delete mode 100644 ipaserver/rpc.py create mode 100644 ipaserver/rpcserver.py delete mode 100644 tests/test_ipaserver/test_rpc.py create mode 100644 tests/test_ipaserver/test_rpcserver.py diff --git a/ipaserver/rpc.py b/ipaserver/rpc.py deleted file mode 100644 index c2cb0bd5..00000000 --- a/ipaserver/rpc.py +++ /dev/null @@ -1,60 +0,0 @@ -# Authors: -# Jason Gerard DeRose -# -# Copyright (C) 2008 Red Hat -# see file 'COPYING' for use and warranty information -# -# This program is free software; you can redistribute it and/or -# modify it under the terms of the GNU General Public License as -# published by the Free Software Foundation; version 2 only -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program; if not, write to the Free Software -# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA - -""" -Execute an RPC request. -""" - -from xmlrpclib import dumps, loads, Fault -from ipalib import Backend -from ipalib.errors import HandledError, CommandError -from ipalib.rpc import xml_wrap, xml_unwrap - - -def params_2_args_options(params): - assert type(params) is tuple - if len(params) == 0: - return (tuple(), dict()) - if type(params[-1]) is dict: - return (params[:-1], params[-1]) - return (params, dict()) - - -class xmlrpc(Backend): - - def dispatch(self, method, params): - assert type(method) is str - assert type(params) is tuple - self.info('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)) - result = self.Command[method](*args, **options) - return (xml_wrap(result),) - - def execute(self, data, ccache=None, client_ip=None, locale=None): - try: - (params, method) = 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) - return dumps(response) diff --git a/ipaserver/rpcserver.py b/ipaserver/rpcserver.py new file mode 100644 index 00000000..c2cb0bd5 --- /dev/null +++ b/ipaserver/rpcserver.py @@ -0,0 +1,60 @@ +# Authors: +# Jason Gerard DeRose +# +# Copyright (C) 2008 Red Hat +# see file 'COPYING' for use and warranty information +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation; version 2 only +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + +""" +Execute an RPC request. +""" + +from xmlrpclib import dumps, loads, Fault +from ipalib import Backend +from ipalib.errors import HandledError, CommandError +from ipalib.rpc import xml_wrap, xml_unwrap + + +def params_2_args_options(params): + assert type(params) is tuple + if len(params) == 0: + return (tuple(), dict()) + if type(params[-1]) is dict: + return (params[:-1], params[-1]) + return (params, dict()) + + +class xmlrpc(Backend): + + def dispatch(self, method, params): + assert type(method) is str + assert type(params) is tuple + self.info('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)) + result = self.Command[method](*args, **options) + return (xml_wrap(result),) + + def execute(self, data, ccache=None, client_ip=None, locale=None): + try: + (params, method) = 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) + return dumps(response) diff --git a/tests/test_ipaserver/test_rpc.py b/tests/test_ipaserver/test_rpc.py deleted file mode 100644 index 07191fda..00000000 --- a/tests/test_ipaserver/test_rpc.py +++ /dev/null @@ -1,80 +0,0 @@ -# Authors: -# Jason Gerard DeRose -# -# Copyright (C) 2008 Red Hat -# see file 'COPYING' for use and warranty information -# -# This program is free software; you can redistribute it and/or -# modify it under the terms of the GNU General Public License as -# published by the Free Software Foundation; version 2 only -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU General Public License for more details. -# -# You should have received a copy of the GNU General Public License -# along with this program; if not, write to the Free Software -# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA - -""" -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 ipaserver import rpc - - -def test_params_2_args_options(): - """ - Test the `ipaserver.rpc.params_2_args_options` function. - """ - f = rpc.params_2_args_options - args = ('Hello', u'world!') - options = dict(one=1, two=u'Two', three='Three') - assert f(tuple()) == (tuple(), dict()) - assert f(args) == (args, dict()) - assert f((options,)) == (tuple(), options) - assert f(args + (options,)) == (args, options) - assert f((options,) + args) == ((options,) + args, dict()) - - -class test_xmlrpc(PluginTester): - """ - Test the `ipaserver.rpc.xmlrpc` plugin. - """ - - _plugin = rpc.xmlrpc - - def test_dispatch(self): - """ - Test the `ipaserver.rpc.xmlrpc.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' - - class echo(Command): - takes_args = ['arg1', 'arg2+'] - takes_options = ['option1?', 'option2?'] - def execute(self, *args, **options): - assert type(args[1]) is tuple - return args + (options,) - - (o, api, home) = self.instance('Backend', echo, in_server=True) - def call(params): - response = o.dispatch('echo', params) - assert type(response) is tuple and len(response) == 1 - return response[0] - arg1 = unicode_str - arg2 = (u'Hello', unicode_str, u'world!') - options = dict(option1=u'How are you?', option2=unicode_str) - assert call((arg1, arg2, options)) == (arg1, arg2, options) - assert call((arg1,) + arg2 + (options,)) == (arg1, arg2, options) - - - def test_execute(self): - (o, api, home) = self.instance('Backend', in_server=True) diff --git a/tests/test_ipaserver/test_rpcserver.py b/tests/test_ipaserver/test_rpcserver.py new file mode 100644 index 00000000..52db611b --- /dev/null +++ b/tests/test_ipaserver/test_rpcserver.py @@ -0,0 +1,80 @@ +# Authors: +# Jason Gerard DeRose +# +# Copyright (C) 2008 Red Hat +# see file 'COPYING' for use and warranty information +# +# This program is free software; you can redistribute it and/or +# modify it under the terms of the GNU General Public License as +# published by the Free Software Foundation; version 2 only +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + +""" +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 ipaserver import rpcserver + + +def test_params_2_args_options(): + """ + Test the `ipaserver.rpcserver.params_2_args_options` function. + """ + f = rpcserver.params_2_args_options + args = ('Hello', u'world!') + options = dict(one=1, two=u'Two', three='Three') + assert f(tuple()) == (tuple(), dict()) + assert f(args) == (args, dict()) + assert f((options,)) == (tuple(), options) + assert f(args + (options,)) == (args, options) + assert f((options,) + args) == ((options,) + args, dict()) + + +class test_xmlrpc(PluginTester): + """ + Test the `ipaserver.rpcserver.xmlrpc` plugin. + """ + + _plugin = rpcserver.xmlrpc + + def test_dispatch(self): + """ + Test the `ipaserver.rpcserver.xmlrpc.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' + + class echo(Command): + takes_args = ['arg1', 'arg2+'] + takes_options = ['option1?', 'option2?'] + def execute(self, *args, **options): + assert type(args[1]) is tuple + return args + (options,) + + (o, api, home) = self.instance('Backend', echo, in_server=True) + def call(params): + response = o.dispatch('echo', params) + assert type(response) is tuple and len(response) == 1 + return response[0] + arg1 = unicode_str + arg2 = (u'Hello', unicode_str, u'world!') + options = dict(option1=u'How are you?', option2=unicode_str) + assert call((arg1, arg2, options)) == (arg1, arg2, options) + assert call((arg1,) + arg2 + (options,)) == (arg1, arg2, options) + + + def test_execute(self): + (o, api, home) = self.instance('Backend', in_server=True) -- cgit