← Back to branch summary

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

« back to all changes in this revision

Viewing changes to nova/tests/test_melange_ipam_lib.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Dave Walker (Daviey), Chuck Short
  • Date: 2012-02-17 10:59:59 UTC
  • mfrom: (1.1.45)
  • Revision ID: package-import@ubuntu.com-20120217105959-ut218n638vv7cre0
Tags: 2012.1~e4~20120217.12709-0ubuntu1
https://launchpad.net/bugs/929127
https://launchpad.net/bugs/923681
https://launchpad.net/bugs/932787
[ Dave Walker (Daviey) ]
* New upstream snapshot
* debian/patches/temp_fix_linux_net.patch:
  - Dropped, applied upstream. LP: #929127
* debian/patches/libvirt-use-console-pipe.patch:
  - Rebased against latest trunk

[ Chuck Short ]
* debian/nova.conf: Re-enable default iscsi_helper.
* debian/nova.conf: More fixups.
* debian/control: Dont depend and conflicts on nova-compute-
  hypervisor. (LP: #923681)
* debian/patches/libvirt-us-console-pipe.patch: Refreshed.
* Temporarily disable console patch. (LP: #932787)
* New usptream version.

Show diffs side-by-side

added added

removed removed

Lines of Context:
nova/tests/test_melange_ipam_lib.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright 2012 OpenStack LLC.
 
4
# All Rights Reserved.
 
5
#
 
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
 
9
#
 
10
#         http://www.apache.org/licenses/LICENSE-2.0
 
11
#
 
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
 
16
#    under the License.
 
17
 
 
18
from nova import db
 
19
from nova import exception
 
20
from nova import flags
 
21
from nova import test
 
22
from nova.network.quantum import melange_connection
 
23
from nova.network.quantum import melange_ipam_lib
 
24
 
 
25
FLAGS = flags.FLAGS
 
26
 
 
27
 
 
28
class MelangeIpamLibTestCase(test.TestCase):
 
29
    def setUp(self):
 
30
        super(MelangeIpamLibTestCase, self).setUp()
 
31
        self.m_conn = self.mox.CreateMock(melange_connection.MelangeConnection)
 
32
        self.ipam = melange_ipam_lib.QuantumMelangeIPAMLib()
 
33
        self.ipam.m_conn = self.m_conn
 
34
 
 
35
    def _block_list(self, id='id', cidr='cidr', network_id='network_id'):
 
36
        return {'ip_blocks': [{'id': id,
 
37
                                'cidr': cidr,
 
38
                               'network_id': network_id}]}
 
39
 
 
40
    def test_allocate_fixed_ips_extracts_address(self):
 
41
        self.m_conn.allocate_ip('quantum_net_id', 'network_tenant_id',
 
42
                                'vif_ref_uuid', 'project_id',
 
43
                                'vif_ref_address').AndReturn(
 
44
                                                [{'address': 'ip_address'}])
 
45
 
 
46
        self.mox.ReplayAll()
 
47
 
 
48
        ips = self.ipam.allocate_fixed_ips('context',
 
49
                                           'project_id',
 
50
                                           'quantum_net_id',
 
51
                                           'network_tenant_id',
 
52
                                           {'uuid': 'vif_ref_uuid',
 
53
                                            'address': 'vif_ref_address'})
 
54
        self.assertEqual(ips[0], 'ip_address')
 
55
 
 
56
    def test_get_network_id_by_cidr_finds_block(self):
 
57
        self.m_conn.get_blocks('tenant_id').AndReturn(self._block_list())
 
58
 
 
59
        self.mox.ReplayAll()
 
60
 
 
61
        net_id = self.ipam.get_network_id_by_cidr('context', 'cidr',
 
62
                                                  'tenant_id')
 
63
        self.assertEqual(net_id, 'network_id')
 
64
 
 
65
    def test_get_network_id_by_cidr_raises_on_not_found(self):
 
66
        self.m_conn.get_blocks('tenant_id').AndReturn({'ip_blocks': []})
 
67
        self.mox.ReplayAll()
 
68
        self.assertRaises(exception.NotFound,
 
69
                          self.ipam.get_network_id_by_cidr,
 
70
                          'context', 'cidr', 'tenant_id')
 
71
 
 
72
    def test_delete_subnets_by_net_id_deletes_block(self):
 
73
        context = self.mox.CreateMockAnything()
 
74
        context.elevated().AndReturn('elevated')
 
75
 
 
76
        self.m_conn.get_blocks('project_id').AndReturn(
 
77
                                        self._block_list(id='block_id'))
 
78
        self.m_conn.delete_block('block_id', 'project_id')
 
79
 
 
80
        self.mox.StubOutWithMock(db, 'network_get_by_uuid')
 
81
        db.network_get_by_uuid('elevated', 'network_id').AndReturn(
 
82
                                                         {'id': 'network_id'})
 
83
 
 
84
        self.mox.StubOutWithMock(db, 'network_delete_safe')
 
85
        db.network_delete_safe(context, 'network_id')
 
86
 
 
87
        self.mox.ReplayAll()
 
88
        self.ipam.delete_subnets_by_net_id(context, 'network_id', 'project_id')
 
89
 
 
90
    def test_get_networks_by_tenant_gets_all_networks(self):
 
91
        block_list = self._block_list(network_id='net_1')
 
92
        block_list['ip_blocks'] += self._block_list(
 
93
                                        network_id='net_2')['ip_blocks']
 
94
 
 
95
        self.m_conn.get_blocks('tenant_id').AndReturn(block_list)
 
96
 
 
97
        self.mox.StubOutWithMock(db, 'network_get_by_uuid')
 
98
        db.network_get_by_uuid('admin_context', 'net_1').AndReturn('network1')
 
99
        db.network_get_by_uuid('admin_context', 'net_2').AndReturn('network2')
 
100
 
 
101
        self.mox.ReplayAll()
 
102
        values = self.ipam.get_networks_by_tenant('admin_context', 'tenant_id')
 
103
        self.assertEquals(values, ['network1', 'network2'])
 
104
 
 
105
    def test_get_global_networks(self):
 
106
        FLAGS.quantum_default_tenant_id = 'quantum_default_tenant_id'
 
107
        self.mox.StubOutWithMock(self.ipam, 'get_networks_by_tenant')
 
108
        self.ipam.get_networks_by_tenant('admin_context',
 
109
                                    'quantum_default_tenant_id')
 
110
 
 
111
        self.mox.ReplayAll()
 
112
 
 
113
        self.ipam.get_global_networks('admin_context')
 
114
 
 
115
    def test_get_project_networks(self):
 
116
        context = self.mox.CreateMockAnything()
 
117
        context.elevated().AndReturn('elevated')
 
118
 
 
119
        networks = [{'project_id': 1}, {'project_id': None}]
 
120
 
 
121
        self.mox.StubOutWithMock(db, 'network_get_all')
 
122
        db.network_get_all('elevated').AndReturn(networks)
 
123
 
 
124
        self.mox.ReplayAll()
 
125
        values = self.ipam.get_project_networks(context)
 
126
        self.assertEquals(values, [networks[0]])
 
127
 
 
128
    def test_get_project_and_global_net_ids__by_priority(self):
 
129
        context = self.mox.CreateMockAnything()
 
130
        context.elevated().AndReturn('elevated')
 
131
 
 
132
        FLAGS.quantum_default_tenant_id = 'default_tenant_id'
 
133
 
 
134
        net1 = {'uuid': 'net1_uuid', 'priority': 'net1_priority'}
 
135
        net2 = {'uuid': 'net2_uuid', 'priority': 'net2_priority'}
 
136
 
 
137
        self.mox.StubOutWithMock(self.ipam, 'get_networks_by_tenant')
 
138
        self.ipam.get_networks_by_tenant('elevated',
 
139
                                         'project_id').AndReturn([net1])
 
140
        self.ipam.get_networks_by_tenant('elevated',
 
141
                                         'default_tenant_id').AndReturn([net2])
 
142
        self.mox.ReplayAll()
 
143
        self.ipam.get_project_and_global_net_ids(context, 'project_id')
 
144
 
 
145
    def test_get_tenant_id_by_net_id(self):
 
146
        FLAGS.quantum_default_tenant_id = 'qdti'
 
147
 
 
148
        self.m_conn.get_allocated_ips('net_id', 'vif_id',
 
149
                                      None).AndReturn('tenant_id')
 
150
        self.mox.ReplayAll()
 
151
        self.ipam.get_tenant_id_by_net_id('context', 'net_id', 'vif_id',
 
152
                                          'project_id')
 
153
 
 
154
    def test_get_subnets_by_net_id(self):
 
155
        ips = [{'ip_block': {'network_id': 'network_id',
 
156
                            'id': 'id',
 
157
                            'cidr': 'cidr',
 
158
                            'gateway': 'gateway',
 
159
                            'broadcast': 'broadcast',
 
160
                            'netmask': 'netmask',
 
161
                            'dns1': 'dns1',
 
162
                            'dns2': 'dns2'},
 
163
                'version': 4}]
 
164
 
 
165
        self.m_conn.get_allocated_ips('net_id', 'vif_id',
 
166
                                      'tenant_id').AndReturn(ips)
 
167
 
 
168
        self.mox.ReplayAll()
 
169
        value = self.ipam.get_subnets_by_net_id('context', 'tenant_id',
 
170
                                                'net_id', 'vif_id')
 
171
        self.assertEquals(value[0]['cidr'], 'cidr')
 
172
 
 
173
    def test_get_routes_by_ip_block(self):
 
174
        self.m_conn.get_routes('block_id', 'project_id')
 
175
        self.mox.ReplayAll()
 
176
        self.ipam.get_routes_by_ip_block('context', 'block_id', 'project_id')
 
177
 
 
178
    def test_get_v4_ips_by_interface(self):
 
179
        self.mox.StubOutWithMock(self.ipam, '_get_ips_by_interface')
 
180
        self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
 
181
                                        'project_id', 4)
 
182
        self.mox.ReplayAll()
 
183
        self.ipam.get_v4_ips_by_interface('context', 'net_id', 'vif_id',
 
184
                                          'project_id')
 
185
 
 
186
    def test_get_v6_ips_by_interface(self):
 
187
        self.mox.StubOutWithMock(self.ipam, '_get_ips_by_interface')
 
188
        self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
 
189
                                        'project_id', 6)
 
190
        self.mox.ReplayAll()
 
191
        self.ipam.get_v6_ips_by_interface('context', 'net_id', 'vif_id',
 
192
                                          'project_id')
 
193
 
 
194
    def test_get_ips_by_interface(self):
 
195
        ips = [{'address': '10.10.10.10'}, {'address': '2001::CAFE'}]
 
196
        self.m_conn.get_allocated_ips('net_id', 'vif_id',
 
197
                                      'tenant_id').AndReturn(ips)
 
198
        self.m_conn.get_allocated_ips('net_id', 'vif_id',
 
199
                                      'tenant_id').AndReturn(ips)
 
200
        self.mox.ReplayAll()
 
201
        values = self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
 
202
                                        'tenant_id', 4)
 
203
        self.assertEquals(values, ["10.10.10.10"])
 
204
        values = self.ipam._get_ips_by_interface('context', 'net_id', 'vif_id',
 
205
                                        'tenant_id', 6)
 
206
        self.assertEquals(values, ["2001::CAFE"])
 
207
 
 
208
    def test_verify_subnet_exists(self):
 
209
        blocks = {'ip_blocks': [{'network_id': 'quantum_net_id'}]}
 
210
        self.m_conn.get_blocks('tenant_id').AndReturn(blocks)
 
211
        self.mox.ReplayAll()
 
212
        value = self.ipam.verify_subnet_exists('context', 'tenant_id',
 
213
                                       'quantum_net_id')
 
214
        self.assertEquals(value, True)
 
215
 
 
216
    def test_deallocate_ips_by_vif(self):
 
217
        self.m_conn.deallocate_ips('net_id', 'uuid', 'tenant_id')
 
218
        self.mox.ReplayAll()
 
219
        self.ipam.deallocate_ips_by_vif('context', 'tenant_id', 'net_id',
 
220
                {'uuid': 'uuid'})
 
221
 
 
222
    def test_get_allocated_ips(self):
 
223
        ips = [{'address': 'ip_address', 'interface_id': 'interface_id'}]
 
224
        self.m_conn.get_allocated_ips_for_network('subnet_id',
 
225
                                                  'project_id').AndReturn(ips)
 
226
        self.mox.ReplayAll()
 
227
        self.ipam.get_allocated_ips('context', 'subnet_id', 'project_id')
 
228
 
 
229
    def test_create_vif(self):
 
230
        self.m_conn.create_vif('vif_id', 'instance_id', 'project_id')
 
231
        self.mox.ReplayAll()
 
232
        self.ipam.create_vif('vif_id', 'instance_id', 'project_id')
 
233
 
 
234
    def test_get_floating_ips_by_fixed_address(self):
 
235
        value = self.ipam.get_floating_ips_by_fixed_address('context',
 
236
                                                      'fixed_address')
 
237
        self.assertEquals(value, [])