← Back to branch summary

~ubuntu-branches/ubuntu/quantal/nova/quantal-proposed

« back to all changes in this revision

Viewing changes to nova/tests/api/openstack/compute/contrib/test_floating_ips.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2012-01-20 11:54:15 UTC
  • mto: This revision was merged to the branch mainline in revision 62.
  • Revision ID: package-import@ubuntu.com-20120120115415-h2ujma9o536o1ut6
Tags: upstream-2012.1~e3~20120120.12170
Import upstream version 2012.1~e3~20120120.12170

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/api/openstack/compute/contrib/test_floating_ips.py
 
1
# Copyright (c) 2011 X.commerce, a business unit of eBay Inc.
 
2
# Copyright 2011 Eldar Nugaev
 
3
# All Rights Reserved.
 
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
from lxml import etree
 
18
import webob
 
19
 
 
20
from nova.api.openstack.compute.contrib import floating_ips
 
21
from nova import context
 
22
from nova import db
 
23
from nova import network
 
24
from nova import compute
 
25
from nova import rpc
 
26
from nova import test
 
27
from nova.tests.api.openstack import fakes
 
28
from nova import utils
 
29
 
 
30
FAKE_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
 
31
 
 
32
 
 
33
def network_api_get_fixed_ip(self, context, id):
 
34
    if id is None:
 
35
        return None
 
36
    return {'address': '10.0.0.1', 'id': id, 'instance_id': 1}
 
37
 
 
38
 
 
39
def network_api_get_floating_ip(self, context, id):
 
40
    return {'id': 1, 'address': '10.10.10.10', 'pool': 'nova',
 
41
            'fixed_ip_id': None}
 
42
 
 
43
 
 
44
def network_api_get_floating_ip_by_address(self, context, address):
 
45
    return {'id': 1, 'address': '10.10.10.10', 'pool': 'nova',
 
46
            'fixed_ip_id': 10}
 
47
 
 
48
 
 
49
def network_api_get_floating_ips_by_project(self, context):
 
50
    return [{'id': 1,
 
51
             'address': '10.10.10.10',
 
52
             'pool': 'nova',
 
53
             'fixed_ip_id': 20},
 
54
            {'id': 2,
 
55
             'pool': 'nova', 'interface': 'eth0',
 
56
             'address': '10.10.10.11',
 
57
            'fixed_ip_id': None}]
 
58
 
 
59
 
 
60
def compute_api_get(self, context, instance_id):
 
61
    return dict(uuid=FAKE_UUID)
 
62
 
 
63
 
 
64
def network_api_allocate(self, context):
 
65
    return '10.10.10.10'
 
66
 
 
67
 
 
68
def network_api_release(self, context, address):
 
69
    pass
 
70
 
 
71
 
 
72
def compute_api_associate(self, context, instance_id, address):
 
73
    pass
 
74
 
 
75
 
 
76
def network_api_associate(self, context, floating_address, fixed_address):
 
77
    pass
 
78
 
 
79
 
 
80
def network_api_disassociate(self, context, floating_address):
 
81
    pass
 
82
 
 
83
 
 
84
def network_get_instance_nw_info(self, context, instance):
 
85
    info = {
 
86
        'label': 'fake',
 
87
        'gateway': 'fake',
 
88
        'dhcp_server': 'fake',
 
89
        'broadcast': 'fake',
 
90
        'mac': 'fake',
 
91
        'vif_uuid': 'fake',
 
92
        'rxtx_cap': 'fake',
 
93
        'dns': [],
 
94
        'ips': [{'ip': '10.0.0.1'}],
 
95
        'should_create_bridge': False,
 
96
        'should_create_vlan': False}
 
97
 
 
98
    return [['ignore', info]]
 
99
 
 
100
 
 
101
def fake_instance_get(context, instance_id):
 
102
        return {
 
103
        "id": 1,
 
104
        "uuid": utils.gen_uuid(),
 
105
        "name": 'fake',
 
106
        "user_id": 'fakeuser',
 
107
        "project_id": '123'}
 
108
 
 
109
 
 
110
class FloatingIpTest(test.TestCase):
 
111
    floating_ip = "10.10.10.10"
 
112
 
 
113
    def _create_floating_ip(self):
 
114
        """Create a floating ip object."""
 
115
        host = "fake_host"
 
116
        return db.floating_ip_create(self.context,
 
117
                                     {'address': self.floating_ip,
 
118
                                      'pool': 'nova',
 
119
                                      'host': host})
 
120
 
 
121
    def _delete_floating_ip(self):
 
122
        db.floating_ip_destroy(self.context, self.floating_ip)
 
123
 
 
124
    def setUp(self):
 
125
        super(FloatingIpTest, self).setUp()
 
126
        self.stubs.Set(network.api.API, "get_fixed_ip",
 
127
                       network_api_get_fixed_ip)
 
128
        self.stubs.Set(compute.api.API, "get",
 
129
                       compute_api_get)
 
130
        self.stubs.Set(network.api.API, "get_floating_ip",
 
131
                       network_api_get_floating_ip)
 
132
        self.stubs.Set(network.api.API, "get_floating_ip_by_address",
 
133
                       network_api_get_floating_ip_by_address)
 
134
        self.stubs.Set(network.api.API, "get_floating_ips_by_project",
 
135
                       network_api_get_floating_ips_by_project)
 
136
        self.stubs.Set(network.api.API, "release_floating_ip",
 
137
                       network_api_release)
 
138
        self.stubs.Set(network.api.API, "disassociate_floating_ip",
 
139
                       network_api_disassociate)
 
140
        self.stubs.Set(network.api.API, "get_instance_nw_info",
 
141
                       network_get_instance_nw_info)
 
142
        self.stubs.Set(db, 'instance_get',
 
143
                       fake_instance_get)
 
144
 
 
145
        self.context = context.get_admin_context()
 
146
        self._create_floating_ip()
 
147
 
 
148
        self.controller = floating_ips.FloatingIPController()
 
149
        self.manager = floating_ips.FloatingIPActionController()
 
150
 
 
151
    def tearDown(self):
 
152
        self._delete_floating_ip()
 
153
        super(FloatingIpTest, self).tearDown()
 
154
 
 
155
    def test_translate_floating_ip_view(self):
 
156
        floating_ip_address = self._create_floating_ip()
 
157
        floating_ip = db.floating_ip_get_by_address(self.context,
 
158
                                                    floating_ip_address)
 
159
        floating_ip['fixed_ip'] = None
 
160
        floating_ip['instance'] = None
 
161
        view = floating_ips._translate_floating_ip_view(floating_ip)
 
162
        self.assertTrue('floating_ip' in view)
 
163
        self.assertTrue(view['floating_ip']['id'])
 
164
        self.assertEqual(view['floating_ip']['ip'], self.floating_ip)
 
165
        self.assertEqual(view['floating_ip']['fixed_ip'], None)
 
166
        self.assertEqual(view['floating_ip']['instance_id'], None)
 
167
 
 
168
    def test_translate_floating_ip_view_dict(self):
 
169
        floating_ip = {'id': 0, 'address': '10.0.0.10', 'pool': 'nova',
 
170
                       'fixed_ip': None}
 
171
        view = floating_ips._translate_floating_ip_view(floating_ip)
 
172
        self.assertTrue('floating_ip' in view)
 
173
 
 
174
    def test_floating_ips_list(self):
 
175
        req = fakes.HTTPRequest.blank('/v2/fake/os-floating-ips')
 
176
        res_dict = self.controller.index(req)
 
177
 
 
178
        response = {'floating_ips': [{'instance_id': FAKE_UUID,
 
179
                                      'ip': '10.10.10.10',
 
180
                                      'pool': 'nova',
 
181
                                      'fixed_ip': '10.0.0.1',
 
182
                                      'id': 1},
 
183
                                     {'instance_id': None,
 
184
                                      'ip': '10.10.10.11',
 
185
                                      'pool': 'nova',
 
186
                                      'fixed_ip': None,
 
187
                                      'id': 2}]}
 
188
        self.assertEqual(res_dict, response)
 
189
 
 
190
    def test_floating_ip_show(self):
 
191
        req = fakes.HTTPRequest.blank('/v2/fake/os-floating-ips/1')
 
192
        res_dict = self.controller.show(req, 1)
 
193
 
 
194
        self.assertEqual(res_dict['floating_ip']['id'], 1)
 
195
        self.assertEqual(res_dict['floating_ip']['ip'], '10.10.10.10')
 
196
        self.assertEqual(res_dict['floating_ip']['instance_id'], None)
 
197
 
 
198
    def test_show_associated_floating_ip(self):
 
199
        def get_floating_ip(self, context, id):
 
200
            return {'id': 1, 'address': '10.10.10.10', 'pool': 'nova',
 
201
                    'fixed_ip_id': 11}
 
202
 
 
203
        def get_fixed_ip(self, context, id):
 
204
            return {'address': '10.0.0.1', 'instance_id': 1}
 
205
 
 
206
        self.stubs.Set(network.api.API, "get_floating_ip", get_floating_ip)
 
207
        self.stubs.Set(network.api.API, "get_fixed_ip", get_fixed_ip)
 
208
 
 
209
        req = fakes.HTTPRequest.blank('/v2/fake/os-floating-ips/1')
 
210
        res_dict = self.controller.show(req, 1)
 
211
 
 
212
        self.assertEqual(res_dict['floating_ip']['id'], 1)
 
213
        self.assertEqual(res_dict['floating_ip']['ip'], '10.10.10.10')
 
214
        self.assertEqual(res_dict['floating_ip']['instance_id'], FAKE_UUID)
 
215
 
 
216
# test floating ip allocate/release(deallocate)
 
217
    def test_floating_ip_allocate_no_free_ips(self):
 
218
        def fake_call(*args, **kwargs):
 
219
            raise(rpc.RemoteError('NoMoreFloatingIps', '', ''))
 
220
 
 
221
        self.stubs.Set(rpc, "call", fake_call)
 
222
 
 
223
        req = fakes.HTTPRequest.blank('/v2/fake/os-floating-ips')
 
224
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
 
225
                          req)
 
226
 
 
227
    def test_floating_ip_allocate(self):
 
228
        def fake1(*args, **kwargs):
 
229
            pass
 
230
 
 
231
        def fake2(*args, **kwargs):
 
232
            return {'id': 1, 'address': '10.10.10.10', 'pool': 'nova'}
 
233
 
 
234
        self.stubs.Set(network.api.API, "allocate_floating_ip",
 
235
                       fake1)
 
236
        self.stubs.Set(network.api.API, "get_floating_ip_by_address",
 
237
                       fake2)
 
238
 
 
239
        req = fakes.HTTPRequest.blank('/v2/fake/os-floating-ips')
 
240
        res_dict = self.controller.create(req)
 
241
 
 
242
        ip = res_dict['floating_ip']
 
243
 
 
244
        expected = {
 
245
            "id": 1,
 
246
            "instance_id": None,
 
247
            "ip": "10.10.10.10",
 
248
            "fixed_ip": None,
 
249
            "pool": 'nova'}
 
250
        self.assertEqual(ip, expected)
 
251
 
 
252
    def test_floating_ip_release(self):
 
253
        req = fakes.HTTPRequest.blank('/v2/fake/os-floating-ips/1')
 
254
        self.controller.delete(req, 1)
 
255
 
 
256
# test floating ip add/remove -> associate/disassociate
 
257
 
 
258
    def test_floating_ip_associate(self):
 
259
        body = dict(addFloatingIp=dict(address=self.floating_ip))
 
260
 
 
261
        req = fakes.HTTPRequest.blank('/v2/fake/servers/test_inst/action')
 
262
        self.manager._add_floating_ip(req, 'test_inst', body)
 
263
 
 
264
    def test_floating_ip_disassociate(self):
 
265
        body = dict(removeFloatingIp=dict(address='10.10.10.10'))
 
266
 
 
267
        req = fakes.HTTPRequest.blank('/v2/fake/servers/test_inst/action')
 
268
        self.manager._remove_floating_ip(req, 'test_inst', body)
 
269
 
 
270
# these are a few bad param tests
 
271
 
 
272
    def test_bad_address_param_in_remove_floating_ip(self):
 
273
        body = dict(removeFloatingIp=dict(badparam='11.0.0.1'))
 
274
 
 
275
        req = fakes.HTTPRequest.blank('/v2/fake/servers/test_inst/action')
 
276
        self.assertRaises(webob.exc.HTTPBadRequest,
 
277
                          self.manager._add_floating_ip, req, 'test_inst',
 
278
                          body)
 
279
 
 
280
    def test_missing_dict_param_in_remove_floating_ip(self):
 
281
        body = dict(removeFloatingIp='11.0.0.1')
 
282
 
 
283
        req = fakes.HTTPRequest.blank('/v2/fake/servers/test_inst/action')
 
284
        self.assertRaises(webob.exc.HTTPBadRequest,
 
285
                          self.manager._remove_floating_ip, req, 'test_inst',
 
286
                          body)
 
287
 
 
288
    def test_missing_dict_param_in_add_floating_ip(self):
 
289
        body = dict(addFloatingIp='11.0.0.1')
 
290
 
 
291
        req = fakes.HTTPRequest.blank('/v2/fake/servers/test_inst/action')
 
292
        self.assertRaises(webob.exc.HTTPBadRequest,
 
293
                          self.manager._add_floating_ip, req, 'test_inst',
 
294
                          body)
 
295
 
 
296
 
 
297
class FloatingIpSerializerTest(test.TestCase):
 
298
    def test_default_serializer(self):
 
299
        serializer = floating_ips.FloatingIPTemplate()
 
300
        text = serializer.serialize(dict(
 
301
                floating_ip=dict(
 
302
                    instance_id=1,
 
303
                    ip='10.10.10.10',
 
304
                    fixed_ip='10.0.0.1',
 
305
                    id=1)))
 
306
 
 
307
        tree = etree.fromstring(text)
 
308
 
 
309
        self.assertEqual('floating_ip', tree.tag)
 
310
        self.assertEqual('1', tree.get('instance_id'))
 
311
        self.assertEqual('10.10.10.10', tree.get('ip'))
 
312
        self.assertEqual('10.0.0.1', tree.get('fixed_ip'))
 
313
        self.assertEqual('1', tree.get('id'))
 
314
 
 
315
    def test_index_serializer(self):
 
316
        serializer = floating_ips.FloatingIPsTemplate()
 
317
        text = serializer.serialize(dict(
 
318
                floating_ips=[
 
319
                    dict(instance_id=1,
 
320
                         ip='10.10.10.10',
 
321
                         fixed_ip='10.0.0.1',
 
322
                         id=1),
 
323
                    dict(instance_id=None,
 
324
                         ip='10.10.10.11',
 
325
                         fixed_ip=None,
 
326
                         id=2)]))
 
327
 
 
328
        tree = etree.fromstring(text)
 
329
 
 
330
        self.assertEqual('floating_ips', tree.tag)
 
331
        self.assertEqual(2, len(tree))
 
332
        self.assertEqual('floating_ip', tree[0].tag)
 
333
        self.assertEqual('floating_ip', tree[1].tag)
 
334
        self.assertEqual('1', tree[0].get('instance_id'))
 
335
        self.assertEqual('None', tree[1].get('instance_id'))
 
336
        self.assertEqual('10.10.10.10', tree[0].get('ip'))
 
337
        self.assertEqual('10.10.10.11', tree[1].get('ip'))
 
338
        self.assertEqual('10.0.0.1', tree[0].get('fixed_ip'))
 
339
        self.assertEqual('None', tree[1].get('fixed_ip'))
 
340
        self.assertEqual('1', tree[0].get('id'))
 
341
        self.assertEqual('2', tree[1].get('id'))