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 vlan network code
24
from nova import context
26
from nova import exception
27
from nova import flags
28
from nova import log as logging
30
from nova import utils
31
from nova.auth import manager
32
from nova.tests.network import base
33
from nova.tests.network import binpath,\
37
LOG = logging.getLogger('nova.tests.network')
40
class VlanNetworkTestCase(base.NetworkTestCase):
41
"""Test cases for network code"""
42
def test_public_network_association(self):
43
"""Makes sure that we can allocaate a public ip"""
44
# TODO(vish): better way of adding floating ips
45
self.context._project = self.projects[0]
46
self.context.project_id = self.projects[0].id
47
pubnet = IPy.IP(flags.FLAGS.floating_range)
48
address = str(pubnet[0])
50
db.floating_ip_get_by_address(context.get_admin_context(), address)
51
except exception.NotFound:
52
db.floating_ip_create(context.get_admin_context(),
55
float_addr = self.network.allocate_floating_ip(self.context,
57
fix_addr = self._create_address(0)
59
self.assertEqual(float_addr, str(pubnet[0]))
60
self.network.associate_floating_ip(self.context, float_addr, fix_addr)
61
address = db.instance_get_floating_address(context.get_admin_context(),
63
self.assertEqual(address, float_addr)
64
self.network.disassociate_floating_ip(self.context, float_addr)
65
address = db.instance_get_floating_address(context.get_admin_context(),
67
self.assertEqual(address, None)
68
self.network.deallocate_floating_ip(self.context, float_addr)
69
self.network.deallocate_fixed_ip(self.context, fix_addr)
71
db.floating_ip_destroy(context.get_admin_context(), float_addr)
73
def test_allocate_deallocate_fixed_ip(self):
74
"""Makes sure that we can allocate and deallocate a fixed ip"""
75
address = self._create_address(0)
76
self.assertTrue(self._is_allocated_in_project(address,
79
self._deallocate_address(0, address)
81
# Doesn't go away until it's dhcp released
82
self.assertTrue(self._is_allocated_in_project(address,
86
self.assertFalse(self._is_allocated_in_project(address,
89
def test_side_effects(self):
90
"""Ensures allocating and releasing has no side effects"""
91
address = self._create_address(0)
92
address2 = self._create_address(1, self.instance2_id)
94
self.assertTrue(self._is_allocated_in_project(address,
96
self.assertTrue(self._is_allocated_in_project(address2,
98
self.assertFalse(self._is_allocated_in_project(address,
101
# Addresses are allocated before they're issued
105
self._deallocate_address(0, address)
107
self.assertFalse(self._is_allocated_in_project(address,
108
self.projects[0].id))
110
# First address release shouldn't affect the second
111
self.assertTrue(self._is_allocated_in_project(address2,
112
self.projects[1].id))
114
self._deallocate_address(1, address2)
116
self.assertFalse(self._is_allocated_in_project(address2,
117
self.projects[1].id))
119
def test_subnet_edge(self):
120
"""Makes sure that private ips don't overlap"""
121
first = self._create_address(0)
124
for i in range(1, FLAGS.num_networks):
125
instance_ref = self._create_instance(i, mac=utils.generate_mac())
126
instance_ids.append(instance_ref['id'])
127
address = self._create_address(i, instance_ref['id'])
128
instance_ref = self._create_instance(i, mac=utils.generate_mac())
129
instance_ids.append(instance_ref['id'])
130
address2 = self._create_address(i, instance_ref['id'])
131
instance_ref = self._create_instance(i, mac=utils.generate_mac())
132
instance_ids.append(instance_ref['id'])
133
address3 = self._create_address(i, instance_ref['id'])
137
self.context._project = self.projects[i]
138
self.context.project_id = self.projects[i].id
139
self.assertFalse(self._is_allocated_in_project(address,
140
self.projects[0].id))
141
self.assertFalse(self._is_allocated_in_project(address2,
142
self.projects[0].id))
143
self.assertFalse(self._is_allocated_in_project(address3,
144
self.projects[0].id))
145
self.network.deallocate_fixed_ip(self.context, address)
146
self.network.deallocate_fixed_ip(self.context, address2)
147
self.network.deallocate_fixed_ip(self.context, address3)
151
for instance_id in instance_ids:
152
db.instance_destroy(context.get_admin_context(), instance_id)
153
self.context._project = self.projects[0]
154
self.context.project_id = self.projects[0].id
155
self.network.deallocate_fixed_ip(self.context, first)
156
self._deallocate_address(0, first)
159
def test_vpn_ip_and_port_looks_valid(self):
160
"""Ensure the vpn ip and port are reasonable"""
161
self.assert_(self.projects[0].vpn_ip)
162
self.assert_(self.projects[0].vpn_port >= FLAGS.vpn_start)
163
self.assert_(self.projects[0].vpn_port <= FLAGS.vpn_start +
166
def test_too_many_networks(self):
167
"""Ensure error is raised if we run out of networks"""
169
networks_left = (FLAGS.num_networks -
170
db.network_count(context.get_admin_context()))
171
for i in range(networks_left):
172
project = self.manager.create_project('many%s' % i, self.user)
173
projects.append(project)
174
db.project_get_network(context.get_admin_context(), project.id)
175
project = self.manager.create_project('last', self.user)
176
projects.append(project)
177
self.assertRaises(db.NoMoreNetworks,
178
db.project_get_network,
179
context.get_admin_context(),
181
for project in projects:
182
self.manager.delete_project(project)
184
def test_ips_are_reused(self):
185
"""Makes sure that ip addresses that are deallocated get reused"""
186
address = self._create_address(0)
188
self.network.deallocate_fixed_ip(self.context, address)
191
address2 = self._create_address(0)
192
self.assertEqual(address, address2)
194
self.network.deallocate_fixed_ip(self.context, address2)
197
def test_too_many_addresses(self):
198
"""Test for a NoMoreAddresses exception when all fixed ips are used.
200
admin_context = context.get_admin_context()
201
network = db.project_get_network(admin_context, self.projects[0].id)
202
num_available_ips = db.network_count_available_ips(admin_context,
206
for i in range(num_available_ips):
207
instance_ref = self._create_instance(0)
208
instance_ids.append(instance_ref['id'])
209
address = self._create_address(0, instance_ref['id'])
210
addresses.append(address)
213
ip_count = db.network_count_available_ips(context.get_admin_context(),
215
self.assertEqual(ip_count, 0)
216
self.assertRaises(db.NoMoreAddresses,
217
self.network.allocate_fixed_ip,
221
for i in range(num_available_ips):
222
self.network.deallocate_fixed_ip(self.context, addresses[i])
223
release_ip(addresses[i])
224
db.instance_destroy(context.get_admin_context(), instance_ids[i])
225
ip_count = db.network_count_available_ips(context.get_admin_context(),
227
self.assertEqual(ip_count, num_available_ips)
229
def _is_allocated_in_project(self, address, project_id):
230
"""Returns true if address is in specified project"""
231
project_net = db.project_get_network(context.get_admin_context(),
233
network = db.fixed_ip_get_network(context.get_admin_context(),
235
instance = db.fixed_ip_get_instance(context.get_admin_context(),
237
# instance exists until release
238
return instance is not None and network['id'] == project_net['id']
240
def run(self, result=None):
241
if(FLAGS.network_manager == 'nova.network.manager.VlanManager'):
242
super(VlanNetworkTestCase, self).run(result)
added
removed
nova/tests/test_vlan_network.py
