1
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
3
# Copyright 2011 Rackspace
3
# Copyright 2010 United States Government as represented by the
4
# Administrator of the National Aeronautics and Space Administration.
4
5
# All Rights Reserved.
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
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
Base class of Unit Tests for all network models
18
24
from nova import context
19
25
from nova import db
26
from nova import exception
20
27
from nova import flags
21
29
from nova import log as logging
22
30
from nova import test
23
31
from nova import utils
24
32
from nova.auth import manager
25
from nova.tests.db import fakes as db_fakes
27
34
FLAGS = flags.FLAGS
28
35
LOG = logging.getLogger('nova.tests.network')
31
38
class NetworkTestCase(test.TestCase):
39
"""Test cases for network code"""
33
41
super(NetworkTestCase, self).setUp()
42
# NOTE(vish): if you change these flags, make sure to change the
43
# flags in the corresponding section in nova-dhcpbridge
34
44
self.flags(connection_type='fake',
37
network_manager=self.network_manager)
38
47
self.manager = manager.AuthManager()
39
self.user = self.manager.create_user('netuser',
48
self.user = self.manager.create_user('netuser', 'netuser', 'netuser')
43
50
self.network = utils.import_object(FLAGS.network_manager)
44
db_fakes.stub_out_db_network_api(self.stubs)
46
self.network.network_api.db = db
47
self.context = context.RequestContext(project='fake', user=self.user)
51
self.context = context.RequestContext(project=None, user=self.user)
52
for i in range(FLAGS.num_networks):
53
name = 'project%s' % i
54
project = self.manager.create_project(name, 'netuser', name)
55
self.projects.append(project)
56
# create the necessary network data for the project
57
user_context = context.RequestContext(project=self.projects[i],
59
host = self.network.get_network_host(user_context.elevated())
60
instance_ref = self._create_instance(0)
61
self.instance_id = instance_ref['id']
62
instance_ref = self._create_instance(1)
63
self.instance2_id = instance_ref['id']
49
65
def tearDown(self):
66
# TODO(termie): this should really be instantiating clean datastores
67
# in between runs, one failure kills all the tests
68
db.instance_destroy(context.get_admin_context(), self.instance_id)
69
db.instance_destroy(context.get_admin_context(), self.instance2_id)
70
for project in self.projects:
71
self.manager.delete_project(project)
72
self.manager.delete_user(self.user)
50
73
super(NetworkTestCase, self).tearDown()
51
self.manager.delete_user(self.user.id)
55
class TestFuncs(object):
56
def _compare_fields(self, dict1, dict2, fields):
58
self.assertEqual(dict1[field], dict2[field])
60
def test_set_network_hosts(self):
61
self.network.set_network_hosts(self.context)
63
def test_set_network_host(self):
64
host = self.network.host
65
self.assertEqual(self.network.set_network_host(self.context, 0),
68
def test_allocate_for_instance(self):
70
project_id = self.context.project_id
72
self.network.set_network_hosts(self.context)
73
nw = self.network.allocate_for_instance(self.context,
74
instance_id=instance_id,
75
project_id=project_id,
76
instance_type_id=type_id)
77
static_info = [({'bridge': 'fa0', 'id': 0},
78
{'broadcast': '192.168.0.255',
79
'dns': ['192.168.0.1'],
80
'gateway': '192.168.0.1',
81
'gateway6': 'dead:beef::1',
82
'ip6s': [{'enabled': '1',
83
'ip': 'dead:beef::dcad:beff:feef:0',
85
'ips': [{'enabled': '1',
86
'ip': '192.168.0.100',
87
'netmask': '255.255.255.0'}],
89
'mac': 'DE:AD:BE:EF:00:00',
92
self._compare_fields(nw[0][0], static_info[0][0], ('bridge',))
93
self._compare_fields(nw[0][1], static_info[0][1], ('ips',
98
def test_deallocate_for_instance(self):
101
self.network.set_network_hosts(self.context)
102
self.network.add_fixed_ip_to_instance(self.context,
103
instance_id=instance_id,
104
network_id=network_id)
105
ips = db.fixed_ip_get_by_instance(self.context, instance_id)
107
self.assertTrue(ip['allocated'])
108
self.network.deallocate_for_instance(self.context,
109
instance_id=instance_id)
110
ips = db.fixed_ip_get_by_instance(self.context, instance_id)
112
self.assertFalse(ip['allocated'])
114
def test_lease_release_fixed_ip(self):
116
project_id = self.context.project_id
118
self.network.set_network_hosts(self.context)
119
nw = self.network.allocate_for_instance(self.context,
120
instance_id=instance_id,
121
project_id=project_id,
122
instance_type_id=type_id)
124
self.assertTrue(nw[0])
125
network_id = nw[0][0]['id']
127
ips = db.fixed_ip_get_by_instance(self.context, instance_id)
128
vif = db.virtual_interface_get_by_instance_and_network(self.context,
132
address = ips[0]['address']
134
db.fixed_ip_associate(self.context, address, instance_id)
135
db.fixed_ip_update(self.context, address,
136
{'virtual_interface_id': vif['id']})
138
self.network.lease_fixed_ip(self.context, vif['address'], address)
139
ip = db.fixed_ip_get_by_address(self.context, address)
140
self.assertTrue(ip['leased'])
142
self.network.release_fixed_ip(self.context, vif['address'], address)
143
ip = db.fixed_ip_get_by_address(self.context, address)
144
self.assertFalse(ip['leased'])
75
def _create_instance(self, project_num, mac=None):
77
mac = utils.generate_mac()
78
project = self.projects[project_num]
79
self.context._project = project
80
self.context.project_id = project.id
81
return db.instance_create(self.context,
82
{'project_id': project.id,
85
def _create_address(self, project_num, instance_id=None):
86
"""Create an address in given project num"""
87
if instance_id is None:
88
instance_id = self.instance_id
89
self.context._project = self.projects[project_num]
90
self.context.project_id = self.projects[project_num].id
91
return self.network.allocate_fixed_ip(self.context, instance_id)
93
def _deallocate_address(self, project_num, address):
94
self.context._project = self.projects[project_num]
95
self.context.project_id = self.projects[project_num].id
96
self.network.deallocate_fixed_ip(self.context, address)
98
def _is_allocated_in_project(self, address, project_id):
99
"""Returns true if address is in specified project"""
100
project_net = db.network_get_by_bridge(context.get_admin_context(),
101
FLAGS.flat_network_bridge)
102
network = db.fixed_ip_get_network(context.get_admin_context(),
104
instance = db.fixed_ip_get_instance(context.get_admin_context(),
106
# instance exists until release
107
return instance is not None and network['id'] == project_net['id']
109
def test_private_ipv6(self):
110
"""Make sure ipv6 is OK"""
112
instance_ref = self._create_instance(0)
113
address = self._create_address(0, instance_ref['id'])
114
network_ref = db.project_get_network(
115
context.get_admin_context(),
116
self.context.project_id)
117
address_v6 = db.instance_get_fixed_address_v6(
118
context.get_admin_context(),
120
self.assertEqual(instance_ref['mac_address'],
121
ipv6.to_mac(address_v6))
122
instance_ref2 = db.fixed_ip_get_instance_v6(
123
context.get_admin_context(),
125
self.assertEqual(instance_ref['id'], instance_ref2['id'])
126
self.assertEqual(address_v6,
127
ipv6.to_global(network_ref['cidr_v6'],
128
instance_ref['mac_address'],
130
self._deallocate_address(0, address)
131
db.instance_destroy(context.get_admin_context(),
134
def test_available_ips(self):
135
"""Make sure the number of available ips for the network is correct
137
The number of available IP addresses depends on the test
140
Network size is set in test fixture's setUp method.
142
There are ips reserved at the bottom and top of the range.
143
services (network, gateway, CloudPipe, broadcast)
145
network = db.project_get_network(context.get_admin_context(),
147
net_size = flags.FLAGS.network_size
148
admin_context = context.get_admin_context()
149
total_ips = (db.network_count_available_ips(admin_context,
151
db.network_count_reserved_ips(admin_context,
153
db.network_count_allocated_ips(admin_context,
155
self.assertEqual(total_ips, net_size)
added
removed
nova/tests/network/base.py
