← Back to branch summary

~ubuntu-branches/ubuntu/saucy/nova/saucy-proposed

« back to all changes in this revision

Viewing changes to nova/tests/rpc/test_proxy.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2012-05-24 13:12:53 UTC
  • mfrom: (1.1.55)
  • Revision ID: package-import@ubuntu.com-20120524131253-ommql08fg1en06ut
Tags: 2012.2~f1-0ubuntu1
https://launchpad.net/bugs/907197
* New upstream release.
* Prepare for quantal:
  - Dropped debian/patches/upstream/0006-Use-project_id-in-ec2.cloud._format_image.patch
  - Dropped debian/patches/upstream/0005-Populate-image-properties-with-project_id-again.patch
  - Dropped debian/patches/upstream/0004-Fixed-bug-962840-added-a-test-case.patch
  - Dropped debian/patches/upstream/0003-Allow-unprivileged-RADOS-users-to-access-rbd-volumes.patch
  - Dropped debian/patches/upstream/0002-Stop-libvirt-test-from-deleting-instances-dir.patch
  - Dropped debian/patches/upstream/0001-fix-bug-where-nova-ignores-glance-host-in-imageref.patch 
  - Dropped debian/patches/0001-fix-useexisting-deprecation-warnings.patch
* debian/control: Add python-keystone as a dependency. (LP: #907197)
* debian/patches/kombu_tests_timeout.patch: Refreshed.
* debian/nova.conf, debian/nova-common.postinst: Convert to new ini
  file configuration
* debian/patches/nova-manage_flagfile_location.patch: Refreshed

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/rpc/test_proxy.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright 2012, Red Hat, Inc.
 
4
#
 
5
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
 
6
#    not use this file except in compliance with the License. You may obtain
 
7
#    a copy of the License at
 
8
#
 
9
#         http://www.apache.org/licenses/LICENSE-2.0
 
10
#
 
11
#    Unless required by applicable law or agreed to in writing, software
 
12
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
13
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
14
#    License for the specific language governing permissions and limitations
 
15
#    under the License.
 
16
 
 
17
"""
 
18
Unit Tests for rpc.proxy
 
19
"""
 
20
 
 
21
import copy
 
22
 
 
23
from nova import context
 
24
from nova import rpc
 
25
from nova.rpc import proxy
 
26
from nova import test
 
27
 
 
28
 
 
29
class RpcProxyTestCase(test.TestCase):
 
30
 
 
31
    def setUp(self):
 
32
        super(RpcProxyTestCase, self).setUp()
 
33
 
 
34
    def tearDown(self):
 
35
        super(RpcProxyTestCase, self).tearDown()
 
36
 
 
37
    def _test_rpc_method(self, rpc_method, has_timeout=False, has_retval=False,
 
38
            server_params=None, supports_topic_override=True):
 
39
        topic = 'fake_topic'
 
40
        timeout = 123
 
41
        rpc_proxy = proxy.RpcProxy(topic, '1.0')
 
42
        ctxt = context.RequestContext('fake_user', 'fake_project')
 
43
        msg = {'method': 'fake_method', 'args': {'x': 'y'}}
 
44
        expected_msg = {'method': 'fake_method', 'args': {'x': 'y'},
 
45
                'version': '1.0'}
 
46
 
 
47
        expected_retval = 'hi' if has_retval else None
 
48
 
 
49
        self.fake_args = None
 
50
        self.fake_kwargs = None
 
51
 
 
52
        def _fake_rpc_method(*args, **kwargs):
 
53
            self.fake_args = args
 
54
            self.fake_kwargs = kwargs
 
55
            if has_retval:
 
56
                return expected_retval
 
57
 
 
58
        self.stubs.Set(rpc, rpc_method, _fake_rpc_method)
 
59
 
 
60
        args = [ctxt, msg]
 
61
        if server_params:
 
62
            args.insert(1, server_params)
 
63
 
 
64
        # Base method usage
 
65
        retval = getattr(rpc_proxy, rpc_method)(*args)
 
66
        self.assertEqual(retval, expected_retval)
 
67
        expected_args = [ctxt, topic, expected_msg]
 
68
        if server_params:
 
69
            expected_args.insert(1, server_params)
 
70
        for arg, expected_arg in zip(self.fake_args, expected_args):
 
71
            self.assertEqual(arg, expected_arg)
 
72
 
 
73
        # overriding the version
 
74
        retval = getattr(rpc_proxy, rpc_method)(*args, version='1.1')
 
75
        self.assertEqual(retval, expected_retval)
 
76
        new_msg = copy.deepcopy(expected_msg)
 
77
        new_msg['version'] = '1.1'
 
78
        expected_args = [ctxt, topic, new_msg]
 
79
        if server_params:
 
80
            expected_args.insert(1, server_params)
 
81
        for arg, expected_arg in zip(self.fake_args, expected_args):
 
82
            self.assertEqual(arg, expected_arg)
 
83
 
 
84
        if has_timeout:
 
85
            # set a timeout
 
86
            retval = getattr(rpc_proxy, rpc_method)(ctxt, msg, timeout=timeout)
 
87
            self.assertEqual(retval, expected_retval)
 
88
            expected_args = [ctxt, topic, expected_msg, timeout]
 
89
            for arg, expected_arg in zip(self.fake_args, expected_args):
 
90
                self.assertEqual(arg, expected_arg)
 
91
 
 
92
        if supports_topic_override:
 
93
            # set a topic
 
94
            new_topic = 'foo.bar'
 
95
            retval = getattr(rpc_proxy, rpc_method)(*args, topic=new_topic)
 
96
            self.assertEqual(retval, expected_retval)
 
97
            expected_args = [ctxt, new_topic, expected_msg]
 
98
            if server_params:
 
99
                expected_args.insert(1, server_params)
 
100
            for arg, expected_arg in zip(self.fake_args, expected_args):
 
101
                self.assertEqual(arg, expected_arg)
 
102
 
 
103
    def test_call(self):
 
104
        self._test_rpc_method('call', has_timeout=True, has_retval=True)
 
105
 
 
106
    def test_multicall(self):
 
107
        self._test_rpc_method('multicall', has_timeout=True, has_retval=True)
 
108
 
 
109
    def test_cast(self):
 
110
        self._test_rpc_method('cast')
 
111
 
 
112
    def test_fanout_cast(self):
 
113
        self._test_rpc_method('fanout_cast', supports_topic_override=False)
 
114
 
 
115
    def test_cast_to_server(self):
 
116
        self._test_rpc_method('cast_to_server', server_params={'blah': 1})
 
117
 
 
118
    def test_fanout_cast_to_server(self):
 
119
        self._test_rpc_method('fanout_cast_to_server',
 
120
                server_params={'blah': 1}, supports_topic_override=False)
 
121
 
 
122
    def test_make_msg(self):
 
123
        self.assertEqual(proxy.RpcProxy.make_msg('test_method', a=1, b=2),
 
124
                         {'method': 'test_method', 'args': {'a': 1, 'b': 2}})