1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2012 OpenStack LLC.
6
# Licensed under the Apache License, Version 2.0 (the "License"); you may
7
# not use this file except in compliance with the License. You may obtain
8
# a copy of the License at
10
# http://www.apache.org/licenses/LICENSE-2.0
12
# Unless required by applicable law or agreed to in writing, software
13
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15
# License for the specific language governing permissions and limitations
19
from nova import flags
20
from nova.network.quantum import melange_connection
21
from nova.network.quantum import melange_ipam_lib
27
class MelangeIpamLibTestCase(test.TestCase):
29
super(MelangeIpamLibTestCase, self).setUp()
30
self.m_conn = self.mox.CreateMock(melange_connection.MelangeConnection)
31
self.ipam = melange_ipam_lib.QuantumMelangeIPAMLib()
32
self.ipam.m_conn = self.m_conn
34
def _block_list(self, id='id', cidr='cidr', network_id='network_id'):
35
return {'ip_blocks': [{'id': id,
37
'network_id': network_id}]}
39
def test_allocate_fixed_ips_extracts_address(self):
40
self.m_conn.allocate_ip('quantum_net_id', 'network_tenant_id',
41
'vif_ref_uuid', 'project_id',
42
'vif_ref_address').AndReturn(
43
[{'address': 'ip_address'}])
47
ips = self.ipam.allocate_fixed_ips('context',
51
{'uuid': 'vif_ref_uuid',
52
'address': 'vif_ref_address'})
53
self.assertEqual(ips[0], 'ip_address')
55
def test_delete_subnets_by_net_id_deletes_block(self):
56
context = self.mox.CreateMockAnything()
57
context.elevated().AndReturn('elevated')
59
self.m_conn.get_blocks('project_id').AndReturn(
60
self._block_list(id='block_id'))
61
self.m_conn.delete_block('block_id', 'project_id')
63
self.mox.StubOutWithMock(db, 'network_get_by_uuid')
64
db.network_get_by_uuid('elevated', 'network_id').AndReturn(
67
self.mox.StubOutWithMock(db, 'network_delete_safe')
68
db.network_delete_safe(context, 'network_id')
71
self.ipam.delete_subnets_by_net_id(context, 'network_id', 'project_id')
73
def test_get_networks_by_tenant_gets_all_networks(self):
74
block_list = self._block_list(network_id='net_1')
75
block_list['ip_blocks'] += self._block_list(
76
network_id='net_2')['ip_blocks']
78
self.m_conn.get_blocks('tenant_id').AndReturn(block_list)
80
self.mox.StubOutWithMock(db, 'network_get_by_uuid')
81
db.network_get_by_uuid('admin_context', 'net_1').AndReturn('network1')
82
db.network_get_by_uuid('admin_context', 'net_2').AndReturn('network2')
85
values = self.ipam.get_networks_by_tenant('admin_context', 'tenant_id')
86
self.assertEquals(values, ['network1', 'network2'])
88
def test_get_global_networks(self):
89
FLAGS.quantum_default_tenant_id = 'quantum_default_tenant_id'
90
self.mox.StubOutWithMock(self.ipam, 'get_networks_by_tenant')
91
self.ipam.get_networks_by_tenant('admin_context',
92
'quantum_default_tenant_id')
96
self.ipam.get_global_networks('admin_context')
98
def test_get_project_networks(self):
99
context = self.mox.CreateMockAnything()
100
context.elevated().AndReturn('elevated')
102
networks = [{'project_id': 1}, {'project_id': None}]
104
self.mox.StubOutWithMock(db, 'network_get_all')
105
db.network_get_all('elevated').AndReturn(networks)
108
values = self.ipam.get_project_networks(context)
109
self.assertEquals(values, [networks[0]])
111
def test_get_project_and_global_net_ids__by_priority(self):
112
context = self.mox.CreateMockAnything()
113
context.elevated().AndReturn('elevated')
115
FLAGS.quantum_default_tenant_id = 'default_tenant_id'
117
net1 = {'uuid': 'net1_uuid', 'priority': 'net1_priority'}
118
net2 = {'uuid': 'net2_uuid', 'priority': 'net2_priority'}
120
self.mox.StubOutWithMock(self.ipam, 'get_networks_by_tenant')
121
self.ipam.get_networks_by_tenant('elevated',
122
'project_id').AndReturn([net1])
123
self.ipam.get_networks_by_tenant('elevated',
124
'default_tenant_id').AndReturn([net2])
126
self.ipam.get_project_and_global_net_ids(context, 'project_id')
128
def test_get_tenant_id_by_net_id_returns_id(self):
129
FLAGS.quantum_default_tenant_id = 'qdti'
131
self.m_conn.get_allocated_ips('net_id', 'vif_id',
132
'qdti').AndReturn({})
134
value = self.ipam.get_tenant_id_by_net_id('context', 'net_id',
135
'vif_id', 'project_id')
136
self.assertEqual(value, 'qdti')
138
def test_get_tenant_id_by_net_id_returns_none_if_none_found(self):
139
FLAGS.quantum_default_tenant_id = 'qdti'
141
self.m_conn.get_allocated_ips('net_id', 'vif_id',
142
'qdti').AndRaise(KeyError())
143
self.m_conn.get_allocated_ips('net_id', 'vif_id',
144
'project_id').AndRaise(KeyError())
145
self.m_conn.get_allocated_ips('net_id', 'vif_id',
146
None).AndRaise(KeyError())
148
value = self.ipam.get_tenant_id_by_net_id('context', 'net_id',
149
'vif_id', 'project_id')
150
self.assertEqual(value, None)
152
def test_get_subnets_by_net_id(self):
153
ips = [{'ip_block': {'network_id': 'network_id',
156
'gateway': 'gateway',
157
'broadcast': 'broadcast',
158
'netmask': 'netmask',
163
self.m_conn.get_allocated_ips('net_id', 'vif_id',
164
'tenant_id').AndReturn(ips)
167
value = self.ipam.get_subnets_by_net_id('context', 'tenant_id',
169
self.assertEquals(value[0]['cidr'], 'cidr')
171
def test_get_routes_by_ip_block(self):
172
self.m_conn.get_routes('block_id', 'project_id')
174
self.ipam.get_routes_by_ip_block('context', 'block_id', 'project_id')
176
def test_get_v4_ips_by_interface(self):
177
self.mox.StubOutWithMock(self.ipam, '_get_ips_by_interface')
178
self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
181
self.ipam.get_v4_ips_by_interface('context', 'net_id', 'vif_id',
184
def test_get_v6_ips_by_interface(self):
185
self.mox.StubOutWithMock(self.ipam, '_get_ips_by_interface')
186
self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
189
self.ipam.get_v6_ips_by_interface('context', 'net_id', 'vif_id',
192
def test_get_ips_by_interface(self):
193
ips = [{'address': '10.10.10.10'}, {'address': '2001::CAFE'}]
194
self.m_conn.get_allocated_ips('net_id', 'vif_id',
195
'tenant_id').AndReturn(ips)
196
self.m_conn.get_allocated_ips('net_id', 'vif_id',
197
'tenant_id').AndReturn(ips)
199
values = self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
201
self.assertEquals(values, ["10.10.10.10"])
202
values = self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
204
self.assertEquals(values, ["2001::CAFE"])
206
def test_get_instance_ids_by_ip_address(self):
207
ips = [{'used_by_device': 'some_vif_uuid'}]
208
self.m_conn.get_allocated_ips_by_address('ip').AndReturn(ips)
211
uuid = self.ipam.get_instance_ids_by_ip_address('context', 'ip')
212
self.assertEqual(uuid, ['some_vif_uuid'])
214
def test_verify_subnet_exists(self):
215
blocks = {'ip_blocks': [{'network_id': 'quantum_net_id'}]}
216
self.m_conn.get_blocks('tenant_id').AndReturn(blocks)
218
value = self.ipam.verify_subnet_exists('context', 'tenant_id',
220
self.assertEquals(value, True)
222
def test_deallocate_ips_by_vif(self):
223
self.m_conn.deallocate_ips('net_id', 'uuid', 'tenant_id')
225
self.ipam.deallocate_ips_by_vif('context', 'tenant_id', 'net_id',
228
def test_get_allocated_ips(self):
229
ips = [{'address': 'ip_address', 'interface_id': 'interface_id'}]
230
self.m_conn.get_allocated_ips_for_network('subnet_id',
231
'project_id').AndReturn(ips)
233
self.ipam.get_allocated_ips('context', 'subnet_id', 'project_id')
235
def test_create_vif(self):
236
self.m_conn.create_vif('vif_id', 'instance_id', 'project_id')
238
self.ipam.create_vif('vif_id', 'instance_id', 'project_id')
240
def test_get_floating_ips_by_fixed_address(self):
241
value = self.ipam.get_floating_ips_by_fixed_address('context',
243
self.assertEquals(value, [])
added
removed
nova/tests/test_melange_ipam_lib.py
