From 1831f31af0ac21ded3535f15777bd5147c615c34 Mon Sep 17 00:00:00 2001 From: Tushar Patil Date: Fri, 4 Mar 2011 11:52:18 -0800 Subject: added flatmanager unit testcases and renamed test_network.py to test_vlan_network.py --- nova/tests/test_flat_network.py | 276 +++++++++++++++++++++++++++++ nova/tests/test_network.py | 369 --------------------------------------- nova/tests/test_vlan_network.py | 373 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 649 insertions(+), 369 deletions(-) create mode 100644 nova/tests/test_flat_network.py delete mode 100644 nova/tests/test_network.py create mode 100644 nova/tests/test_vlan_network.py (limited to 'nova/tests') diff --git a/nova/tests/test_flat_network.py b/nova/tests/test_flat_network.py new file mode 100644 index 000000000..91a49920d --- /dev/null +++ b/nova/tests/test_flat_network.py @@ -0,0 +1,276 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +""" +Unit Tests for network code +""" +import IPy +import os +import unittest + +from nova import context +from nova import db +from nova import exception +from nova import flags +from nova import log as logging +from nova import test +from nova import utils +from nova.auth import manager + +FLAGS = flags.FLAGS +LOG = logging.getLogger('nova.tests.network') + + +class FlatNetworkTestCase(test.TestCase): + """Test cases for network code""" + def setUp(self): + super(FlatNetworkTestCase, self).setUp() + # NOTE(vish): if you change these flags, make sure to change the + # flags in the corresponding section in nova-dhcpbridge + self.flags(connection_type='fake', + fake_call=True, + fake_network=True) + self.manager = manager.AuthManager() + self.user = self.manager.create_user('netuser', 'netuser', 'netuser') + self.projects = [] + self.network = utils.import_object(FLAGS.network_manager) + self.context = context.RequestContext(project=None, user=self.user) + for i in range(5): + name = 'project%s' % i + project = self.manager.create_project(name, 'netuser', name) + self.projects.append(project) + # create the necessary network data for the project + user_context = context.RequestContext(project=self.projects[i], + user=self.user) + host = self.network.get_network_host(user_context.elevated()) + instance_ref = self._create_instance(0) + self.instance_id = instance_ref['id'] + instance_ref = self._create_instance(1) + self.instance2_id = instance_ref['id'] + + def tearDown(self): + # TODO(termie): this should really be instantiating clean datastores + # in between runs, one failure kills all the tests + db.instance_destroy(context.get_admin_context(), self.instance_id) + db.instance_destroy(context.get_admin_context(), self.instance2_id) + for project in self.projects: + self.manager.delete_project(project) + self.manager.delete_user(self.user) + super(FlatNetworkTestCase, self).tearDown() + + def _create_instance(self, project_num, mac=None): + if not mac: + mac = utils.generate_mac() + project = self.projects[project_num] + self.context._project = project + self.context.project_id = project.id + return db.instance_create(self.context, + {'project_id': project.id, + 'mac_address': mac}) + + def _create_address(self, project_num, instance_id=None): + """Create an address in given project num""" + if instance_id is None: + instance_id = self.instance_id + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + return self.network.allocate_fixed_ip(self.context, instance_id) + + def _deallocate_address(self, project_num, address): + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + self.network.deallocate_fixed_ip(self.context, address) + + def test_private_ipv6(self): + """Make sure ipv6 is OK""" + if FLAGS.use_ipv6: + instance_ref = self._create_instance(0) + address = self._create_address(0, instance_ref['id']) + network_ref = db.project_get_network( + context.get_admin_context(), + self.context.project_id) + address_v6 = db.instance_get_fixed_address_v6( + context.get_admin_context(), + instance_ref['id']) + self.assertEqual(instance_ref['mac_address'], + utils.to_mac(address_v6)) + instance_ref2 = db.fixed_ip_get_instance_v6( + context.get_admin_context(), + address_v6) + self.assertEqual(instance_ref['id'], instance_ref2['id']) + self.assertEqual(address_v6, + utils.to_global_ipv6( + network_ref['cidr_v6'], + instance_ref['mac_address'])) + self._deallocate_address(0, address) + db.instance_destroy(context.get_admin_context(), + instance_ref['id']) + + def test_public_network_association(self): + """Makes sure that we can allocate a public ip""" + # TODO(vish): better way of adding floating ips + + self.context._project = self.projects[0] + self.context.project_id = self.projects[0].id + pubnet = IPy.IP(flags.FLAGS.floating_range) + address = str(pubnet[0]) + try: + db.floating_ip_get_by_address(context.get_admin_context(), address) + except exception.NotFound: + db.floating_ip_create(context.get_admin_context(), + {'address': address, + 'host': FLAGS.host}) + + self.assertRaises(NotImplementedError, + self.network.allocate_floating_ip, + self.context, self.projects[0].id) + + fix_addr = self._create_address(0) + float_addr = address + self.assertRaises(NotImplementedError, + self.network.associate_floating_ip, + self.context, float_addr, fix_addr) + + address = db.instance_get_floating_address(context.get_admin_context(), + self.instance_id) + self.assertEqual(address, None) + + self.assertRaises(NotImplementedError, + self.network.disassociate_floating_ip, + self.context, float_addr) + + address = db.instance_get_floating_address(context.get_admin_context(), + self.instance_id) + self.assertEqual(address, None) + + self.assertRaises(NotImplementedError, + self.network.deallocate_floating_ip, + self.context, float_addr) + + self.network.deallocate_fixed_ip(self.context, fix_addr) + db.floating_ip_destroy(context.get_admin_context(), float_addr) + + def test_allocate_deallocate_fixed_ip(self): + """Makes sure that we can allocate and deallocate a fixed ip""" + address = self._create_address(0) + self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + self._deallocate_address(0, address) + + # check if the fixed ip address is really deallocated + self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + + def test_side_effects(self): + """Ensures allocating and releasing has no side effects""" + address = self._create_address(0) + address2 = self._create_address(1, self.instance2_id) + + self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) + + self._deallocate_address(0, address) + self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + + # First address release shouldn't affect the second + self.assertTrue(is_allocated_in_project(address2, self.projects[0].id)) + + self._deallocate_address(1, address2) + self.assertFalse(is_allocated_in_project(address2, + self.projects[1].id)) + + def test_ips_are_reused(self): + """Makes sure that ip addresses that are deallocated get reused""" + address = self._create_address(0) + self.network.deallocate_fixed_ip(self.context, address) + + address2 = self._create_address(0) + self.assertEqual(address, address2) + + self.network.deallocate_fixed_ip(self.context, address2) + + def test_available_ips(self): + """Make sure the number of available ips for the network is correct + + The number of available IP addresses depends on the test + environment's setup. + + Network size is set in test fixture's setUp method. + + There are ips reserved at the bottom and top of the range. + services (network, gateway, CloudPipe, broadcast) + """ + network = db.project_get_network(context.get_admin_context(), + self.projects[0].id) + net_size = flags.FLAGS.network_size + admin_context = context.get_admin_context() + total_ips = (db.network_count_available_ips(admin_context, + network['id']) + + db.network_count_reserved_ips(admin_context, + network['id']) + + db.network_count_allocated_ips(admin_context, + network['id'])) + self.assertEqual(total_ips, net_size) + + def test_too_many_addresses(self): + """Test for a NoMoreAddresses exception when all fixed ips are used. + """ + admin_context = context.get_admin_context() + network = db.project_get_network(admin_context, self.projects[0].id) + num_available_ips = db.network_count_available_ips(admin_context, + network['id']) + addresses = [] + instance_ids = [] + for i in range(num_available_ips): + instance_ref = self._create_instance(0) + instance_ids.append(instance_ref['id']) + address = self._create_address(0, instance_ref['id']) + addresses.append(address) + + ip_count = db.network_count_available_ips(context.get_admin_context(), + network['id']) + self.assertEqual(ip_count, 0) + self.assertRaises(db.NoMoreAddresses, + self.network.allocate_fixed_ip, + self.context, + 'foo') + + for i in range(num_available_ips): + self.network.deallocate_fixed_ip(self.context, addresses[i]) + db.instance_destroy(context.get_admin_context(), instance_ids[i]) + ip_count = db.network_count_available_ips(context.get_admin_context(), + network['id']) + self.assertEqual(ip_count, num_available_ips) + + def run(self, result=None): + if(FLAGS.network_manager == 'nova.network.manager.FlatManager'): + super(FlatNetworkTestCase, self).run(result) + + +def is_allocated_in_project(address, project_id): + """Returns true if address is in specified project""" + #project_net = db.project_get_network(context.get_admin_context(), + # project_id) + project_net = db.network_get_by_bridge(context.get_admin_context(), + FLAGS.flat_network_bridge) + network = db.fixed_ip_get_network(context.get_admin_context(), address) + instance = db.fixed_ip_get_instance(context.get_admin_context(), address) + # instance exists until release + return instance is not None and network['id'] == project_net['id'] + + +def binpath(script): + """Returns the absolute path to a script in bin""" + return os.path.abspath(os.path.join(__file__, "../../../bin", script)) diff --git a/nova/tests/test_network.py b/nova/tests/test_network.py deleted file mode 100644 index ce1c77210..000000000 --- a/nova/tests/test_network.py +++ /dev/null @@ -1,369 +0,0 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -""" -Unit Tests for network code -""" -import IPy -import os - -from nova import context -from nova import db -from nova import exception -from nova import flags -from nova import log as logging -from nova import test -from nova import utils -from nova.auth import manager - -FLAGS = flags.FLAGS -LOG = logging.getLogger('nova.tests.network') - - -class NetworkTestCase(test.TestCase): - """Test cases for network code""" - def setUp(self): - super(NetworkTestCase, self).setUp() - # NOTE(vish): if you change these flags, make sure to change the - # flags in the corresponding section in nova-dhcpbridge - self.flags(connection_type='fake', - fake_call=True, - fake_network=True) - self.manager = manager.AuthManager() - self.user = self.manager.create_user('netuser', 'netuser', 'netuser') - self.projects = [] - self.network = utils.import_object(FLAGS.network_manager) - self.context = context.RequestContext(project=None, user=self.user) - for i in range(FLAGS.num_networks): - name = 'project%s' % i - project = self.manager.create_project(name, 'netuser', name) - self.projects.append(project) - # create the necessary network data for the project - user_context = context.RequestContext(project=self.projects[i], - user=self.user) - host = self.network.get_network_host(user_context.elevated()) - instance_ref = self._create_instance(0) - self.instance_id = instance_ref['id'] - instance_ref = self._create_instance(1) - self.instance2_id = instance_ref['id'] - - def tearDown(self): - # TODO(termie): this should really be instantiating clean datastores - # in between runs, one failure kills all the tests - db.instance_destroy(context.get_admin_context(), self.instance_id) - db.instance_destroy(context.get_admin_context(), self.instance2_id) - for project in self.projects: - self.manager.delete_project(project) - self.manager.delete_user(self.user) - super(NetworkTestCase, self).tearDown() - - def _create_instance(self, project_num, mac=None): - if not mac: - mac = utils.generate_mac() - project = self.projects[project_num] - self.context._project = project - self.context.project_id = project.id - return db.instance_create(self.context, - {'project_id': project.id, - 'mac_address': mac}) - - def _create_address(self, project_num, instance_id=None): - """Create an address in given project num""" - if instance_id is None: - instance_id = self.instance_id - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - return self.network.allocate_fixed_ip(self.context, instance_id) - - def _deallocate_address(self, project_num, address): - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - self.network.deallocate_fixed_ip(self.context, address) - - def test_private_ipv6(self): - """Make sure ipv6 is OK""" - if FLAGS.use_ipv6: - instance_ref = self._create_instance(0) - address = self._create_address(0, instance_ref['id']) - network_ref = db.project_get_network( - context.get_admin_context(), - self.context.project_id) - address_v6 = db.instance_get_fixed_address_v6( - context.get_admin_context(), - instance_ref['id']) - self.assertEqual(instance_ref['mac_address'], - utils.to_mac(address_v6)) - instance_ref2 = db.fixed_ip_get_instance_v6( - context.get_admin_context(), - address_v6) - self.assertEqual(instance_ref['id'], instance_ref2['id']) - self.assertEqual(address_v6, - utils.to_global_ipv6( - network_ref['cidr_v6'], - instance_ref['mac_address'])) - self._deallocate_address(0, address) - db.instance_destroy(context.get_admin_context(), - instance_ref['id']) - - def test_public_network_association(self): - """Makes sure that we can allocaate a public ip""" - # TODO(vish): better way of adding floating ips - self.context._project = self.projects[0] - self.context.project_id = self.projects[0].id - pubnet = IPy.IP(flags.FLAGS.floating_range) - address = str(pubnet[0]) - try: - db.floating_ip_get_by_address(context.get_admin_context(), address) - except exception.NotFound: - db.floating_ip_create(context.get_admin_context(), - {'address': address, - 'host': FLAGS.host}) - float_addr = self.network.allocate_floating_ip(self.context, - self.projects[0].id) - fix_addr = self._create_address(0) - lease_ip(fix_addr) - self.assertEqual(float_addr, str(pubnet[0])) - self.network.associate_floating_ip(self.context, float_addr, fix_addr) - address = db.instance_get_floating_address(context.get_admin_context(), - self.instance_id) - self.assertEqual(address, float_addr) - self.network.disassociate_floating_ip(self.context, float_addr) - address = db.instance_get_floating_address(context.get_admin_context(), - self.instance_id) - self.assertEqual(address, None) - self.network.deallocate_floating_ip(self.context, float_addr) - self.network.deallocate_fixed_ip(self.context, fix_addr) - release_ip(fix_addr) - db.floating_ip_destroy(context.get_admin_context(), float_addr) - - def test_allocate_deallocate_fixed_ip(self): - """Makes sure that we can allocate and deallocate a fixed ip""" - address = self._create_address(0) - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) - lease_ip(address) - self._deallocate_address(0, address) - - # Doesn't go away until it's dhcp released - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) - - release_ip(address) - self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) - - def test_side_effects(self): - """Ensures allocating and releasing has no side effects""" - address = self._create_address(0) - address2 = self._create_address(1, self.instance2_id) - - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) - self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) - self.assertFalse(is_allocated_in_project(address, self.projects[1].id)) - - # Addresses are allocated before they're issued - lease_ip(address) - lease_ip(address2) - - self._deallocate_address(0, address) - release_ip(address) - self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) - - # First address release shouldn't affect the second - self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) - - self._deallocate_address(1, address2) - release_ip(address2) - self.assertFalse(is_allocated_in_project(address2, - self.projects[1].id)) - - def test_subnet_edge(self): - """Makes sure that private ips don't overlap""" - first = self._create_address(0) - lease_ip(first) - instance_ids = [] - for i in range(1, FLAGS.num_networks): - instance_ref = self._create_instance(i, mac=utils.generate_mac()) - instance_ids.append(instance_ref['id']) - address = self._create_address(i, instance_ref['id']) - instance_ref = self._create_instance(i, mac=utils.generate_mac()) - instance_ids.append(instance_ref['id']) - address2 = self._create_address(i, instance_ref['id']) - instance_ref = self._create_instance(i, mac=utils.generate_mac()) - instance_ids.append(instance_ref['id']) - address3 = self._create_address(i, instance_ref['id']) - lease_ip(address) - lease_ip(address2) - lease_ip(address3) - self.context._project = self.projects[i] - self.context.project_id = self.projects[i].id - self.assertFalse(is_allocated_in_project(address, - self.projects[0].id)) - self.assertFalse(is_allocated_in_project(address2, - self.projects[0].id)) - self.assertFalse(is_allocated_in_project(address3, - self.projects[0].id)) - self.network.deallocate_fixed_ip(self.context, address) - self.network.deallocate_fixed_ip(self.context, address2) - self.network.deallocate_fixed_ip(self.context, address3) - release_ip(address) - release_ip(address2) - release_ip(address3) - for instance_id in instance_ids: - db.instance_destroy(context.get_admin_context(), instance_id) - self.context._project = self.projects[0] - self.context.project_id = self.projects[0].id - self.network.deallocate_fixed_ip(self.context, first) - self._deallocate_address(0, first) - release_ip(first) - - def test_vpn_ip_and_port_looks_valid(self): - """Ensure the vpn ip and port are reasonable""" - self.assert_(self.projects[0].vpn_ip) - self.assert_(self.projects[0].vpn_port >= FLAGS.vpn_start) - self.assert_(self.projects[0].vpn_port <= FLAGS.vpn_start + - FLAGS.num_networks) - - def test_too_many_networks(self): - """Ensure error is raised if we run out of networks""" - projects = [] - networks_left = (FLAGS.num_networks - - db.network_count(context.get_admin_context())) - for i in range(networks_left): - project = self.manager.create_project('many%s' % i, self.user) - projects.append(project) - db.project_get_network(context.get_admin_context(), project.id) - project = self.manager.create_project('last', self.user) - projects.append(project) - self.assertRaises(db.NoMoreNetworks, - db.project_get_network, - context.get_admin_context(), - project.id) - for project in projects: - self.manager.delete_project(project) - - def test_ips_are_reused(self): - """Makes sure that ip addresses that are deallocated get reused""" - address = self._create_address(0) - lease_ip(address) - self.network.deallocate_fixed_ip(self.context, address) - release_ip(address) - - address2 = self._create_address(0) - self.assertEqual(address, address2) - lease_ip(address) - self.network.deallocate_fixed_ip(self.context, address2) - release_ip(address) - - def test_available_ips(self): - """Make sure the number of available ips for the network is correct - - The number of available IP addresses depends on the test - environment's setup. - - Network size is set in test fixture's setUp method. - - There are ips reserved at the bottom and top of the range. - services (network, gateway, CloudPipe, broadcast) - """ - network = db.project_get_network(context.get_admin_context(), - self.projects[0].id) - net_size = flags.FLAGS.network_size - admin_context = context.get_admin_context() - total_ips = (db.network_count_available_ips(admin_context, - network['id']) + - db.network_count_reserved_ips(admin_context, - network['id']) + - db.network_count_allocated_ips(admin_context, - network['id'])) - self.assertEqual(total_ips, net_size) - - def test_too_many_addresses(self): - """Test for a NoMoreAddresses exception when all fixed ips are used. - """ - admin_context = context.get_admin_context() - network = db.project_get_network(admin_context, self.projects[0].id) - num_available_ips = db.network_count_available_ips(admin_context, - network['id']) - addresses = [] - instance_ids = [] - for i in range(num_available_ips): - instance_ref = self._create_instance(0) - instance_ids.append(instance_ref['id']) - address = self._create_address(0, instance_ref['id']) - addresses.append(address) - lease_ip(address) - - ip_count = db.network_count_available_ips(context.get_admin_context(), - network['id']) - self.assertEqual(ip_count, 0) - self.assertRaises(db.NoMoreAddresses, - self.network.allocate_fixed_ip, - self.context, - 'foo') - - for i in range(num_available_ips): - self.network.deallocate_fixed_ip(self.context, addresses[i]) - release_ip(addresses[i]) - db.instance_destroy(context.get_admin_context(), instance_ids[i]) - ip_count = db.network_count_available_ips(context.get_admin_context(), - network['id']) - self.assertEqual(ip_count, num_available_ips) - - -def is_allocated_in_project(address, project_id): - """Returns true if address is in specified project""" - project_net = db.project_get_network(context.get_admin_context(), - project_id) - network = db.fixed_ip_get_network(context.get_admin_context(), address) - instance = db.fixed_ip_get_instance(context.get_admin_context(), address) - # instance exists until release - return instance is not None and network['id'] == project_net['id'] - - -def binpath(script): - """Returns the absolute path to a script in bin""" - return os.path.abspath(os.path.join(__file__, "../../../bin", script)) - - -def lease_ip(private_ip): - """Run add command on dhcpbridge""" - network_ref = db.fixed_ip_get_network(context.get_admin_context(), - private_ip) - instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), - private_ip) - cmd = "%s add %s %s fake" % (binpath('nova-dhcpbridge'), - instance_ref['mac_address'], - private_ip) - env = {'DNSMASQ_INTERFACE': network_ref['bridge'], - 'TESTING': '1', - 'FLAGFILE': FLAGS.dhcpbridge_flagfile} - (out, err) = utils.execute(cmd, addl_env=env) - LOG.debug("ISSUE_IP: %s, %s ", out, err) - - -def release_ip(private_ip): - """Run del command on dhcpbridge""" - network_ref = db.fixed_ip_get_network(context.get_admin_context(), - private_ip) - instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), - private_ip) - cmd = "%s del %s %s fake" % (binpath('nova-dhcpbridge'), - instance_ref['mac_address'], - private_ip) - env = {'DNSMASQ_INTERFACE': network_ref['bridge'], - 'TESTING': '1', - 'FLAGFILE': FLAGS.dhcpbridge_flagfile} - (out, err) = utils.execute(cmd, addl_env=env) - LOG.debug("RELEASE_IP: %s, %s ", out, err) diff --git a/nova/tests/test_vlan_network.py b/nova/tests/test_vlan_network.py new file mode 100644 index 000000000..be39313bb --- /dev/null +++ b/nova/tests/test_vlan_network.py @@ -0,0 +1,373 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +""" +Unit Tests for network code +""" +import IPy +import os + +from nova import context +from nova import db +from nova import exception +from nova import flags +from nova import log as logging +from nova import test +from nova import utils +from nova.auth import manager + +FLAGS = flags.FLAGS +LOG = logging.getLogger('nova.tests.network') + + +class VlanNetworkTestCase(test.TestCase): + """Test cases for network code""" + def setUp(self): + super(VlanNetworkTestCase, self).setUp() + # NOTE(vish): if you change these flags, make sure to change the + # flags in the corresponding section in nova-dhcpbridge + self.flags(connection_type='fake', + fake_call=True, + fake_network=True) + self.manager = manager.AuthManager() + self.user = self.manager.create_user('netuser', 'netuser', 'netuser') + self.projects = [] + self.network = utils.import_object(FLAGS.network_manager) + self.context = context.RequestContext(project=None, user=self.user) + for i in range(FLAGS.num_networks): + name = 'project%s' % i + project = self.manager.create_project(name, 'netuser', name) + self.projects.append(project) + # create the necessary network data for the project + user_context = context.RequestContext(project=self.projects[i], + user=self.user) + host = self.network.get_network_host(user_context.elevated()) + instance_ref = self._create_instance(0) + self.instance_id = instance_ref['id'] + instance_ref = self._create_instance(1) + self.instance2_id = instance_ref['id'] + + def tearDown(self): + # TODO(termie): this should really be instantiating clean datastores + # in between runs, one failure kills all the tests + db.instance_destroy(context.get_admin_context(), self.instance_id) + db.instance_destroy(context.get_admin_context(), self.instance2_id) + for project in self.projects: + self.manager.delete_project(project) + self.manager.delete_user(self.user) + super(VlanNetworkTestCase, self).tearDown() + + def run(self, result=None): + if(FLAGS.network_manager == 'nova.network.manager.VlanManager'): + super(VlanNetworkTestCase, self).run(result) + + def _create_instance(self, project_num, mac=None): + if not mac: + mac = utils.generate_mac() + project = self.projects[project_num] + self.context._project = project + self.context.project_id = project.id + return db.instance_create(self.context, + {'project_id': project.id, + 'mac_address': mac}) + + def _create_address(self, project_num, instance_id=None): + """Create an address in given project num""" + if instance_id is None: + instance_id = self.instance_id + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + return self.network.allocate_fixed_ip(self.context, instance_id) + + def _deallocate_address(self, project_num, address): + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + self.network.deallocate_fixed_ip(self.context, address) + + def test_private_ipv6(self): + """Make sure ipv6 is OK""" + if FLAGS.use_ipv6: + instance_ref = self._create_instance(0) + address = self._create_address(0, instance_ref['id']) + network_ref = db.project_get_network( + context.get_admin_context(), + self.context.project_id) + address_v6 = db.instance_get_fixed_address_v6( + context.get_admin_context(), + instance_ref['id']) + self.assertEqual(instance_ref['mac_address'], + utils.to_mac(address_v6)) + instance_ref2 = db.fixed_ip_get_instance_v6( + context.get_admin_context(), + address_v6) + self.assertEqual(instance_ref['id'], instance_ref2['id']) + self.assertEqual(address_v6, + utils.to_global_ipv6( + network_ref['cidr_v6'], + instance_ref['mac_address'])) + self._deallocate_address(0, address) + db.instance_destroy(context.get_admin_context(), + instance_ref['id']) + + def test_public_network_association(self): + """Makes sure that we can allocaate a public ip""" + # TODO(vish): better way of adding floating ips + self.context._project = self.projects[0] + self.context.project_id = self.projects[0].id + pubnet = IPy.IP(flags.FLAGS.floating_range) + address = str(pubnet[0]) + try: + db.floating_ip_get_by_address(context.get_admin_context(), address) + except exception.NotFound: + db.floating_ip_create(context.get_admin_context(), + {'address': address, + 'host': FLAGS.host}) + float_addr = self.network.allocate_floating_ip(self.context, + self.projects[0].id) + fix_addr = self._create_address(0) + lease_ip(fix_addr) + self.assertEqual(float_addr, str(pubnet[0])) + self.network.associate_floating_ip(self.context, float_addr, fix_addr) + address = db.instance_get_floating_address(context.get_admin_context(), + self.instance_id) + self.assertEqual(address, float_addr) + self.network.disassociate_floating_ip(self.context, float_addr) + address = db.instance_get_floating_address(context.get_admin_context(), + self.instance_id) + self.assertEqual(address, None) + self.network.deallocate_floating_ip(self.context, float_addr) + self.network.deallocate_fixed_ip(self.context, fix_addr) + release_ip(fix_addr) + db.floating_ip_destroy(context.get_admin_context(), float_addr) + + def test_allocate_deallocate_fixed_ip(self): + """Makes sure that we can allocate and deallocate a fixed ip""" + address = self._create_address(0) + self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + lease_ip(address) + self._deallocate_address(0, address) + + # Doesn't go away until it's dhcp released + self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + + release_ip(address) + self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + + def test_side_effects(self): + """Ensures allocating and releasing has no side effects""" + address = self._create_address(0) + address2 = self._create_address(1, self.instance2_id) + + self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) + self.assertFalse(is_allocated_in_project(address, self.projects[1].id)) + + # Addresses are allocated before they're issued + lease_ip(address) + lease_ip(address2) + + self._deallocate_address(0, address) + release_ip(address) + self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + + # First address release shouldn't affect the second + self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) + + self._deallocate_address(1, address2) + release_ip(address2) + self.assertFalse(is_allocated_in_project(address2, + self.projects[1].id)) + + def test_subnet_edge(self): + """Makes sure that private ips don't overlap""" + first = self._create_address(0) + lease_ip(first) + instance_ids = [] + for i in range(1, FLAGS.num_networks): + instance_ref = self._create_instance(i, mac=utils.generate_mac()) + instance_ids.append(instance_ref['id']) + address = self._create_address(i, instance_ref['id']) + instance_ref = self._create_instance(i, mac=utils.generate_mac()) + instance_ids.append(instance_ref['id']) + address2 = self._create_address(i, instance_ref['id']) + instance_ref = self._create_instance(i, mac=utils.generate_mac()) + instance_ids.append(instance_ref['id']) + address3 = self._create_address(i, instance_ref['id']) + lease_ip(address) + lease_ip(address2) + lease_ip(address3) + self.context._project = self.projects[i] + self.context.project_id = self.projects[i].id + self.assertFalse(is_allocated_in_project(address, + self.projects[0].id)) + self.assertFalse(is_allocated_in_project(address2, + self.projects[0].id)) + self.assertFalse(is_allocated_in_project(address3, + self.projects[0].id)) + self.network.deallocate_fixed_ip(self.context, address) + self.network.deallocate_fixed_ip(self.context, address2) + self.network.deallocate_fixed_ip(self.context, address3) + release_ip(address) + release_ip(address2) + release_ip(address3) + for instance_id in instance_ids: + db.instance_destroy(context.get_admin_context(), instance_id) + self.context._project = self.projects[0] + self.context.project_id = self.projects[0].id + self.network.deallocate_fixed_ip(self.context, first) + self._deallocate_address(0, first) + release_ip(first) + + def test_vpn_ip_and_port_looks_valid(self): + """Ensure the vpn ip and port are reasonable""" + self.assert_(self.projects[0].vpn_ip) + self.assert_(self.projects[0].vpn_port >= FLAGS.vpn_start) + self.assert_(self.projects[0].vpn_port <= FLAGS.vpn_start + + FLAGS.num_networks) + + def test_too_many_networks(self): + """Ensure error is raised if we run out of networks""" + projects = [] + networks_left = (FLAGS.num_networks - + db.network_count(context.get_admin_context())) + for i in range(networks_left): + project = self.manager.create_project('many%s' % i, self.user) + projects.append(project) + db.project_get_network(context.get_admin_context(), project.id) + project = self.manager.create_project('last', self.user) + projects.append(project) + self.assertRaises(db.NoMoreNetworks, + db.project_get_network, + context.get_admin_context(), + project.id) + for project in projects: + self.manager.delete_project(project) + + def test_ips_are_reused(self): + """Makes sure that ip addresses that are deallocated get reused""" + address = self._create_address(0) + lease_ip(address) + self.network.deallocate_fixed_ip(self.context, address) + release_ip(address) + + address2 = self._create_address(0) + self.assertEqual(address, address2) + lease_ip(address) + self.network.deallocate_fixed_ip(self.context, address2) + release_ip(address) + + def test_available_ips(self): + """Make sure the number of available ips for the network is correct + + The number of available IP addresses depends on the test + environment's setup. + + Network size is set in test fixture's setUp method. + + There are ips reserved at the bottom and top of the range. + services (network, gateway, CloudPipe, broadcast) + """ + network = db.project_get_network(context.get_admin_context(), + self.projects[0].id) + net_size = flags.FLAGS.network_size + admin_context = context.get_admin_context() + total_ips = (db.network_count_available_ips(admin_context, + network['id']) + + db.network_count_reserved_ips(admin_context, + network['id']) + + db.network_count_allocated_ips(admin_context, + network['id'])) + self.assertEqual(total_ips, net_size) + + def test_too_many_addresses(self): + """Test for a NoMoreAddresses exception when all fixed ips are used. + """ + admin_context = context.get_admin_context() + network = db.project_get_network(admin_context, self.projects[0].id) + num_available_ips = db.network_count_available_ips(admin_context, + network['id']) + addresses = [] + instance_ids = [] + for i in range(num_available_ips): + instance_ref = self._create_instance(0) + instance_ids.append(instance_ref['id']) + address = self._create_address(0, instance_ref['id']) + addresses.append(address) + lease_ip(address) + + ip_count = db.network_count_available_ips(context.get_admin_context(), + network['id']) + self.assertEqual(ip_count, 0) + self.assertRaises(db.NoMoreAddresses, + self.network.allocate_fixed_ip, + self.context, + 'foo') + + for i in range(num_available_ips): + self.network.deallocate_fixed_ip(self.context, addresses[i]) + release_ip(addresses[i]) + db.instance_destroy(context.get_admin_context(), instance_ids[i]) + ip_count = db.network_count_available_ips(context.get_admin_context(), + network['id']) + self.assertEqual(ip_count, num_available_ips) + + +def is_allocated_in_project(address, project_id): + """Returns true if address is in specified project""" + project_net = db.project_get_network(context.get_admin_context(), + project_id) + network = db.fixed_ip_get_network(context.get_admin_context(), address) + instance = db.fixed_ip_get_instance(context.get_admin_context(), address) + # instance exists until release + return instance is not None and network['id'] == project_net['id'] + + +def binpath(script): + """Returns the absolute path to a script in bin""" + return os.path.abspath(os.path.join(__file__, "../../../bin", script)) + + +def lease_ip(private_ip): + """Run add command on dhcpbridge""" + network_ref = db.fixed_ip_get_network(context.get_admin_context(), + private_ip) + instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), + private_ip) + cmd = "%s add %s %s fake" % (binpath('nova-dhcpbridge'), + instance_ref['mac_address'], + private_ip) + env = {'DNSMASQ_INTERFACE': network_ref['bridge'], + 'TESTING': '1', + 'FLAGFILE': FLAGS.dhcpbridge_flagfile} + (out, err) = utils.execute(cmd, addl_env=env) + LOG.debug("ISSUE_IP: %s, %s ", out, err) + + +def release_ip(private_ip): + """Run del command on dhcpbridge""" + network_ref = db.fixed_ip_get_network(context.get_admin_context(), + private_ip) + instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), + private_ip) + cmd = "%s del %s %s fake" % (binpath('nova-dhcpbridge'), + instance_ref['mac_address'], + private_ip) + env = {'DNSMASQ_INTERFACE': network_ref['bridge'], + 'TESTING': '1', + 'FLAGFILE': FLAGS.dhcpbridge_flagfile} + (out, err) = utils.execute(cmd, addl_env=env) + LOG.debug("RELEASE_IP: %s, %s ", out, err) -- cgit From a74bf3381ada34f35c43d6f307fbae9abecfb255 Mon Sep 17 00:00:00 2001 From: Tushar Patil Date: Tue, 8 Mar 2011 10:30:48 -0800 Subject: abstracted network code in the base class for flat and vlan --- nova/tests/network/__init__.py | 47 +++++++++ nova/tests/network/base.py | 154 ++++++++++++++++++++++++++++++ nova/tests/test_flat_network.py | 147 ++++------------------------ nova/tests/test_vlan_network.py | 205 ++++++++-------------------------------- 4 files changed, 254 insertions(+), 299 deletions(-) create mode 100644 nova/tests/network/__init__.py create mode 100644 nova/tests/network/base.py (limited to 'nova/tests') diff --git a/nova/tests/network/__init__.py b/nova/tests/network/__init__.py new file mode 100644 index 000000000..e51065983 --- /dev/null +++ b/nova/tests/network/__init__.py @@ -0,0 +1,47 @@ +import os + +from nova import context +from nova import db +from nova import flags +from nova import log as logging +from nova import utils + +FLAGS = flags.FLAGS +LOG = logging.getLogger('nova.tests.network') + + +def binpath(script): + """Returns the absolute path to a script in bin""" + return os.path.abspath(os.path.join(__file__, "../../../../bin", script)) + + +def lease_ip(private_ip): + """Run add command on dhcpbridge""" + network_ref = db.fixed_ip_get_network(context.get_admin_context(), + private_ip) + instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), + private_ip) + cmd = "%s add %s %s fake" % (binpath('nova-dhcpbridge'), + instance_ref['mac_address'], + private_ip) + env = {'DNSMASQ_INTERFACE': network_ref['bridge'], + 'TESTING': '1', + 'FLAGFILE': FLAGS.dhcpbridge_flagfile} + (out, err) = utils.execute(cmd, addl_env=env) + LOG.debug("ISSUE_IP: %s, %s ", out, err) + + +def release_ip(private_ip): + """Run del command on dhcpbridge""" + network_ref = db.fixed_ip_get_network(context.get_admin_context(), + private_ip) + instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), + private_ip) + cmd = "%s del %s %s fake" % (binpath('nova-dhcpbridge'), + instance_ref['mac_address'], + private_ip) + env = {'DNSMASQ_INTERFACE': network_ref['bridge'], + 'TESTING': '1', + 'FLAGFILE': FLAGS.dhcpbridge_flagfile} + (out, err) = utils.execute(cmd, addl_env=env) + LOG.debug("RELEASE_IP: %s, %s ", out, err) diff --git a/nova/tests/network/base.py b/nova/tests/network/base.py new file mode 100644 index 000000000..2dd8178ff --- /dev/null +++ b/nova/tests/network/base.py @@ -0,0 +1,154 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +""" +Base class of Unit Tests for all network models +""" +import IPy +import os + +from nova import context +from nova import db +from nova import exception +from nova import flags +from nova import log as logging +from nova import test +from nova import utils +from nova.auth import manager + +FLAGS = flags.FLAGS +LOG = logging.getLogger('nova.tests.network') + + +class NetworkTestCase(test.TestCase): + """Test cases for network code""" + def setUp(self): + super(NetworkTestCase, self).setUp() + # NOTE(vish): if you change these flags, make sure to change the + # flags in the corresponding section in nova-dhcpbridge + self.flags(connection_type='fake', + fake_call=True, + fake_network=True) + self.manager = manager.AuthManager() + self.user = self.manager.create_user('netuser', 'netuser', 'netuser') + self.projects = [] + self.network = utils.import_object(FLAGS.network_manager) + self.context = context.RequestContext(project=None, user=self.user) + for i in range(FLAGS.num_networks): + name = 'project%s' % i + project = self.manager.create_project(name, 'netuser', name) + self.projects.append(project) + # create the necessary network data for the project + user_context = context.RequestContext(project=self.projects[i], + user=self.user) + host = self.network.get_network_host(user_context.elevated()) + instance_ref = self._create_instance(0) + self.instance_id = instance_ref['id'] + instance_ref = self._create_instance(1) + self.instance2_id = instance_ref['id'] + + def tearDown(self): + # TODO(termie): this should really be instantiating clean datastores + # in between runs, one failure kills all the tests + db.instance_destroy(context.get_admin_context(), self.instance_id) + db.instance_destroy(context.get_admin_context(), self.instance2_id) + for project in self.projects: + self.manager.delete_project(project) + self.manager.delete_user(self.user) + super(NetworkTestCase, self).tearDown() + + def _create_instance(self, project_num, mac=None): + if not mac: + mac = utils.generate_mac() + project = self.projects[project_num] + self.context._project = project + self.context.project_id = project.id + return db.instance_create(self.context, + {'project_id': project.id, + 'mac_address': mac}) + + def _create_address(self, project_num, instance_id=None): + """Create an address in given project num""" + if instance_id is None: + instance_id = self.instance_id + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + return self.network.allocate_fixed_ip(self.context, instance_id) + + def _deallocate_address(self, project_num, address): + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + self.network.deallocate_fixed_ip(self.context, address) + + def _is_allocated_in_project(self, address, project_id): + """Returns true if address is in specified project""" + project_net = db.network_get_by_bridge(context.get_admin_context(), + FLAGS.flat_network_bridge) + network = db.fixed_ip_get_network(context.get_admin_context(), + address) + instance = db.fixed_ip_get_instance(context.get_admin_context(), + address) + # instance exists until release + return instance is not None and network['id'] == project_net['id'] + + def test_private_ipv6(self): + """Make sure ipv6 is OK""" + if FLAGS.use_ipv6: + instance_ref = self._create_instance(0) + address = self._create_address(0, instance_ref['id']) + network_ref = db.project_get_network( + context.get_admin_context(), + self.context.project_id) + address_v6 = db.instance_get_fixed_address_v6( + context.get_admin_context(), + instance_ref['id']) + self.assertEqual(instance_ref['mac_address'], + utils.to_mac(address_v6)) + instance_ref2 = db.fixed_ip_get_instance_v6( + context.get_admin_context(), + address_v6) + self.assertEqual(instance_ref['id'], instance_ref2['id']) + self.assertEqual(address_v6, + utils.to_global_ipv6( + network_ref['cidr_v6'], + instance_ref['mac_address'])) + self._deallocate_address(0, address) + db.instance_destroy(context.get_admin_context(), + instance_ref['id']) + + def test_available_ips(self): + """Make sure the number of available ips for the network is correct + + The number of available IP addresses depends on the test + environment's setup. + + Network size is set in test fixture's setUp method. + + There are ips reserved at the bottom and top of the range. + services (network, gateway, CloudPipe, broadcast) + """ + network = db.project_get_network(context.get_admin_context(), + self.projects[0].id) + net_size = flags.FLAGS.network_size + admin_context = context.get_admin_context() + total_ips = (db.network_count_available_ips(admin_context, + network['id']) + + db.network_count_reserved_ips(admin_context, + network['id']) + + db.network_count_allocated_ips(admin_context, + network['id'])) + self.assertEqual(total_ips, net_size) diff --git a/nova/tests/test_flat_network.py b/nova/tests/test_flat_network.py index 91a49920d..a9a934158 100644 --- a/nova/tests/test_flat_network.py +++ b/nova/tests/test_flat_network.py @@ -30,96 +30,15 @@ from nova import log as logging from nova import test from nova import utils from nova.auth import manager +from nova.tests.network import base + FLAGS = flags.FLAGS LOG = logging.getLogger('nova.tests.network') -class FlatNetworkTestCase(test.TestCase): +class FlatNetworkTestCase(base.NetworkTestCase): """Test cases for network code""" - def setUp(self): - super(FlatNetworkTestCase, self).setUp() - # NOTE(vish): if you change these flags, make sure to change the - # flags in the corresponding section in nova-dhcpbridge - self.flags(connection_type='fake', - fake_call=True, - fake_network=True) - self.manager = manager.AuthManager() - self.user = self.manager.create_user('netuser', 'netuser', 'netuser') - self.projects = [] - self.network = utils.import_object(FLAGS.network_manager) - self.context = context.RequestContext(project=None, user=self.user) - for i in range(5): - name = 'project%s' % i - project = self.manager.create_project(name, 'netuser', name) - self.projects.append(project) - # create the necessary network data for the project - user_context = context.RequestContext(project=self.projects[i], - user=self.user) - host = self.network.get_network_host(user_context.elevated()) - instance_ref = self._create_instance(0) - self.instance_id = instance_ref['id'] - instance_ref = self._create_instance(1) - self.instance2_id = instance_ref['id'] - - def tearDown(self): - # TODO(termie): this should really be instantiating clean datastores - # in between runs, one failure kills all the tests - db.instance_destroy(context.get_admin_context(), self.instance_id) - db.instance_destroy(context.get_admin_context(), self.instance2_id) - for project in self.projects: - self.manager.delete_project(project) - self.manager.delete_user(self.user) - super(FlatNetworkTestCase, self).tearDown() - - def _create_instance(self, project_num, mac=None): - if not mac: - mac = utils.generate_mac() - project = self.projects[project_num] - self.context._project = project - self.context.project_id = project.id - return db.instance_create(self.context, - {'project_id': project.id, - 'mac_address': mac}) - - def _create_address(self, project_num, instance_id=None): - """Create an address in given project num""" - if instance_id is None: - instance_id = self.instance_id - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - return self.network.allocate_fixed_ip(self.context, instance_id) - - def _deallocate_address(self, project_num, address): - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - self.network.deallocate_fixed_ip(self.context, address) - - def test_private_ipv6(self): - """Make sure ipv6 is OK""" - if FLAGS.use_ipv6: - instance_ref = self._create_instance(0) - address = self._create_address(0, instance_ref['id']) - network_ref = db.project_get_network( - context.get_admin_context(), - self.context.project_id) - address_v6 = db.instance_get_fixed_address_v6( - context.get_admin_context(), - instance_ref['id']) - self.assertEqual(instance_ref['mac_address'], - utils.to_mac(address_v6)) - instance_ref2 = db.fixed_ip_get_instance_v6( - context.get_admin_context(), - address_v6) - self.assertEqual(instance_ref['id'], instance_ref2['id']) - self.assertEqual(address_v6, - utils.to_global_ipv6( - network_ref['cidr_v6'], - instance_ref['mac_address'])) - self._deallocate_address(0, address) - db.instance_destroy(context.get_admin_context(), - instance_ref['id']) - def test_public_network_association(self): """Makes sure that we can allocate a public ip""" # TODO(vish): better way of adding floating ips @@ -167,28 +86,34 @@ class FlatNetworkTestCase(test.TestCase): def test_allocate_deallocate_fixed_ip(self): """Makes sure that we can allocate and deallocate a fixed ip""" address = self._create_address(0) - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + self.assertTrue(self._is_allocated_in_project(address, + self.projects[0].id)) self._deallocate_address(0, address) # check if the fixed ip address is really deallocated - self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + self.assertFalse(self._is_allocated_in_project(address, + self.projects[0].id)) def test_side_effects(self): """Ensures allocating and releasing has no side effects""" address = self._create_address(0) address2 = self._create_address(1, self.instance2_id) - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) - self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) + self.assertTrue(self._is_allocated_in_project(address, + self.projects[0].id)) + self.assertTrue(self._is_allocated_in_project(address2, + self.projects[1].id)) self._deallocate_address(0, address) - self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + self.assertFalse(self._is_allocated_in_project(address, + self.projects[0].id)) # First address release shouldn't affect the second - self.assertTrue(is_allocated_in_project(address2, self.projects[0].id)) + self.assertTrue(self._is_allocated_in_project(address2, + self.projects[0].id)) self._deallocate_address(1, address2) - self.assertFalse(is_allocated_in_project(address2, + self.assertFalse(self._is_allocated_in_project(address2, self.projects[1].id)) def test_ips_are_reused(self): @@ -201,29 +126,6 @@ class FlatNetworkTestCase(test.TestCase): self.network.deallocate_fixed_ip(self.context, address2) - def test_available_ips(self): - """Make sure the number of available ips for the network is correct - - The number of available IP addresses depends on the test - environment's setup. - - Network size is set in test fixture's setUp method. - - There are ips reserved at the bottom and top of the range. - services (network, gateway, CloudPipe, broadcast) - """ - network = db.project_get_network(context.get_admin_context(), - self.projects[0].id) - net_size = flags.FLAGS.network_size - admin_context = context.get_admin_context() - total_ips = (db.network_count_available_ips(admin_context, - network['id']) + - db.network_count_reserved_ips(admin_context, - network['id']) + - db.network_count_allocated_ips(admin_context, - network['id'])) - self.assertEqual(total_ips, net_size) - def test_too_many_addresses(self): """Test for a NoMoreAddresses exception when all fixed ips are used. """ @@ -257,20 +159,3 @@ class FlatNetworkTestCase(test.TestCase): def run(self, result=None): if(FLAGS.network_manager == 'nova.network.manager.FlatManager'): super(FlatNetworkTestCase, self).run(result) - - -def is_allocated_in_project(address, project_id): - """Returns true if address is in specified project""" - #project_net = db.project_get_network(context.get_admin_context(), - # project_id) - project_net = db.network_get_by_bridge(context.get_admin_context(), - FLAGS.flat_network_bridge) - network = db.fixed_ip_get_network(context.get_admin_context(), address) - instance = db.fixed_ip_get_instance(context.get_admin_context(), address) - # instance exists until release - return instance is not None and network['id'] == project_net['id'] - - -def binpath(script): - """Returns the absolute path to a script in bin""" - return os.path.abspath(os.path.join(__file__, "../../../bin", script)) diff --git a/nova/tests/test_vlan_network.py b/nova/tests/test_vlan_network.py index be39313bb..f3478320b 100644 --- a/nova/tests/test_vlan_network.py +++ b/nova/tests/test_vlan_network.py @@ -29,100 +29,16 @@ from nova import log as logging from nova import test from nova import utils from nova.auth import manager +from nova.tests.network import base +from nova.tests.network import binpath,\ + lease_ip, release_ip FLAGS = flags.FLAGS LOG = logging.getLogger('nova.tests.network') -class VlanNetworkTestCase(test.TestCase): +class VlanNetworkTestCase(base.NetworkTestCase): """Test cases for network code""" - def setUp(self): - super(VlanNetworkTestCase, self).setUp() - # NOTE(vish): if you change these flags, make sure to change the - # flags in the corresponding section in nova-dhcpbridge - self.flags(connection_type='fake', - fake_call=True, - fake_network=True) - self.manager = manager.AuthManager() - self.user = self.manager.create_user('netuser', 'netuser', 'netuser') - self.projects = [] - self.network = utils.import_object(FLAGS.network_manager) - self.context = context.RequestContext(project=None, user=self.user) - for i in range(FLAGS.num_networks): - name = 'project%s' % i - project = self.manager.create_project(name, 'netuser', name) - self.projects.append(project) - # create the necessary network data for the project - user_context = context.RequestContext(project=self.projects[i], - user=self.user) - host = self.network.get_network_host(user_context.elevated()) - instance_ref = self._create_instance(0) - self.instance_id = instance_ref['id'] - instance_ref = self._create_instance(1) - self.instance2_id = instance_ref['id'] - - def tearDown(self): - # TODO(termie): this should really be instantiating clean datastores - # in between runs, one failure kills all the tests - db.instance_destroy(context.get_admin_context(), self.instance_id) - db.instance_destroy(context.get_admin_context(), self.instance2_id) - for project in self.projects: - self.manager.delete_project(project) - self.manager.delete_user(self.user) - super(VlanNetworkTestCase, self).tearDown() - - def run(self, result=None): - if(FLAGS.network_manager == 'nova.network.manager.VlanManager'): - super(VlanNetworkTestCase, self).run(result) - - def _create_instance(self, project_num, mac=None): - if not mac: - mac = utils.generate_mac() - project = self.projects[project_num] - self.context._project = project - self.context.project_id = project.id - return db.instance_create(self.context, - {'project_id': project.id, - 'mac_address': mac}) - - def _create_address(self, project_num, instance_id=None): - """Create an address in given project num""" - if instance_id is None: - instance_id = self.instance_id - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - return self.network.allocate_fixed_ip(self.context, instance_id) - - def _deallocate_address(self, project_num, address): - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - self.network.deallocate_fixed_ip(self.context, address) - - def test_private_ipv6(self): - """Make sure ipv6 is OK""" - if FLAGS.use_ipv6: - instance_ref = self._create_instance(0) - address = self._create_address(0, instance_ref['id']) - network_ref = db.project_get_network( - context.get_admin_context(), - self.context.project_id) - address_v6 = db.instance_get_fixed_address_v6( - context.get_admin_context(), - instance_ref['id']) - self.assertEqual(instance_ref['mac_address'], - utils.to_mac(address_v6)) - instance_ref2 = db.fixed_ip_get_instance_v6( - context.get_admin_context(), - address_v6) - self.assertEqual(instance_ref['id'], instance_ref2['id']) - self.assertEqual(address_v6, - utils.to_global_ipv6( - network_ref['cidr_v6'], - instance_ref['mac_address'])) - self._deallocate_address(0, address) - db.instance_destroy(context.get_admin_context(), - instance_ref['id']) - def test_public_network_association(self): """Makes sure that we can allocaate a public ip""" # TODO(vish): better way of adding floating ips @@ -157,24 +73,30 @@ class VlanNetworkTestCase(test.TestCase): def test_allocate_deallocate_fixed_ip(self): """Makes sure that we can allocate and deallocate a fixed ip""" address = self._create_address(0) - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + self.assertTrue(self._is_allocated_in_project(address, + self.projects[0].id)) lease_ip(address) self._deallocate_address(0, address) # Doesn't go away until it's dhcp released - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) + self.assertTrue(self._is_allocated_in_project(address, + self.projects[0].id)) release_ip(address) - self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + self.assertFalse(self._is_allocated_in_project(address, + self.projects[0].id)) def test_side_effects(self): """Ensures allocating and releasing has no side effects""" address = self._create_address(0) address2 = self._create_address(1, self.instance2_id) - self.assertTrue(is_allocated_in_project(address, self.projects[0].id)) - self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) - self.assertFalse(is_allocated_in_project(address, self.projects[1].id)) + self.assertTrue(self._is_allocated_in_project(address, + self.projects[0].id)) + self.assertTrue(self._is_allocated_in_project(address2, + self.projects[1].id)) + self.assertFalse(self._is_allocated_in_project(address, + self.projects[1].id)) # Addresses are allocated before they're issued lease_ip(address) @@ -182,14 +104,16 @@ class VlanNetworkTestCase(test.TestCase): self._deallocate_address(0, address) release_ip(address) - self.assertFalse(is_allocated_in_project(address, self.projects[0].id)) + self.assertFalse(self._is_allocated_in_project(address, + self.projects[0].id)) # First address release shouldn't affect the second - self.assertTrue(is_allocated_in_project(address2, self.projects[1].id)) + self.assertTrue(self._is_allocated_in_project(address2, + self.projects[1].id)) self._deallocate_address(1, address2) release_ip(address2) - self.assertFalse(is_allocated_in_project(address2, + self.assertFalse(self._is_allocated_in_project(address2, self.projects[1].id)) def test_subnet_edge(self): @@ -212,11 +136,11 @@ class VlanNetworkTestCase(test.TestCase): lease_ip(address3) self.context._project = self.projects[i] self.context.project_id = self.projects[i].id - self.assertFalse(is_allocated_in_project(address, + self.assertFalse(self._is_allocated_in_project(address, self.projects[0].id)) - self.assertFalse(is_allocated_in_project(address2, + self.assertFalse(self._is_allocated_in_project(address2, self.projects[0].id)) - self.assertFalse(is_allocated_in_project(address3, + self.assertFalse(self._is_allocated_in_project(address3, self.projects[0].id)) self.network.deallocate_fixed_ip(self.context, address) self.network.deallocate_fixed_ip(self.context, address2) @@ -270,29 +194,6 @@ class VlanNetworkTestCase(test.TestCase): self.network.deallocate_fixed_ip(self.context, address2) release_ip(address) - def test_available_ips(self): - """Make sure the number of available ips for the network is correct - - The number of available IP addresses depends on the test - environment's setup. - - Network size is set in test fixture's setUp method. - - There are ips reserved at the bottom and top of the range. - services (network, gateway, CloudPipe, broadcast) - """ - network = db.project_get_network(context.get_admin_context(), - self.projects[0].id) - net_size = flags.FLAGS.network_size - admin_context = context.get_admin_context() - total_ips = (db.network_count_available_ips(admin_context, - network['id']) + - db.network_count_reserved_ips(admin_context, - network['id']) + - db.network_count_allocated_ips(admin_context, - network['id'])) - self.assertEqual(total_ips, net_size) - def test_too_many_addresses(self): """Test for a NoMoreAddresses exception when all fixed ips are used. """ @@ -325,49 +226,17 @@ class VlanNetworkTestCase(test.TestCase): network['id']) self.assertEqual(ip_count, num_available_ips) + def _is_allocated_in_project(self, address, project_id): + """Returns true if address is in specified project""" + project_net = db.project_get_network(context.get_admin_context(), + project_id) + network = db.fixed_ip_get_network(context.get_admin_context(), + address) + instance = db.fixed_ip_get_instance(context.get_admin_context(), + address) + # instance exists until release + return instance is not None and network['id'] == project_net['id'] -def is_allocated_in_project(address, project_id): - """Returns true if address is in specified project""" - project_net = db.project_get_network(context.get_admin_context(), - project_id) - network = db.fixed_ip_get_network(context.get_admin_context(), address) - instance = db.fixed_ip_get_instance(context.get_admin_context(), address) - # instance exists until release - return instance is not None and network['id'] == project_net['id'] - - -def binpath(script): - """Returns the absolute path to a script in bin""" - return os.path.abspath(os.path.join(__file__, "../../../bin", script)) - - -def lease_ip(private_ip): - """Run add command on dhcpbridge""" - network_ref = db.fixed_ip_get_network(context.get_admin_context(), - private_ip) - instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), - private_ip) - cmd = "%s add %s %s fake" % (binpath('nova-dhcpbridge'), - instance_ref['mac_address'], - private_ip) - env = {'DNSMASQ_INTERFACE': network_ref['bridge'], - 'TESTING': '1', - 'FLAGFILE': FLAGS.dhcpbridge_flagfile} - (out, err) = utils.execute(cmd, addl_env=env) - LOG.debug("ISSUE_IP: %s, %s ", out, err) - - -def release_ip(private_ip): - """Run del command on dhcpbridge""" - network_ref = db.fixed_ip_get_network(context.get_admin_context(), - private_ip) - instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), - private_ip) - cmd = "%s del %s %s fake" % (binpath('nova-dhcpbridge'), - instance_ref['mac_address'], - private_ip) - env = {'DNSMASQ_INTERFACE': network_ref['bridge'], - 'TESTING': '1', - 'FLAGFILE': FLAGS.dhcpbridge_flagfile} - (out, err) = utils.execute(cmd, addl_env=env) - LOG.debug("RELEASE_IP: %s, %s ", out, err) + def run(self, result=None): + if(FLAGS.network_manager == 'nova.network.manager.VlanManager'): + super(VlanNetworkTestCase, self).run(result) -- cgit From 998975651ac2f2df7a3f8af16d62d197f451180f Mon Sep 17 00:00:00 2001 From: Justin Santa Barbara Date: Thu, 10 Mar 2011 13:53:27 -0800 Subject: Test login. Uncovered bug732866 --- nova/tests/integrated/integrated_helpers.py | 182 ++++++++++++++++++++++++++++ nova/tests/integrated/test_login.py | 77 ++++++++++++ 2 files changed, 259 insertions(+) create mode 100644 nova/tests/integrated/integrated_helpers.py create mode 100644 nova/tests/integrated/test_login.py (limited to 'nova/tests') diff --git a/nova/tests/integrated/integrated_helpers.py b/nova/tests/integrated/integrated_helpers.py new file mode 100644 index 000000000..691ead6e1 --- /dev/null +++ b/nova/tests/integrated/integrated_helpers.py @@ -0,0 +1,182 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 Justin Santa Barbara +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +""" +Provides common functionality for integrated unit tests +""" + +import random +import string + +from nova import exception +from nova import flags +from nova import service +from nova import test # For the flags +from nova.auth import manager +from nova.exception import Error +from nova.log import logging +from nova.tests.integrated.api import client + + +FLAGS = flags.FLAGS + +LOG = logging.getLogger('nova.tests.integrated') + + +def generate_random_alphanumeric(length): + """Creates a random alphanumeric string of specified length""" + return ''.join(random.choice(string.ascii_uppercase + string.digits) + for _x in range(length)) + + +def generate_random_numeric(length): + """Creates a random numeric string of specified length""" + return ''.join(random.choice(string.digits) + for _x in range(length)) + + +def generate_new_element(items, prefix, numeric=False): + """Creates a random string with prefix, that is not in 'items' list""" + while True: + if numeric: + candidate = prefix + generate_random_numeric(8) + else: + candidate = prefix + generate_random_alphanumeric(8) + if not candidate in items: + return candidate + print "Random collision on %s" % candidate + + +class TestUser(object): + def __init__(self, name, secret, auth_url): + self.name = name + self.secret = secret + self.auth_url = auth_url + + if not auth_url: + raise exception.Error("auth_url is required") + self.openstack_api = client.TestOpenStackClient(self.name, + self.secret, + self.auth_url) + + +class IntegratedUnitTestContext(object): + __INSTANCE = None + + def __init__(self): + self.auth_manager = manager.AuthManager() + + self.wsgi_server = None + self.wsgi_apps = [] + self.api_service = None + + self.services = [] + self.auth_url = None + self.project_name = None + + self.setup() + + def setup(self): + self._start_services() + + self._create_test_user() + + def _create_test_user(self): + self.test_user = self._create_unittest_user() + + # No way to currently pass this through the OpenStack API + self.project_name = 'openstack' + self._configure_project(self.project_name, self.test_user) + + def _start_services(self): + # WSGI shutdown broken :-( + if not self.api_service: + self._start_api_service() + + def cleanup(self): + for service in self.services: + service.kill() + self.services = [] + # TODO(justinsb): Shutdown WSGI & anything else we startup + # WSGI shutdown broken :-( + # self.wsgi_server.terminate() + # self.wsgi_server = None + self.test_user = None + + def _create_unittest_user(self): + users = self.auth_manager.get_users() + user_names = [user.name for user in users] + auth_name = generate_new_element(user_names, 'unittest_user_') + auth_key = generate_random_alphanumeric(16) + + # Right now there's a bug where auth_name and auth_key are reversed + auth_key = auth_name + + self.auth_manager.create_user(auth_name, auth_name, auth_key, False) + return TestUser(auth_name, auth_key, self.auth_url) + + def _configure_project(self, project_name, user): + projects = self.auth_manager.get_projects() + project_names = [project.name for project in projects] + if not project_name in project_names: + project = self.auth_manager.create_project(project_name, + user.name, + description=None, + member_users=None) + else: + self.auth_manager.add_to_project(user.name, project_name) + + def _start_api_service(self): + api_service = service.ApiService.create() + api_service.start() + + if not api_service: + raise Exception("API Service was None") + + # WSGI shutdown broken :-( + #self.services.append(volume_service) + self.api_service = api_service + + self.auth_url = 'http://localhost:8774/v1.0' + + return api_service + + # WSGI shutdown broken :-( + #@staticmethod + #def get(): + # if not IntegratedUnitTestContext.__INSTANCE: + # IntegratedUnitTestContext.startup() + # #raise Error("Must call IntegratedUnitTestContext::startup") + # return IntegratedUnitTestContext.__INSTANCE + + @staticmethod + def startup(): + # Because WSGI shutdown is broken at the moment, we have to recycle + if IntegratedUnitTestContext.__INSTANCE: + #raise Error("Multiple calls to IntegratedUnitTestContext.startup") + IntegratedUnitTestContext.__INSTANCE.setup() + else: + IntegratedUnitTestContext.__INSTANCE = IntegratedUnitTestContext() + return IntegratedUnitTestContext.__INSTANCE + + @staticmethod + def shutdown(): + if not IntegratedUnitTestContext.__INSTANCE: + raise Error("Must call IntegratedUnitTestContext::startup") + IntegratedUnitTestContext.__INSTANCE.cleanup() + # WSGI shutdown broken :-( + #IntegratedUnitTestContext.__INSTANCE = None diff --git a/nova/tests/integrated/test_login.py b/nova/tests/integrated/test_login.py new file mode 100644 index 000000000..990dcaaf4 --- /dev/null +++ b/nova/tests/integrated/test_login.py @@ -0,0 +1,77 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2011 Justin Santa Barbara +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. + +import unittest + +from nova import flags +from nova.log import logging +from nova.tests.integrated import integrated_helpers +from nova.tests.integrated.api import client + +LOG = logging.getLogger('nova.tests.integrated') + +FLAGS = flags.FLAGS +FLAGS.verbose = True + + +class LoginTest(unittest.TestCase): + def setUp(self): + super(LoginTest, self).setUp() + context = integrated_helpers.IntegratedUnitTestContext.startup() + self.user = context.test_user + self.api = self.user.openstack_api + + def tearDown(self): + integrated_helpers.IntegratedUnitTestContext.shutdown() + super(LoginTest, self).tearDown() + + def test_login(self): + """Simple check - we list flavors - so we know we're logged in""" + flavors = self.api.get_flavors() + for flavor in flavors: + LOG.debug(_("flavor: %s") % flavor) + + def test_bad_login_password(self): + """Test that I get a 401 with a bad username""" + bad_credentials_api = client.TestOpenStackClient(self.user.name, + "notso_password", + self.user.auth_url) + + self.assertRaises(client.OpenstackApiAuthenticationException, + bad_credentials_api.get_flavors) + + def test_bad_login_username(self): + """Test that I get a 401 with a bad password""" + bad_credentials_api = client.TestOpenStackClient("notso_username", + self.user.secret, + self.user.auth_url) + + self.assertRaises(client.OpenstackApiAuthenticationException, + bad_credentials_api.get_flavors) + + + def test_bad_login_both_bad(self): + """Test that I get a 401 with both bad username and bad password""" + bad_credentials_api = client.TestOpenStackClient("notso_username", + "notso_password", + self.user.auth_url) + + self.assertRaises(client.OpenstackApiAuthenticationException, + bad_credentials_api.get_flavors) + +if __name__ == "__main__": + unittest.main() -- cgit From 3d6430ecd114daa21c72c3d215daaa94f0e87e62 Mon Sep 17 00:00:00 2001 From: Justin Santa Barbara Date: Thu, 10 Mar 2011 14:12:41 -0800 Subject: Re-removed the code that was deleted upstream but somehow didn't get merged in. Bizarre! --- nova/tests/integrated/api/client.py | 21 --------------------- 1 file changed, 21 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/integrated/api/client.py b/nova/tests/integrated/api/client.py index 0ce480ae7..da8d87e07 100644 --- a/nova/tests/integrated/api/client.py +++ b/nova/tests/integrated/api/client.py @@ -172,27 +172,6 @@ class TestOpenStackClient(object): kwargs.setdefault('check_response_status', [200, 202]) return self.api_request(relative_uri, **kwargs) - def get_keys_detail(self): - return self.api_get('/keys/detail')['keys'] - - def post_key(self, key): - return self.api_post('/keys', key)['key'] - - def delete_key(self, key_id): - return self.api_delete('/keys/%s' % key_id) - - def get_volume(self, volume_id): - return self.api_get('/volumes/%s' % volume_id)['volume'] - - def get_volumes_detail(self): - return self.api_get('/volumes/detail')['volumes'] - - def post_volume(self, volume): - return self.api_post('/volumes', volume)['volume'] - - def delete_volume(self, volume_id): - return self.api_delete('/volumes/%s' % volume_id) - def get_server(self, server_id): return self.api_get('/servers/%s' % server_id)['server'] -- cgit From 29bc4f5074ca3ada98a25a745077b998b4c5509c Mon Sep 17 00:00:00 2001 From: Justin Santa Barbara Date: Thu, 10 Mar 2011 14:14:01 -0800 Subject: Pep8 / Style --- nova/tests/integrated/test_login.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/integrated/test_login.py b/nova/tests/integrated/test_login.py index 990dcaaf4..e362f92d6 100644 --- a/nova/tests/integrated/test_login.py +++ b/nova/tests/integrated/test_login.py @@ -22,6 +22,7 @@ from nova.log import logging from nova.tests.integrated import integrated_helpers from nova.tests.integrated.api import client + LOG = logging.getLogger('nova.tests.integrated') FLAGS = flags.FLAGS @@ -63,7 +64,6 @@ class LoginTest(unittest.TestCase): self.assertRaises(client.OpenstackApiAuthenticationException, bad_credentials_api.get_flavors) - def test_bad_login_both_bad(self): """Test that I get a 401 with both bad username and bad password""" bad_credentials_api = client.TestOpenStackClient("notso_username", -- cgit From 0d3e950ed4b0c8abbd619d4ac8724b4c3ce45bf1 Mon Sep 17 00:00:00 2001 From: Justin Santa Barbara Date: Thu, 10 Mar 2011 14:21:36 -0800 Subject: Document known bug numbers by the code which is degraded until the bugs are fixed --- nova/tests/integrated/api/client.py | 1 + nova/tests/integrated/integrated_helpers.py | 6 ++++++ 2 files changed, 7 insertions(+) (limited to 'nova/tests') diff --git a/nova/tests/integrated/api/client.py b/nova/tests/integrated/api/client.py index da8d87e07..6fba2930a 100644 --- a/nova/tests/integrated/api/client.py +++ b/nova/tests/integrated/api/client.py @@ -109,6 +109,7 @@ class TestOpenStackClient(object): LOG.debug(_("%(auth_uri)s => code %(http_status)s") % locals()) # Until bug732866 is fixed, we can't check this properly... + # bug732866 #if http_status == 401: if http_status != 204: raise OpenstackApiAuthenticationException(response=response) diff --git a/nova/tests/integrated/integrated_helpers.py b/nova/tests/integrated/integrated_helpers.py index 691ead6e1..47093636e 100644 --- a/nova/tests/integrated/integrated_helpers.py +++ b/nova/tests/integrated/integrated_helpers.py @@ -104,6 +104,7 @@ class IntegratedUnitTestContext(object): def _start_services(self): # WSGI shutdown broken :-( + # bug731668 if not self.api_service: self._start_api_service() @@ -112,6 +113,7 @@ class IntegratedUnitTestContext(object): service.kill() self.services = [] # TODO(justinsb): Shutdown WSGI & anything else we startup + # bug731668 # WSGI shutdown broken :-( # self.wsgi_server.terminate() # self.wsgi_server = None @@ -124,6 +126,7 @@ class IntegratedUnitTestContext(object): auth_key = generate_random_alphanumeric(16) # Right now there's a bug where auth_name and auth_key are reversed + # bug732907 auth_key = auth_name self.auth_manager.create_user(auth_name, auth_name, auth_key, False) @@ -156,6 +159,7 @@ class IntegratedUnitTestContext(object): return api_service # WSGI shutdown broken :-( + # bug731668 #@staticmethod #def get(): # if not IntegratedUnitTestContext.__INSTANCE: @@ -166,6 +170,7 @@ class IntegratedUnitTestContext(object): @staticmethod def startup(): # Because WSGI shutdown is broken at the moment, we have to recycle + # bug731668 if IntegratedUnitTestContext.__INSTANCE: #raise Error("Multiple calls to IntegratedUnitTestContext.startup") IntegratedUnitTestContext.__INSTANCE.setup() @@ -179,4 +184,5 @@ class IntegratedUnitTestContext(object): raise Error("Must call IntegratedUnitTestContext::startup") IntegratedUnitTestContext.__INSTANCE.cleanup() # WSGI shutdown broken :-( + # bug731668 #IntegratedUnitTestContext.__INSTANCE = None -- cgit From 2b20306fcaddcb6b9bc57fb55b17230d709cd1ce Mon Sep 17 00:00:00 2001 From: Justin Santa Barbara Date: Mon, 14 Mar 2011 22:23:38 -0700 Subject: Derive unit test from standard nova.test.TestCase --- nova/tests/integrated/test_login.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) (limited to 'nova/tests') diff --git a/nova/tests/integrated/test_login.py b/nova/tests/integrated/test_login.py index e362f92d6..5fa558bdf 100644 --- a/nova/tests/integrated/test_login.py +++ b/nova/tests/integrated/test_login.py @@ -18,6 +18,7 @@ import unittest from nova import flags +from nova import test from nova.log import logging from nova.tests.integrated import integrated_helpers from nova.tests.integrated.api import client @@ -29,7 +30,7 @@ FLAGS = flags.FLAGS FLAGS.verbose = True -class LoginTest(unittest.TestCase): +class LoginTest(test.TestCase): def setUp(self): super(LoginTest, self).setUp() context = integrated_helpers.IntegratedUnitTestContext.startup() @@ -73,5 +74,6 @@ class LoginTest(unittest.TestCase): self.assertRaises(client.OpenstackApiAuthenticationException, bad_credentials_api.get_flavors) + if __name__ == "__main__": unittest.main() -- cgit From db8beffc9acd90c748512c1fa9c127d39756232c Mon Sep 17 00:00:00 2001 From: Justin Santa Barbara Date: Mon, 14 Mar 2011 22:36:30 -0700 Subject: Reapplied rename of Openstack -> OpenStack. Easier to do it by hand than to ask Bazaar to do it. --- nova/tests/integrated/api/client.py | 20 ++++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/integrated/api/client.py b/nova/tests/integrated/api/client.py index 6fba2930a..568e8c17e 100644 --- a/nova/tests/integrated/api/client.py +++ b/nova/tests/integrated/api/client.py @@ -24,7 +24,7 @@ from nova import log as logging LOG = logging.getLogger('nova.tests.api') -class OpenstackApiException(Exception): +class OpenStackApiException(Exception): def __init__(self, message=None, response=None): self.response = response if not message: @@ -37,22 +37,22 @@ class OpenstackApiException(Exception): message = _('%(message)s\nStatus Code: %(_status)s\n' 'Body: %(_body)s') % locals() - super(OpenstackApiException, self).__init__(message) + super(OpenStackApiException, self).__init__(message) -class OpenstackApiAuthenticationException(OpenstackApiException): +class OpenStackApiAuthenticationException(OpenStackApiException): def __init__(self, response=None, message=None): if not message: message = _("Authentication error") - super(OpenstackApiAuthenticationException, self).__init__(message, + super(OpenStackApiAuthenticationException, self).__init__(message, response) -class OpenstackApiNotFoundException(OpenstackApiException): +class OpenStackApiNotFoundException(OpenStackApiException): def __init__(self, response=None, message=None): if not message: message = _("Item not found") - super(OpenstackApiNotFoundException, self).__init__(message, response) + super(OpenStackApiNotFoundException, self).__init__(message, response) class TestOpenStackClient(object): @@ -82,7 +82,7 @@ class TestOpenStackClient(object): conn = httplib.HTTPSConnection(hostname, port=port) else: - raise OpenstackApiException("Unknown scheme: %s" % url) + raise OpenStackApiException("Unknown scheme: %s" % url) relative_url = parsed_url.path if parsed_url.query: @@ -112,7 +112,7 @@ class TestOpenStackClient(object): # bug732866 #if http_status == 401: if http_status != 204: - raise OpenstackApiAuthenticationException(response=response) + raise OpenStackApiAuthenticationException(response=response) auth_headers = {} for k, v in response.getheaders(): @@ -139,9 +139,9 @@ class TestOpenStackClient(object): if check_response_status: if not http_status in check_response_status: if http_status == 404: - raise OpenstackApiNotFoundException(response=response) + raise OpenStackApiNotFoundException(response=response) else: - raise OpenstackApiException( + raise OpenStackApiException( message=_("Unexpected status code"), response=response) -- cgit From e0563f49792441af106c52e662bdada3c7997feb Mon Sep 17 00:00:00 2001 From: Justin Santa Barbara Date: Mon, 14 Mar 2011 22:43:21 -0700 Subject: Reapplied rename to another file. --- nova/tests/integrated/test_login.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/integrated/test_login.py b/nova/tests/integrated/test_login.py index 5fa558bdf..501f8c919 100644 --- a/nova/tests/integrated/test_login.py +++ b/nova/tests/integrated/test_login.py @@ -53,7 +53,7 @@ class LoginTest(test.TestCase): "notso_password", self.user.auth_url) - self.assertRaises(client.OpenstackApiAuthenticationException, + self.assertRaises(client.OpenStackApiAuthenticationException, bad_credentials_api.get_flavors) def test_bad_login_username(self): @@ -62,7 +62,7 @@ class LoginTest(test.TestCase): self.user.secret, self.user.auth_url) - self.assertRaises(client.OpenstackApiAuthenticationException, + self.assertRaises(client.OpenStackApiAuthenticationException, bad_credentials_api.get_flavors) def test_bad_login_both_bad(self): @@ -71,7 +71,7 @@ class LoginTest(test.TestCase): "notso_password", self.user.auth_url) - self.assertRaises(client.OpenstackApiAuthenticationException, + self.assertRaises(client.OpenStackApiAuthenticationException, bad_credentials_api.get_flavors) -- cgit From f1acc3d199a1a92b531a3e74ed54a8b2fcdb999c Mon Sep 17 00:00:00 2001 From: Justin Santa Barbara Date: Tue, 15 Mar 2011 13:52:03 -0700 Subject: Now that the fix for 732866, stop working around the bug --- nova/tests/integrated/api/client.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/integrated/api/client.py b/nova/tests/integrated/api/client.py index 568e8c17e..fc7c344e7 100644 --- a/nova/tests/integrated/api/client.py +++ b/nova/tests/integrated/api/client.py @@ -108,10 +108,7 @@ class TestOpenStackClient(object): http_status = response.status LOG.debug(_("%(auth_uri)s => code %(http_status)s") % locals()) - # Until bug732866 is fixed, we can't check this properly... - # bug732866 - #if http_status == 401: - if http_status != 204: + if http_status == 401: raise OpenStackApiAuthenticationException(response=response) auth_headers = {} -- cgit From d95187aaf144cb40558f48d584a6bb8e07c6937d Mon Sep 17 00:00:00 2001 From: Tushar Patil Date: Wed, 16 Mar 2011 14:13:57 -0700 Subject: converted new lines from CRLF to LF --- nova/tests/network/__init__.py | 94 ++++++------ nova/tests/network/base.py | 308 +++++++++++++++++++------------------- nova/tests/test_flat_network.py | 322 ++++++++++++++++++++-------------------- 3 files changed, 362 insertions(+), 362 deletions(-) (limited to 'nova/tests') diff --git a/nova/tests/network/__init__.py b/nova/tests/network/__init__.py index 8f71a30ba..e0d479f8c 100644 --- a/nova/tests/network/__init__.py +++ b/nova/tests/network/__init__.py @@ -1,47 +1,47 @@ -import os - -from nova import context -from nova import db -from nova import flags -from nova import log as logging -from nova import utils - -FLAGS = flags.FLAGS -LOG = logging.getLogger('nova.tests.network') - - -def binpath(script): - """Returns the absolute path to a script in bin""" - return os.path.abspath(os.path.join(__file__, "../../../../bin", script)) - - -def lease_ip(private_ip): - """Run add command on dhcpbridge""" - network_ref = db.fixed_ip_get_network(context.get_admin_context(), - private_ip) - instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), - private_ip) - cmd = (binpath('nova-dhcpbridge'), 'add', - instance_ref['mac_address'], - private_ip, 'fake') - env = {'DNSMASQ_INTERFACE': network_ref['bridge'], - 'TESTING': '1', - 'FLAGFILE': FLAGS.dhcpbridge_flagfile} - (out, err) = utils.execute(*cmd, addl_env=env) - LOG.debug("ISSUE_IP: %s, %s ", out, err) - - -def release_ip(private_ip): - """Run del command on dhcpbridge""" - network_ref = db.fixed_ip_get_network(context.get_admin_context(), - private_ip) - instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), - private_ip) - cmd = (binpath('nova-dhcpbridge'), 'del', - instance_ref['mac_address'], - private_ip, 'fake') - env = {'DNSMASQ_INTERFACE': network_ref['bridge'], - 'TESTING': '1', - 'FLAGFILE': FLAGS.dhcpbridge_flagfile} - (out, err) = utils.execute(*cmd, addl_env=env) - LOG.debug("RELEASE_IP: %s, %s ", out, err) +import os + +from nova import context +from nova import db +from nova import flags +from nova import log as logging +from nova import utils + +FLAGS = flags.FLAGS +LOG = logging.getLogger('nova.tests.network') + + +def binpath(script): + """Returns the absolute path to a script in bin""" + return os.path.abspath(os.path.join(__file__, "../../../../bin", script)) + + +def lease_ip(private_ip): + """Run add command on dhcpbridge""" + network_ref = db.fixed_ip_get_network(context.get_admin_context(), + private_ip) + instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), + private_ip) + cmd = (binpath('nova-dhcpbridge'), 'add', + instance_ref['mac_address'], + private_ip, 'fake') + env = {'DNSMASQ_INTERFACE': network_ref['bridge'], + 'TESTING': '1', + 'FLAGFILE': FLAGS.dhcpbridge_flagfile} + (out, err) = utils.execute(*cmd, addl_env=env) + LOG.debug("ISSUE_IP: %s, %s ", out, err) + + +def release_ip(private_ip): + """Run del command on dhcpbridge""" + network_ref = db.fixed_ip_get_network(context.get_admin_context(), + private_ip) + instance_ref = db.fixed_ip_get_instance(context.get_admin_context(), + private_ip) + cmd = (binpath('nova-dhcpbridge'), 'del', + instance_ref['mac_address'], + private_ip, 'fake') + env = {'DNSMASQ_INTERFACE': network_ref['bridge'], + 'TESTING': '1', + 'FLAGFILE': FLAGS.dhcpbridge_flagfile} + (out, err) = utils.execute(*cmd, addl_env=env) + LOG.debug("RELEASE_IP: %s, %s ", out, err) diff --git a/nova/tests/network/base.py b/nova/tests/network/base.py index 2dd8178ff..988a1de72 100644 --- a/nova/tests/network/base.py +++ b/nova/tests/network/base.py @@ -1,154 +1,154 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -""" -Base class of Unit Tests for all network models -""" -import IPy -import os - -from nova import context -from nova import db -from nova import exception -from nova import flags -from nova import log as logging -from nova import test -from nova import utils -from nova.auth import manager - -FLAGS = flags.FLAGS -LOG = logging.getLogger('nova.tests.network') - - -class NetworkTestCase(test.TestCase): - """Test cases for network code""" - def setUp(self): - super(NetworkTestCase, self).setUp() - # NOTE(vish): if you change these flags, make sure to change the - # flags in the corresponding section in nova-dhcpbridge - self.flags(connection_type='fake', - fake_call=True, - fake_network=True) - self.manager = manager.AuthManager() - self.user = self.manager.create_user('netuser', 'netuser', 'netuser') - self.projects = [] - self.network = utils.import_object(FLAGS.network_manager) - self.context = context.RequestContext(project=None, user=self.user) - for i in range(FLAGS.num_networks): - name = 'project%s' % i - project = self.manager.create_project(name, 'netuser', name) - self.projects.append(project) - # create the necessary network data for the project - user_context = context.RequestContext(project=self.projects[i], - user=self.user) - host = self.network.get_network_host(user_context.elevated()) - instance_ref = self._create_instance(0) - self.instance_id = instance_ref['id'] - instance_ref = self._create_instance(1) - self.instance2_id = instance_ref['id'] - - def tearDown(self): - # TODO(termie): this should really be instantiating clean datastores - # in between runs, one failure kills all the tests - db.instance_destroy(context.get_admin_context(), self.instance_id) - db.instance_destroy(context.get_admin_context(), self.instance2_id) - for project in self.projects: - self.manager.delete_project(project) - self.manager.delete_user(self.user) - super(NetworkTestCase, self).tearDown() - - def _create_instance(self, project_num, mac=None): - if not mac: - mac = utils.generate_mac() - project = self.projects[project_num] - self.context._project = project - self.context.project_id = project.id - return db.instance_create(self.context, - {'project_id': project.id, - 'mac_address': mac}) - - def _create_address(self, project_num, instance_id=None): - """Create an address in given project num""" - if instance_id is None: - instance_id = self.instance_id - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - return self.network.allocate_fixed_ip(self.context, instance_id) - - def _deallocate_address(self, project_num, address): - self.context._project = self.projects[project_num] - self.context.project_id = self.projects[project_num].id - self.network.deallocate_fixed_ip(self.context, address) - - def _is_allocated_in_project(self, address, project_id): - """Returns true if address is in specified project""" - project_net = db.network_get_by_bridge(context.get_admin_context(), - FLAGS.flat_network_bridge) - network = db.fixed_ip_get_network(context.get_admin_context(), - address) - instance = db.fixed_ip_get_instance(context.get_admin_context(), - address) - # instance exists until release - return instance is not None and network['id'] == project_net['id'] - - def test_private_ipv6(self): - """Make sure ipv6 is OK""" - if FLAGS.use_ipv6: - instance_ref = self._create_instance(0) - address = self._create_address(0, instance_ref['id']) - network_ref = db.project_get_network( - context.get_admin_context(), - self.context.project_id) - address_v6 = db.instance_get_fixed_address_v6( - context.get_admin_context(), - instance_ref['id']) - self.assertEqual(instance_ref['mac_address'], - utils.to_mac(address_v6)) - instance_ref2 = db.fixed_ip_get_instance_v6( - context.get_admin_context(), - address_v6) - self.assertEqual(instance_ref['id'], instance_ref2['id']) - self.assertEqual(address_v6, - utils.to_global_ipv6( - network_ref['cidr_v6'], - instance_ref['mac_address'])) - self._deallocate_address(0, address) - db.instance_destroy(context.get_admin_context(), - instance_ref['id']) - - def test_available_ips(self): - """Make sure the number of available ips for the network is correct - - The number of available IP addresses depends on the test - environment's setup. - - Network size is set in test fixture's setUp method. - - There are ips reserved at the bottom and top of the range. - services (network, gateway, CloudPipe, broadcast) - """ - network = db.project_get_network(context.get_admin_context(), - self.projects[0].id) - net_size = flags.FLAGS.network_size - admin_context = context.get_admin_context() - total_ips = (db.network_count_available_ips(admin_context, - network['id']) + - db.network_count_reserved_ips(admin_context, - network['id']) + - db.network_count_allocated_ips(admin_context, - network['id'])) - self.assertEqual(total_ips, net_size) +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +""" +Base class of Unit Tests for all network models +""" +import IPy +import os + +from nova import context +from nova import db +from nova import exception +from nova import flags +from nova import log as logging +from nova import test +from nova import utils +from nova.auth import manager + +FLAGS = flags.FLAGS +LOG = logging.getLogger('nova.tests.network') + + +class NetworkTestCase(test.TestCase): + """Test cases for network code""" + def setUp(self): + super(NetworkTestCase, self).setUp() + # NOTE(vish): if you change these flags, make sure to change the + # flags in the corresponding section in nova-dhcpbridge + self.flags(connection_type='fake', + fake_call=True, + fake_network=True) + self.manager = manager.AuthManager() + self.user = self.manager.create_user('netuser', 'netuser', 'netuser') + self.projects = [] + self.network = utils.import_object(FLAGS.network_manager) + self.context = context.RequestContext(project=None, user=self.user) + for i in range(FLAGS.num_networks): + name = 'project%s' % i + project = self.manager.create_project(name, 'netuser', name) + self.projects.append(project) + # create the necessary network data for the project + user_context = context.RequestContext(project=self.projects[i], + user=self.user) + host = self.network.get_network_host(user_context.elevated()) + instance_ref = self._create_instance(0) + self.instance_id = instance_ref['id'] + instance_ref = self._create_instance(1) + self.instance2_id = instance_ref['id'] + + def tearDown(self): + # TODO(termie): this should really be instantiating clean datastores + # in between runs, one failure kills all the tests + db.instance_destroy(context.get_admin_context(), self.instance_id) + db.instance_destroy(context.get_admin_context(), self.instance2_id) + for project in self.projects: + self.manager.delete_project(project) + self.manager.delete_user(self.user) + super(NetworkTestCase, self).tearDown() + + def _create_instance(self, project_num, mac=None): + if not mac: + mac = utils.generate_mac() + project = self.projects[project_num] + self.context._project = project + self.context.project_id = project.id + return db.instance_create(self.context, + {'project_id': project.id, + 'mac_address': mac}) + + def _create_address(self, project_num, instance_id=None): + """Create an address in given project num""" + if instance_id is None: + instance_id = self.instance_id + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + return self.network.allocate_fixed_ip(self.context, instance_id) + + def _deallocate_address(self, project_num, address): + self.context._project = self.projects[project_num] + self.context.project_id = self.projects[project_num].id + self.network.deallocate_fixed_ip(self.context, address) + + def _is_allocated_in_project(self, address, project_id): + """Returns true if address is in specified project""" + project_net = db.network_get_by_bridge(context.get_admin_context(), + FLAGS.flat_network_bridge) + network = db.fixed_ip_get_network(context.get_admin_context(), + address) + instance = db.fixed_ip_get_instance(context.get_admin_context(), + address) + # instance exists until release + return instance is not None and network['id'] == project_net['id'] + + def test_private_ipv6(self): + """Make sure ipv6 is OK""" + if FLAGS.use_ipv6: + instance_ref = self._create_instance(0) + address = self._create_address(0, instance_ref['id']) + network_ref = db.project_get_network( + context.get_admin_context(), + self.context.project_id) + address_v6 = db.instance_get_fixed_address_v6( + context.get_admin_context(), + instance_ref['id']) + self.assertEqual(instance_ref['mac_address'], + utils.to_mac(address_v6)) + instance_ref2 = db.fixed_ip_get_instance_v6( + context.get_admin_context(), + address_v6) + self.assertEqual(instance_ref['id'], instance_ref2['id']) + self.assertEqual(address_v6, + utils.to_global_ipv6( + network_ref['cidr_v6'], + instance_ref['mac_address'])) + self._deallocate_address(0, address) + db.instance_destroy(context.get_admin_context(), + instance_ref['id']) + + def test_available_ips(self): + """Make sure the number of available ips for the network is correct + + The number of available IP addresses depends on the test + environment's setup. + + Network size is set in test fixture's setUp method. + + There are ips reserved at the bottom and top of the range. + services (network, gateway, CloudPipe, broadcast) + """ + network = db.project_get_network(context.get_admin_context(), + self.projects[0].id) + net_size = flags.FLAGS.network_size + admin_context = context.get_admin_context() + total_ips = (db.network_count_available_ips(admin_context, + network['id']) + + db.network_count_reserved_ips(admin_context, + network['id']) + + db.network_count_allocated_ips(admin_context, + network['id'])) + self.assertEqual(total_ips, net_size) diff --git a/nova/tests/test_flat_network.py b/nova/tests/test_flat_network.py index b6f7762c5..dcc617e25 100644 --- a/nova/tests/test_flat_network.py +++ b/nova/tests/test_flat_network.py @@ -1,161 +1,161 @@ -# vim: tabstop=4 shiftwidth=4 softtabstop=4 - -# Copyright 2010 United States Government as represented by the -# Administrator of the National Aeronautics and Space Administration. -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -""" -Unit Tests for flat network code -""" -import IPy -import os -import unittest - -from nova import context -from nova import db -from nova import exception -from nova import flags -from nova import log as logging -from nova import test -from nova import utils -from nova.auth import manager -from nova.tests.network import base - - -FLAGS = flags.FLAGS -LOG = logging.getLogger('nova.tests.network') - - -class FlatNetworkTestCase(base.NetworkTestCase): - """Test cases for network code""" - def test_public_network_association(self): - """Makes sure that we can allocate a public ip""" - # TODO(vish): better way of adding floating ips - - self.context._project = self.projects[0] - self.context.project_id = self.projects[0].id - pubnet = IPy.IP(flags.FLAGS.floating_range) - address = str(pubnet[0]) - try: - db.floating_ip_get_by_address(context.get_admin_context(), address) - except exception.NotFound: - db.floating_ip_create(context.get_admin_context(), - {'address': address, - 'host': FLAGS.host}) - - self.assertRaises(NotImplementedError, - self.network.allocate_floating_ip, - self.context, self.projects[0].id) - - fix_addr = self._create_address(0) - float_addr = address - self.assertRaises(NotImplementedError, - self.network.associate_floating_ip, - self.context, float_addr, fix_addr) - - address = db.instance_get_floating_address(context.get_admin_context(), - self.instance_id) - self.assertEqual(address, None) - - self.assertRaises(NotImplementedError, - self.network.disassociate_floating_ip, - self.context, float_addr) - - address = db.instance_get_floating_address(context.get_admin_context(), - self.instance_id) - self.assertEqual(address, None) - - self.assertRaises(NotImplementedError, - self.network.deallocate_floating_ip, - self.context, float_addr) - - self.network.deallocate_fixed_ip(self.context, fix_addr) - db.floating_ip_destroy(context.get_admin_context(), float_addr) - - def test_allocate_deallocate_fixed_ip(self): - """Makes sure that we can allocate and deallocate a fixed ip""" - address = self._create_address(0) - self.assertTrue(self._is_allocated_in_project(address, - self.projects[0].id)) - self._deallocate_address(0, address) - - # check if the fixed ip address is really deallocated - self.assertFalse(self._is_allocated_in_project(address, - self.projects[0].id)) - - def test_side_effects(self): - """Ensures allocating and releasing has no side effects""" - address = self._create_address(0) - address2 = self._create_address(1, self.instance2_id) - - self.assertTrue(self._is_allocated_in_project(address, - self.projects[0].id)) - self.assertTrue(self._is_allocated_in_project(address2, - self.projects[1].id)) - - self._deallocate_address(0, address) - self.assertFalse(self._is_allocated_in_project(address, - self.projects[0].id)) - - # First address release shouldn't affect the second - self.assertTrue(self._is_allocated_in_project(address2, - self.projects[0].id)) - - self._deallocate_address(1, address2) - self.assertFalse(self._is_allocated_in_project(address2, - self.projects[1].id)) - - def test_ips_are_reused(self): - """Makes sure that ip addresses that are deallocated get reused""" - address = self._create_address(0) - self.network.deallocate_fixed_ip(self.context, address) - - address2 = self._create_address(0) - self.assertEqual(address, address2) - - self.network.deallocate_fixed_ip(self.context, address2) - - def test_too_many_addresses(self): - """Test for a NoMoreAddresses exception when all fixed ips are used. - """ - admin_context = context.get_admin_context() - network = db.project_get_network(admin_context, self.projects[0].id) - num_available_ips = db.network_count_available_ips(admin_context, - network['id']) - addresses = [] - instance_ids = [] - for i in range(num_available_ips): - instance_ref = self._create_instance(0) - instance_ids.append(instance_ref['id']) - address = self._create_address(0, instance_ref['id']) - addresses.append(address) - - ip_count = db.network_count_available_ips(context.get_admin_context(), - network['id']) - self.assertEqual(ip_count, 0) - self.assertRaises(db.NoMoreAddresses, - self.network.allocate_fixed_ip, - self.context, - 'foo') - - for i in range(num_available_ips): - self.network.deallocate_fixed_ip(self.context, addresses[i]) - db.instance_destroy(context.get_admin_context(), instance_ids[i]) - ip_count = db.network_count_available_ips(context.get_admin_context(), - network['id']) - self.assertEqual(ip_count, num_available_ips) - - def run(self, result=None): - if(FLAGS.network_manager == 'nova.network.manager.FlatManager'): - super(FlatNetworkTestCase, self).run(result) +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +""" +Unit Tests for flat network code +""" +import IPy +import os +import unittest + +from nova import context +from nova import db +from nova import exception +from nova import flags +from nova import log as logging +from nova import test +from nova import utils +from nova.auth import manager +from nova.tests.network import base + + +FLAGS = flags.FLAGS +LOG = logging.getLogger('nova.tests.network') + + +class FlatNetworkTestCase(base.NetworkTestCase): + """Test cases for network code""" + def test_public_network_association(self): + """Makes sure that we can allocate a public ip""" + # TODO(vish): better way of adding floating ips + + self.context._project = self.projects[0] + self.context.project_id = self.projects[0].id + pubnet = IPy.IP(flags.FLAGS.floating_range) + address = str(pubnet[0]) + try: + db.floating_ip_get_by_address(context.get_admin_context(), address) + except exception.NotFound: + db.floating_ip_create(context.get_admin_context(), + {'address': address, + 'host': FLAGS.host}) + + self.assertRaises(NotImplementedError, + self.network.allocate_floating_ip, + self.context, self.projects[0].id) + + fix_addr = self._create_address(0) + float_addr = address + self.assertRaises(NotImplementedError, + self.network.associate_floating_ip, + self.context, float_addr, fix_addr) + + address = db.instance_get_floating_address(context.get_admin_context(), + self.instance_id) + self.assertEqual(address, None) + + self.assertRaises(NotImplementedError, + self.network.disassociate_floating_ip, + self.context, float_addr) + + address = db.instance_get_floating_address(context.get_admin_context(), + self.instance_id) + self.assertEqual(address, None) + + self.assertRaises(NotImplementedError, + self.network.deallocate_floating_ip, + self.context, float_addr) + + self.network.deallocate_fixed_ip(self.context, fix_addr) + db.floating_ip_destroy(context.get_admin_context(), float_addr) + + def test_allocate_deallocate_fixed_ip(self): + """Makes sure that we can allocate and deallocate a fixed ip""" + address = self._create_address(0) + self.assertTrue(self._is_allocated_in_project(address, + self.projects[0].id)) + self._deallocate_address(0, address) + + # check if the fixed ip address is really deallocated + self.assertFalse(self._is_allocated_in_project(address, + self.projects[0].id)) + + def test_side_effects(self): + """Ensures allocating and releasing has no side effects""" + address = self._create_address(0) + address2 = self._create_address(1, self.instance2_id) + + self.assertTrue(self._is_allocated_in_project(address, + self.projects[0].id)) + self.assertTrue(self._is_allocated_in_project(address2, + self.projects[1].id)) + + self._deallocate_address(0, address) + self.assertFalse(self._is_allocated_in_project(address, + self.projects[0].id)) + + # First address release shouldn't affect the second + self.assertTrue(self._is_allocated_in_project(address2, + self.projects[0].id)) + + self._deallocate_address(1, address2) + self.assertFalse(self._is_allocated_in_project(address2, + self.projects[1].id)) + + def test_ips_are_reused(self): + """Makes sure that ip addresses that are deallocated get reused""" + address = self._create_address(0) + self.network.deallocate_fixed_ip(self.context, address) + + address2 = self._create_address(0) + self.assertEqual(address, address2) + + self.network.deallocate_fixed_ip(self.context, address2) + + def test_too_many_addresses(self): + """Test for a NoMoreAddresses exception when all fixed ips are used. + """ + admin_context = context.get_admin_context() + network = db.project_get_network(admin_context, self.projects[0].id) + num_available_ips = db.network_count_available_ips(admin_context, + network['id']) + addresses = [] + instance_ids = [] + for i in range(num_available_ips): + instance_ref = self._create_instance(0) + instance_ids.append(instance_ref['id']) + address = self._create_address(0, instance_ref['id']) + addresses.append(address) + + ip_count = db.network_count_available_ips(context.get_admin_context(), + network['id']) + self.assertEqual(ip_count, 0) + self.assertRaises(db.NoMoreAddresses, + self.network.allocate_fixed_ip, + self.context, + 'foo') + + for i in range(num_available_ips): + self.network.deallocate_fixed_ip(self.context, addresses[i]) + db.instance_destroy(context.get_admin_context(), instance_ids[i]) + ip_count = db.network_count_available_ips(context.get_admin_context(), + network['id']) + self.assertEqual(ip_count, num_available_ips) + + def run(self, result=None): + if(FLAGS.network_manager == 'nova.network.manager.FlatManager'): + super(FlatNetworkTestCase, self).run(result) -- cgit From a3fe673108602e27cca132209e87369fa8bf1323 Mon Sep 17 00:00:00 2001 From: Tushar Patil Date: Fri, 18 Mar 2011 19:46:04 -0700 Subject: Changed Copyright to NTT for newly added files for flatmanager ipv6 --- nova/tests/network/__init__.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) (limited to 'nova/tests') diff --git a/nova/tests/network/__init__.py b/nova/tests/network/__init__.py index e0d479f8c..97f96b6fa 100644 --- a/nova/tests/network/__init__.py +++ b/nova/tests/network/__init__.py @@ -1,3 +1,23 @@ +# vim: tabstop=4 shiftwidth=4 softtabstop=4 + +# Copyright 2010 United States Government as represented by the +# Administrator of the National Aeronautics and Space Administration. +# All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +""" +Utility methods +""" import os from nova import context -- cgit