~ubuntu-branches/ubuntu/trusty/freeipa/trusty-proposed

« back to all changes in this revision

Viewing changes to ipatests/test_ipalib/test_rpc.py

  • Committer: Package Import Robot
  • Author(s): Timo Aaltonen
  • Date: 2013-09-03 17:13:27 UTC
  • Revision ID: package-import@ubuntu.com-20130903171327-s3f56x6vxz0o1jq5
Tags: upstream-3.3.4
ImportĀ upstreamĀ versionĀ 3.3.4

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# Authors:
 
2
#   Jason Gerard DeRose <jderose@redhat.com>
 
3
#
 
4
# Copyright (C) 2008  Red Hat
 
5
# see file 'COPYING' for use and warranty information
 
6
#
 
7
# This program is free software; you can redistribute it and/or modify
 
8
# it under the terms of the GNU General Public License as published by
 
9
# the Free Software Foundation, either version 3 of the License, or
 
10
# (at your option) any later version.
 
11
#
 
12
# This program is distributed in the hope that it will be useful,
 
13
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
14
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
15
# GNU General Public License for more details.
 
16
#
 
17
# You should have received a copy of the GNU General Public License
 
18
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
19
 
 
20
"""
 
21
Test the `ipalib.rpc` module.
 
22
"""
 
23
 
 
24
import threading
 
25
from xmlrpclib import Binary, Fault, dumps, loads, ServerProxy
 
26
from ipatests.util import raises, assert_equal, PluginTester, DummyClass
 
27
from ipatests.data import binary_bytes, utf8_bytes, unicode_str
 
28
from ipalib.frontend import Command
 
29
from ipalib.request import context, Connection
 
30
from ipalib import rpc, errors
 
31
 
 
32
 
 
33
std_compound = (binary_bytes, utf8_bytes, unicode_str)
 
34
 
 
35
 
 
36
def dump_n_load(value):
 
37
    (param, method) = loads(
 
38
        dumps((value,), allow_none=True)
 
39
    )
 
40
    return param[0]
 
41
 
 
42
 
 
43
def round_trip(value):
 
44
    return rpc.xml_unwrap(
 
45
        dump_n_load(rpc.xml_wrap(value))
 
46
    )
 
47
 
 
48
 
 
49
def test_round_trip():
 
50
    """
 
51
    Test `ipalib.rpc.xml_wrap` and `ipalib.rpc.xml_unwrap`.
 
52
 
 
53
    This tests the two functions together with ``xmlrpclib.dumps()`` and
 
54
    ``xmlrpclib.loads()`` in a full wrap/dumps/loads/unwrap round trip.
 
55
    """
 
56
    # We first test that our assumptions about xmlrpclib module in the Python
 
57
    # standard library are correct:
 
58
    assert_equal(dump_n_load(utf8_bytes), unicode_str)
 
59
    assert_equal(dump_n_load(unicode_str), unicode_str)
 
60
    assert_equal(dump_n_load(Binary(binary_bytes)).data, binary_bytes)
 
61
    assert isinstance(dump_n_load(Binary(binary_bytes)), Binary)
 
62
    assert type(dump_n_load('hello')) is str
 
63
    assert type(dump_n_load(u'hello')) is str
 
64
    assert_equal(dump_n_load(''), '')
 
65
    assert_equal(dump_n_load(u''), '')
 
66
    assert dump_n_load(None) is None
 
67
 
 
68
    # Now we test our wrap and unwrap methods in combination with dumps, loads:
 
69
    # All str should come back str (because they get wrapped in
 
70
    # xmlrpclib.Binary().  All unicode should come back unicode because str
 
71
    # explicity get decoded by rpc.xml_unwrap() if they weren't already
 
72
    # decoded by xmlrpclib.loads().
 
73
    assert_equal(round_trip(utf8_bytes), utf8_bytes)
 
74
    assert_equal(round_trip(unicode_str), unicode_str)
 
75
    assert_equal(round_trip(binary_bytes), binary_bytes)
 
76
    assert type(round_trip('hello')) is str
 
77
    assert type(round_trip(u'hello')) is unicode
 
78
    assert_equal(round_trip(''), '')
 
79
    assert_equal(round_trip(u''), u'')
 
80
    assert round_trip(None) is None
 
81
    compound = [utf8_bytes, None, binary_bytes, (None, unicode_str),
 
82
        dict(utf8=utf8_bytes, chars=unicode_str, data=binary_bytes)
 
83
    ]
 
84
    assert round_trip(compound) == tuple(compound)
 
85
 
 
86
 
 
87
def test_xml_wrap():
 
88
    """
 
89
    Test the `ipalib.rpc.xml_wrap` function.
 
90
    """
 
91
    f = rpc.xml_wrap
 
92
    assert f([]) == tuple()
 
93
    assert f({}) == dict()
 
94
    b = f('hello')
 
95
    assert isinstance(b, Binary)
 
96
    assert b.data == 'hello'
 
97
    u = f(u'hello')
 
98
    assert type(u) is unicode
 
99
    assert u == u'hello'
 
100
    value = f([dict(one=False, two=u'hello'), None, 'hello'])
 
101
 
 
102
 
 
103
def test_xml_unwrap():
 
104
    """
 
105
    Test the `ipalib.rpc.xml_unwrap` function.
 
106
    """
 
107
    f = rpc.xml_unwrap
 
108
    assert f([]) == tuple()
 
109
    assert f({}) == dict()
 
110
    value = f(Binary(utf8_bytes))
 
111
    assert type(value) is str
 
112
    assert value == utf8_bytes
 
113
    assert f(utf8_bytes) == unicode_str
 
114
    assert f(unicode_str) == unicode_str
 
115
    value = f([True, Binary('hello'), dict(one=1, two=utf8_bytes, three=None)])
 
116
    assert value == (True, 'hello', dict(one=1, two=unicode_str, three=None))
 
117
    assert type(value[1]) is str
 
118
    assert type(value[2]['two']) is unicode
 
119
 
 
120
 
 
121
def test_xml_dumps():
 
122
    """
 
123
    Test the `ipalib.rpc.xml_dumps` function.
 
124
    """
 
125
    f = rpc.xml_dumps
 
126
    params = (binary_bytes, utf8_bytes, unicode_str, None)
 
127
 
 
128
    # Test serializing an RPC request:
 
129
    data = f(params, 'the_method')
 
130
    (p, m) = loads(data)
 
131
    assert_equal(m, u'the_method')
 
132
    assert type(p) is tuple
 
133
    assert rpc.xml_unwrap(p) == params
 
134
 
 
135
    # Test serializing an RPC response:
 
136
    data = f((params,), methodresponse=True)
 
137
    (tup, m) = loads(data)
 
138
    assert m is None
 
139
    assert len(tup) == 1
 
140
    assert type(tup) is tuple
 
141
    assert rpc.xml_unwrap(tup[0]) == params
 
142
 
 
143
    # Test serializing an RPC response containing a Fault:
 
144
    fault = Fault(69, unicode_str)
 
145
    data = f(fault, methodresponse=True)
 
146
    e = raises(Fault, loads, data)
 
147
    assert e.faultCode == 69
 
148
    assert_equal(e.faultString, unicode_str)
 
149
 
 
150
 
 
151
def test_xml_loads():
 
152
    """
 
153
    Test the `ipalib.rpc.xml_loads` function.
 
154
    """
 
155
    f = rpc.xml_loads
 
156
    params = (binary_bytes, utf8_bytes, unicode_str, None)
 
157
    wrapped = rpc.xml_wrap(params)
 
158
 
 
159
    # Test un-serializing an RPC request:
 
160
    data = dumps(wrapped, 'the_method', allow_none=True)
 
161
    (p, m) = f(data)
 
162
    assert_equal(m, u'the_method')
 
163
    assert_equal(p, params)
 
164
 
 
165
    # Test un-serializing an RPC response:
 
166
    data = dumps((wrapped,), methodresponse=True, allow_none=True)
 
167
    (tup, m) = f(data)
 
168
    assert m is None
 
169
    assert len(tup) == 1
 
170
    assert type(tup) is tuple
 
171
    assert_equal(tup[0], params)
 
172
 
 
173
    # Test un-serializing an RPC response containing a Fault:
 
174
    for error in (unicode_str, u'hello'):
 
175
        fault = Fault(69, error)
 
176
        data = dumps(fault, methodresponse=True, allow_none=True, encoding='UTF-8')
 
177
        e = raises(Fault, f, data)
 
178
        assert e.faultCode == 69
 
179
        assert_equal(e.faultString, error)
 
180
        assert type(e.faultString) is unicode
 
181
 
 
182
 
 
183
class test_xmlclient(PluginTester):
 
184
    """
 
185
    Test the `ipalib.rpc.xmlclient` plugin.
 
186
    """
 
187
    _plugin = rpc.xmlclient
 
188
 
 
189
    def test_forward(self):
 
190
        """
 
191
        Test the `ipalib.rpc.xmlclient.forward` method.
 
192
        """
 
193
        class user_add(Command):
 
194
            pass
 
195
 
 
196
        # Test that ValueError is raised when forwarding a command that is not
 
197
        # in api.Command:
 
198
        (o, api, home) = self.instance('Backend', in_server=False)
 
199
        e = raises(ValueError, o.forward, 'user_add')
 
200
        assert str(e) == '%s.forward(): %r not in api.Command' % (
 
201
            'xmlclient', 'user_add'
 
202
        )
 
203
 
 
204
        (o, api, home) = self.instance('Backend', user_add, in_server=False)
 
205
        args = (binary_bytes, utf8_bytes, unicode_str)
 
206
        kw = dict(one=binary_bytes, two=utf8_bytes, three=unicode_str)
 
207
        params = [args, kw]
 
208
        result = (unicode_str, binary_bytes, utf8_bytes)
 
209
        conn = DummyClass(
 
210
            (
 
211
                'user_add',
 
212
                rpc.xml_wrap(params),
 
213
                {},
 
214
                rpc.xml_wrap(result),
 
215
            ),
 
216
            (
 
217
                'user_add',
 
218
                rpc.xml_wrap(params),
 
219
                {},
 
220
                Fault(3007, u"'four' is required"),  # RequirementError
 
221
            ),
 
222
            (
 
223
                'user_add',
 
224
                rpc.xml_wrap(params),
 
225
                {},
 
226
                Fault(700, u'no such error'),  # There is no error 700
 
227
            ),
 
228
 
 
229
        )
 
230
        context.xmlclient = Connection(conn, lambda: None)
 
231
 
 
232
        # Test with a successful return value:
 
233
        assert o.forward('user_add', *args, **kw) == result
 
234
 
 
235
        # Test with an errno the client knows:
 
236
        e = raises(errors.RequirementError, o.forward, 'user_add', *args, **kw)
 
237
        assert_equal(e.args[0], u"'four' is required")
 
238
 
 
239
        # Test with an errno the client doesn't know
 
240
        e = raises(errors.UnknownError, o.forward, 'user_add', *args, **kw)
 
241
        assert_equal(e.code, 700)
 
242
        assert_equal(e.error, u'no such error')
 
243
 
 
244
        assert context.xmlclient.conn._calledall() is True