1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2010 United States Government as represented by the
4
# Administrator of the National Aeronautics and Space Administration.
7
# Licensed under the Apache License, Version 2.0 (the "License"); you may
8
# not use this file except in compliance with the License. You may obtain
9
# a copy of the License at
11
# http://www.apache.org/licenses/LICENSE-2.0
13
# Unless required by applicable law or agreed to in writing, software
14
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
15
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
16
# License for the specific language governing permissions and limitations
19
Unit Tests for remote procedure calls using queue
23
from twisted.internet import defer
25
from nova import context
26
from nova import flags
34
class RpcTestCase(test.TrialTestCase):
35
"""Test cases for rpc"""
37
super(RpcTestCase, self).setUp()
38
self.conn = rpc.Connection.instance()
39
self.receiver = TestReceiver()
40
self.consumer = rpc.AdapterConsumer(connection=self.conn,
43
self.consumer.attach_to_twisted()
44
self.context = context.get_admin_context()
46
def test_call_succeed(self):
47
"""Get a value through rpc call"""
49
result = yield rpc.call_twisted(self.context,
50
'test', {"method": "echo",
51
"args": {"value": value}})
52
self.assertEqual(value, result)
54
def test_context_passed(self):
55
"""Makes sure a context is passed through rpc call"""
57
result = yield rpc.call_twisted(self.context,
58
'test', {"method": "context",
59
"args": {"value": value}})
60
self.assertEqual(self.context.to_dict(), result)
62
def test_call_exception(self):
63
"""Test that exception gets passed back properly
65
rpc.call returns a RemoteError object. The value of the
66
exception is converted to a string, so we convert it back
67
to an int in the test.
70
self.assertFailure(rpc.call_twisted(self.context, 'test',
72
"args": {"value": value}}),
75
yield rpc.call_twisted(self.context,
76
'test', {"method": "fail",
77
"args": {"value": value}})
78
self.fail("should have thrown rpc.RemoteError")
79
except rpc.RemoteError as exc:
80
self.assertEqual(int(exc.value), value)
83
class TestReceiver(object):
84
"""Simple Proxy class so the consumer has methods to call
86
Uses static methods because we aren't actually storing any state"""
89
def echo(context, value):
90
"""Simply returns whatever value is sent in"""
91
logging.debug("Received %s", value)
92
return defer.succeed(value)
95
def context(context, value):
96
"""Returns dictionary version of context"""
97
logging.debug("Received %s", context)
98
return defer.succeed(context.to_dict())
101
def fail(context, value):
102
"""Raises an exception with the value sent in"""
103
raise Exception(value)
added
removed
nova/tests/rpc_unittest.py
