openstack
/
nova
Code Issues Proposed changes

nova-net: Convert remaining unit tests to neutron 

Convert the remaining few unit test that aren't specific to nova-network
to "use neutron". In most cases, this simply means dropping unnecessary
'use_neutron=True' flag overrides, though there are some additional
things that need to be done.

Change-Id: I3d30fc9f823b02a1651646a01ad83b5c3e781325
Signed-off-by: Stephen Finucane <sfinucan@redhat.com>
This commit is contained in:
Stephen Finucane 2019-11-27 10:09:00 +00:00
parent edf7775f16
commit 80e64186e6
8 changed files with 56 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
nova/tests/unit/api/openstack/fakes.py
View File

@ -209,23 +209,15 @@ def stub_out_nw_api(test, cls=None, private=None, publics=None):
if cls is None: if cls is None:
cls = Fake cls = Fake
if CONF.use_neutron:
test.stub_out('nova.network.neutronv2.api.API', cls) test.stub_out('nova.network.neutronv2.api.API', cls)
else:
test.stub_out('nova.network.api.API', cls)
fake_network.stub_out_nw_api_get_instance_nw_info(test)
def stub_out_secgroup_api(test, security_groups=None): def stub_out_secgroup_api(test, security_groups=None):
class FakeSecurityGroupAPI(security_group_base.SecurityGroupBase): class FakeSecurityGroupAPI(security_group_base.SecurityGroupBase):
"""This handles both nova-network and neutron style security group APIs
"""
def get_instances_security_groups_bindings( def get_instances_security_groups_bindings(
self, context, servers, detailed=False): self, context, servers, detailed=False):
# This method shouldn't be called unless using neutron.
if not CONF.use_neutron:
raise Exception('Invalid security group API call for nova-net')
instances_security_group_bindings = {} instances_security_group_bindings = {}
if servers: if servers:
# we don't get security group information for down cells # we don't get security group information for down cells
@ -239,13 +231,9 @@ def stub_out_secgroup_api(test, security_groups=None):
self, context, instance, detailed=False): self, context, instance, detailed=False):
return security_groups if security_groups is not None else [] return security_groups if security_groups is not None else []
if CONF.use_neutron: test.stub_out(
test.stub_out( 'nova.network.security_group.neutron_driver.SecurityGroupAPI',
'nova.network.security_group.neutron_driver.SecurityGroupAPI', FakeSecurityGroupAPI)
FakeSecurityGroupAPI)
else:
test.stub_out(
'nova.compute.api.SecurityGroupAPI', FakeSecurityGroupAPI)
class FakeToken(object): class FakeToken(object):

83
nova/tests/unit/compute/test_compute.py
View File

@ -240,25 +240,13 @@ class BaseTestCase(test.TestCase):
def fake_get_nw_info(cls, ctxt, instance, *args, **kwargs): def fake_get_nw_info(cls, ctxt, instance, *args, **kwargs):
return network_model.NetworkInfo() return network_model.NetworkInfo()
if CONF.use_neutron: self.stub_out(
self.stub_out( 'nova.network.neutronv2.api.API.get_instance_nw_info',
'nova.network.neutronv2.api.API.get_instance_nw_info', fake_get_nw_info)
fake_get_nw_info) self.stub_out(
self.stub_out( 'nova.network.neutronv2.api.API.migrate_instance_start',
'nova.network.neutronv2.api.API.migrate_instance_start', lambda *args, **kwargs: None)
lambda *args, **kwargs: None) self.useFixture(fixtures.NeutronFixture(self))
self.useFixture(fixtures.NeutronFixture(self))
else:
self.stub_out('nova.network.api.API.get_instance_nw_info',
fake_get_nw_info)
def fake_allocate_for_instance(cls, ctxt, instance,
*args, **kwargs):
self.assertFalse(ctxt.is_admin)
return fake_network.fake_get_instance_nw_info(self, 1, 1)
self.stub_out('nova.network.api.API.allocate_for_instance',
fake_allocate_for_instance)
self.compute_api = compute.API() self.compute_api = compute.API()
@ -1704,9 +1692,6 @@ class ComputeTestCase(BaseTestCase,
self.assertEqual(17152, cn.local_gb_used) self.assertEqual(17152, cn.local_gb_used)
def test_create_multiple_instance_with_neutron_port(self): def test_create_multiple_instance_with_neutron_port(self):
def fake_is_neutron():
return True
self.stub_out('nova.utils.is_neutron', fake_is_neutron)
requested_networks = objects.NetworkRequestList( requested_networks = objects.NetworkRequestList(
objects=[objects.NetworkRequest(port_id=uuids.port_instance)]) objects=[objects.NetworkRequest(port_id=uuids.port_instance)])
self.assertRaises(exception.MultiplePortsNotApplicable, self.assertRaises(exception.MultiplePortsNotApplicable,
@ -6145,11 +6130,9 @@ class ComputeTestCase(BaseTestCase,
# Confirm setup_compute_volume is called when volume is mounted. # Confirm setup_compute_volume is called when volume is mounted.
def stupid(*args, **kwargs): def stupid(*args, **kwargs):
return fake_network.fake_get_instance_nw_info(self) return fake_network.fake_get_instance_nw_info(self)
if CONF.use_neutron:
self.stub_out( self.stub_out(
'nova.network.neutronv2.api.API.get_instance_nw_info', stupid) 'nova.network.neutronv2.api.API.get_instance_nw_info', stupid)
else:
self.stub_out('nova.network.api.API.get_instance_nw_info', stupid)
# creating instance testdata # creating instance testdata
instance = self._create_fake_instance_obj({'host': 'dummy'}) instance = self._create_fake_instance_obj({'host': 'dummy'})
@ -7152,23 +7135,9 @@ class ComputeTestCase(BaseTestCase,
CONF.running_deleted_instance_timeout) CONF.running_deleted_instance_timeout)
@mock.patch('nova.network.neutronv2.api.API.list_ports') @mock.patch('nova.network.neutronv2.api.API.list_ports')
@mock.patch.object(nova.utils, 'is_neutron') def test_require_nw_info_update_host_match(self, mock_list_ports):
def test_require_nw_info_update_not_neutron(self, mock_is_neutron,
mock_list_ports):
ctxt = context.get_admin_context() ctxt = context.get_admin_context()
instance = self._create_fake_instance_obj() instance = self._create_fake_instance_obj()
mock_is_neutron.return_value = False
val = self.compute._require_nw_info_update(ctxt, instance)
self.assertFalse(val)
mock_list_ports.assert_not_called()
@mock.patch('nova.network.neutronv2.api.API.list_ports')
@mock.patch.object(nova.utils, 'is_neutron')
def test_require_nw_info_update_host_match(self, mock_is_neutron,
mock_list_ports):
ctxt = context.get_admin_context()
instance = self._create_fake_instance_obj()
mock_is_neutron.return_value = True
mock_list_ports.return_value = {'ports': [ mock_list_ports.return_value = {'ports': [
{'binding:host_id': self.compute.host, {'binding:host_id': self.compute.host,
'binding:vif_type': 'foo'} 'binding:vif_type': 'foo'}
@ -7183,12 +7152,9 @@ class ComputeTestCase(BaseTestCase,
mock_list_ports.assert_called_once_with(ctxt, **search_opts) mock_list_ports.assert_called_once_with(ctxt, **search_opts)
@mock.patch('nova.network.neutronv2.api.API.list_ports') @mock.patch('nova.network.neutronv2.api.API.list_ports')
@mock.patch.object(nova.utils, 'is_neutron') def test_require_nw_info_update_host_mismatch(self, mock_list_ports):
def test_require_nw_info_update_host_mismatch(self, mock_is_neutron,
mock_list_ports):
ctxt = context.get_admin_context() ctxt = context.get_admin_context()
instance = self._create_fake_instance_obj() instance = self._create_fake_instance_obj()
mock_is_neutron.return_value = True
mock_list_ports.return_value = {'ports': [ mock_list_ports.return_value = {'ports': [
{'binding:host_id': 'foo', 'binding:vif_type': 'foo'} {'binding:host_id': 'foo', 'binding:vif_type': 'foo'}
]} ]}
@ -7199,12 +7165,9 @@ class ComputeTestCase(BaseTestCase,
mock_list_ports.assert_called_once_with(ctxt, **search_opts) mock_list_ports.assert_called_once_with(ctxt, **search_opts)
@mock.patch('nova.network.neutronv2.api.API.list_ports') @mock.patch('nova.network.neutronv2.api.API.list_ports')
@mock.patch.object(nova.utils, 'is_neutron') def test_require_nw_info_update_failed_vif_types(self, mock_list_ports):
def test_require_nw_info_update_failed_vif_types(
self, mock_is_neutron, mock_list_ports):
ctxt = context.get_admin_context() ctxt = context.get_admin_context()
instance = self._create_fake_instance_obj() instance = self._create_fake_instance_obj()
mock_is_neutron.return_value = True
search_opts = {'device_id': instance.uuid, search_opts = {'device_id': instance.uuid,
'fields': ['binding:host_id', 'binding:vif_type']} 'fields': ['binding:host_id', 'binding:vif_type']}
@ -7313,20 +7276,12 @@ class ComputeTestCase(BaseTestCase,
'_require_nw_info_update', '_require_nw_info_update',
fake_require_nw_info_update) fake_require_nw_info_update)
if CONF.use_neutron: self.stub_out(
self.stub_out( 'nova.network.neutronv2.api.API.get_instance_nw_info',
'nova.network.neutronv2.api.API.get_instance_nw_info', fake_get_instance_nw_info)
fake_get_instance_nw_info) self.stub_out(
self.stub_out( 'nova.network.neutronv2.api.API.setup_instance_network_on_host',
'nova.network.neutronv2.api.API.' fake_setup_instance_network_on_host)
'setup_instance_network_on_host',
fake_setup_instance_network_on_host)
else:
self.stub_out('nova.network.api.API.get_instance_nw_info',
fake_get_instance_nw_info)
self.stub_out(
'nova.network.api.API.setup_instance_network_on_host',
fake_setup_instance_network_on_host)
expected_require_nw_info = 0 expected_require_nw_info = 0
expect_setup_network = 0 expect_setup_network = 0

19
nova/tests/unit/compute/test_compute_api.py
View File

@ -318,7 +318,6 @@ class _ComputeAPIUnitTestMixIn(object):
# Tests that if port is specified there is only one instance booting # Tests that if port is specified there is only one instance booting
# (i.e max_count == 1) as we can't share the same port across multiple # (i.e max_count == 1) as we can't share the same port across multiple
# instances. # instances.
self.flags(use_neutron=True)
port = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa' port = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
address = '10.0.0.1' address = '10.0.0.1'
min_count = 1 min_count = 1
@ -353,7 +352,6 @@ class _ComputeAPIUnitTestMixIn(object):
requested_networks) requested_networks)
def test_specified_ip_and_multiple_instances_neutronv2(self): def test_specified_ip_and_multiple_instances_neutronv2(self):
self.flags(use_neutron=True)
network = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa' network = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
address = '10.0.0.1' address = '10.0.0.1'
requested_networks = objects.NetworkRequestList( requested_networks = objects.NetworkRequestList(
@ -6050,7 +6048,6 @@ class _ComputeAPIUnitTestMixIn(object):
"""Tests that a list of security groups passed in do not actually get """Tests that a list of security groups passed in do not actually get
stored on with the instance when using neutron. stored on with the instance when using neutron.
""" """
self.flags(use_neutron=True)
flavor = self._create_flavor() flavor = self._create_flavor()
params = {'display_name': 'fake-instance'} params = {'display_name': 'fake-instance'}
instance = self._create_instance_obj(params, flavor) instance = self._create_instance_obj(params, flavor)
@ -6563,8 +6560,8 @@ class ComputeAPIUnitTestCase(_ComputeAPIUnitTestMixIn, test.NoDBTestCase):
@mock.patch.object(neutron_api.API, 'list_ports') @mock.patch.object(neutron_api.API, 'list_ports')
@mock.patch.object(objects.BuildRequestList, 'get_by_filters', @mock.patch.object(objects.BuildRequestList, 'get_by_filters',
new_callable=mock.NonCallableMock) new_callable=mock.NonCallableMock)
def test_get_all_ip_filter_use_neutron(self, mock_buildreq_get, def test_get_all_ip_filter(self, mock_buildreq_get, mock_list_port,
mock_list_port, mock_check_ext): mock_check_ext):
mock_check_ext.return_value = True mock_check_ext.return_value = True
cell_instances = self._list_of_instances(2) cell_instances = self._list_of_instances(2)
mock_list_port.return_value = { mock_list_port.return_value = {
@ -6591,8 +6588,8 @@ class ComputeAPIUnitTestCase(_ComputeAPIUnitTestMixIn, test.NoDBTestCase):
@mock.patch.object(neutron_api.API, 'list_ports') @mock.patch.object(neutron_api.API, 'list_ports')
@mock.patch.object(objects.BuildRequestList, 'get_by_filters', @mock.patch.object(objects.BuildRequestList, 'get_by_filters',
new_callable=mock.NonCallableMock) new_callable=mock.NonCallableMock)
def test_get_all_ip6_filter_use_neutron(self, mock_buildreq_get, def test_get_all_ip6_filter(self, mock_buildreq_get, mock_list_port,
mock_list_port, mock_check_ext): mock_check_ext):
mock_check_ext.return_value = True mock_check_ext.return_value = True
cell_instances = self._list_of_instances(2) cell_instances = self._list_of_instances(2)
mock_list_port.return_value = { mock_list_port.return_value = {
@ -6619,9 +6616,8 @@ class ComputeAPIUnitTestCase(_ComputeAPIUnitTestMixIn, test.NoDBTestCase):
@mock.patch.object(neutron_api.API, 'list_ports') @mock.patch.object(neutron_api.API, 'list_ports')
@mock.patch.object(objects.BuildRequestList, 'get_by_filters', @mock.patch.object(objects.BuildRequestList, 'get_by_filters',
new_callable=mock.NonCallableMock) new_callable=mock.NonCallableMock)
def test_get_all_ip_and_ip6_filter_use_neutron(self, mock_buildreq_get, def test_get_all_ip_and_ip6_filter(self, mock_buildreq_get, mock_list_port,
mock_list_port, mock_check_ext):
mock_check_ext):
mock_check_ext.return_value = True mock_check_ext.return_value = True
cell_instances = self._list_of_instances(2) cell_instances = self._list_of_instances(2)
mock_list_port.return_value = { mock_list_port.return_value = {
@ -6652,8 +6648,7 @@ class ComputeAPIUnitTestCase(_ComputeAPIUnitTestMixIn, test.NoDBTestCase):
@mock.patch.object(neutron_api.API, 'has_substr_port_filtering_extension') @mock.patch.object(neutron_api.API, 'has_substr_port_filtering_extension')
@mock.patch.object(neutron_api.API, 'list_ports') @mock.patch.object(neutron_api.API, 'list_ports')
def test_get_all_ip6_filter_use_neutron_exc(self, mock_list_port, def test_get_all_ip6_filter_exc(self, mock_list_port, mock_check_ext):
mock_check_ext):
mock_check_ext.return_value = True mock_check_ext.return_value = True
mock_list_port.side_effect = exception.InternalError('fake') mock_list_port.side_effect = exception.InternalError('fake')

43
nova/tests/unit/compute/test_compute_mgr.py
View File

@ -6138,7 +6138,6 @@ class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase):
def test_rescheduled_exception(self, mock_hooks, mock_build_run, def test_rescheduled_exception(self, mock_hooks, mock_build_run,
mock_build, mock_set, mock_nil, mock_build, mock_set, mock_nil,
mock_save, mock_start, mock_finish): mock_save, mock_start, mock_finish):
self.flags(use_neutron=False)
self._do_build_instance_update(mock_save, reschedule_update=True) self._do_build_instance_update(mock_save, reschedule_update=True)
mock_build_run.side_effect = exception.RescheduledException(reason='', mock_build_run.side_effect = exception.RescheduledException(reason='',
instance_uuid=self.instance.uuid) instance_uuid=self.instance.uuid)
@ -6166,8 +6165,7 @@ class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase):
self.requested_networks, self.security_groups, self.requested_networks, self.security_groups,
self.block_device_mapping, self.node, self.limits, self.block_device_mapping, self.node, self.limits,
self.filter_properties, {}) self.filter_properties, {})
mock_clean.assert_called_once_with(self.context, self.instance, mock_clean.assert_not_called()
self.compute.host)
mock_nil.assert_called_once_with(self.instance) mock_nil.assert_called_once_with(self.instance)
mock_build.assert_called_once_with(self.context, mock_build.assert_called_once_with(self.context,
[self.instance], self.image, self.filter_properties, [self.instance], self.image, self.filter_properties,
@ -6224,7 +6222,6 @@ class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase):
mock_event_finish, mock_event_finish,
mock_event_start, mock_ins_save, mock_event_start, mock_ins_save,
mock_build_ins, mock_build_and_run): mock_build_ins, mock_build_and_run):
self.flags(use_neutron=False)
instance = fake_instance.fake_instance_obj(self.context, instance = fake_instance.fake_instance_obj(self.context,
vm_state=vm_states.ACTIVE, vm_state=vm_states.ACTIVE,
system_metadata={'network_allocated': 'True'}, system_metadata={'network_allocated': 'True'},
@ -6252,8 +6249,7 @@ class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase):
self.requested_networks, self.security_groups, self.requested_networks, self.security_groups,
self.block_device_mapping, self.node, self.limits, self.block_device_mapping, self.node, self.limits,
self.filter_properties, {}) self.filter_properties, {})
mock_cleanup_network.assert_called_once_with( mock_cleanup_network.assert_not_called()
self.context, instance, self.compute.host)
mock_build_ins.assert_called_once_with(self.context, mock_build_ins.assert_called_once_with(self.context,
[instance], self.image, self.filter_properties, [instance], self.image, self.filter_properties,
self.admin_pass, self.injected_files, self.requested_networks, self.admin_pass, self.injected_files, self.requested_networks,
@ -6424,7 +6420,6 @@ class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase):
mock_build_run, mock_build, mock_deallocate, mock_nil, mock_build_run, mock_build, mock_deallocate, mock_nil,
mock_clean_net, mock_save, mock_start, mock_clean_net, mock_save, mock_start,
mock_finish): mock_finish):
self.flags(use_neutron=False)
self._do_build_instance_update(mock_save, reschedule_update=True) self._do_build_instance_update(mock_save, reschedule_update=True)
mock_build_run.side_effect = exception.RescheduledException(reason='', mock_build_run.side_effect = exception.RescheduledException(reason='',
instance_uuid=self.instance.uuid) instance_uuid=self.instance.uuid)
@ -6454,8 +6449,7 @@ class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase):
self.block_device_mapping, self.node, self.limits, self.block_device_mapping, self.node, self.limits,
self.filter_properties, {}) self.filter_properties, {})
mock_deallocate.assert_called_once_with(self.instance) mock_deallocate.assert_called_once_with(self.instance)
mock_clean_inst.assert_called_once_with(self.context, self.instance, mock_clean_inst.assert_not_called()
self.compute.host)
mock_nil.assert_called_once_with(self.instance) mock_nil.assert_called_once_with(self.instance)
mock_build.assert_called_once_with(self.context, mock_build.assert_called_once_with(self.context,
[self.instance], self.image, self.filter_properties, [self.instance], self.image, self.filter_properties,
@ -6877,7 +6871,6 @@ class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase):
def test_reschedule_on_resources_unavailable(self, mock_claim, def test_reschedule_on_resources_unavailable(self, mock_claim,
mock_build, mock_nil, mock_save, mock_start, mock_build, mock_nil, mock_save, mock_start,
mock_finish, mock_notify): mock_finish, mock_notify):
self.flags(use_neutron=False)
reason = 'resource unavailable' reason = 'resource unavailable'
exc = exception.ComputeResourcesUnavailable(reason=reason) exc = exception.ComputeResourcesUnavailable(reason=reason)
mock_claim.side_effect = exc mock_claim.side_effect = exc
@ -6911,8 +6904,7 @@ class ComputeManagerBuildInstanceTestCase(test.NoDBTestCase):
self.security_groups, self.block_device_mapping, self.security_groups, self.block_device_mapping,
request_spec={}, host_lists=[fake_host_list]) request_spec={}, host_lists=[fake_host_list])
mock_nil.assert_called_once_with(self.instance) mock_nil.assert_called_once_with(self.instance)
mock_clean.assert_called_once_with(self.context, self.instance, mock_clean.assert_not_called()
self.compute.host)
@mock.patch.object(manager.ComputeManager, '_build_resources') @mock.patch.object(manager.ComputeManager, '_build_resources')
@mock.patch.object(objects.Instance, 'save') @mock.patch.object(objects.Instance, 'save')
@ -8759,19 +8751,24 @@ class ComputeManagerMigrationTestCase(test.NoDBTestCase,
"""Tests the various ways that _get_neutron_events_for_live_migration """Tests the various ways that _get_neutron_events_for_live_migration
will return an empty list. will return an empty list.
""" """
nw = network_model.NetworkInfo([network_model.VIF(uuids.port1)])
# 1. no timeout # 1. no timeout
self.flags(vif_plugging_timeout=0) self.flags(vif_plugging_timeout=0)
self.assertEqual(
[], self.compute._get_neutron_events_for_live_migration(nw)) with mock.patch.object(self.instance, 'get_network_info') as nw_info:
# 2. not neutron nw_info.return_value = network_model.NetworkInfo(
self.flags(vif_plugging_timeout=300, use_neutron=False) [network_model.VIF(uuids.port1)])
self.assertEqual( self.assertEqual(
[], self.compute._get_neutron_events_for_live_migration(nw)) [], self.compute._get_neutron_events_for_live_migration(
# 3. no VIFs self.instance))
self.flags(vif_plugging_timeout=300, use_neutron=True)
self.assertEqual( # 2. no VIFs
[], self.compute._get_neutron_events_for_live_migration([])) self.flags(vif_plugging_timeout=300)
with mock.patch.object(self.instance, 'get_network_info') as nw_info:
nw_info.return_value = []
self.assertEqual(
[], self.compute._get_neutron_events_for_live_migration(
self.instance))
@mock.patch('nova.compute.rpcapi.ComputeAPI.pre_live_migration') @mock.patch('nova.compute.rpcapi.ComputeAPI.pre_live_migration')
@mock.patch('nova.compute.manager.ComputeManager._post_live_migration') @mock.patch('nova.compute.manager.ComputeManager._post_live_migration')

9
nova/tests/unit/test_metadata.py
View File

@ -293,7 +293,7 @@ class MetadataTestCase(test.TestCase):
md.get_ec2_metadata(version='2009-04-04').get('user-data', obj), md.get_ec2_metadata(version='2009-04-04').get('user-data', obj),
obj) obj)
def _test_security_groups(self): def test_security_groups(self):
inst = self.instance.obj_clone() inst = self.instance.obj_clone()
sgroups = [{'name': name} for name in ('default', 'other')] sgroups = [{'name': name} for name in ('default', 'other')]
expected = ['default', 'other'] expected = ['default', 'other']
@ -302,13 +302,6 @@ class MetadataTestCase(test.TestCase):
data = md.get_ec2_metadata(version='2009-04-04') data = md.get_ec2_metadata(version='2009-04-04')
self.assertEqual(data['meta-data']['security-groups'], expected) self.assertEqual(data['meta-data']['security-groups'], expected)
def test_security_groups(self):
self._test_security_groups()
def test_neutron_security_groups(self):
self.flags(use_neutron=True)
self._test_security_groups()
def test_local_hostname(self): def test_local_hostname(self):
self.flags(dhcp_domain=None, group='api') self.flags(dhcp_domain=None, group='api')
md = fake_InstanceMetadata(self, self.instance.obj_clone()) md = fake_InstanceMetadata(self, self.instance.obj_clone())

53
nova/tests/unit/test_quota.py
View File

@ -22,7 +22,6 @@ from six.moves import range
from nova.compute import api as compute from nova.compute import api as compute
import nova.conf import nova.conf
from nova import context from nova import context
from nova.db import api as db
from nova.db.sqlalchemy import models as sqa_models from nova.db.sqlalchemy import models as sqa_models
from nova import exception from nova import exception
from nova import objects from nova import objects
@ -152,58 +151,6 @@ class QuotaIntegrationTestCase(test.TestCase):
for i in range(3): for i in range(3):
self._create_instance(flavor_name='m1.xlarge') self._create_instance(flavor_name='m1.xlarge')
@mock.patch('nova.privsep.linux_net.bind_ip')
@mock.patch('nova.privsep.linux_net.iptables_get_rules',
return_value=('', ''))
@mock.patch('nova.privsep.linux_net.iptables_set_rules',
return_value=('', ''))
def test_too_many_addresses(self, mock_iptables_set_rules,
mock_iptables_get_rules, mock_bind_ip):
# This test is specifically relying on nova-network.
self.flags(use_neutron=False,
network_manager='nova.network.manager.FlatDHCPManager')
self.flags(floating_ips=1, group='quota')
# Apparently needed by the RPC tests...
self.network = self.start_service('network',
manager=CONF.network_manager)
address = '192.168.0.100'
db.floating_ip_create(context.get_admin_context(),
{'address': address,
'pool': 'nova',
'project_id': self.project_id})
self.assertRaises(exception.QuotaError,
self.network.allocate_floating_ip,
self.context,
self.project_id)
db.floating_ip_destroy(context.get_admin_context(), address)
@mock.patch('nova.privsep.linux_net.bind_ip')
@mock.patch('nova.privsep.linux_net.iptables_get_rules',
return_value=('', ''))
@mock.patch('nova.privsep.linux_net.iptables_set_rules',
return_value=('', ''))
def test_auto_assigned(self, mock_iptables_set_rules,
mock_iptables_get_rules, mock_bind_ip):
# This test is specifically relying on nova-network.
self.flags(use_neutron=False,
network_manager='nova.network.manager.FlatDHCPManager')
self.flags(floating_ips=1, group='quota')
# Apparently needed by the RPC tests...
self.network = self.start_service('network',
manager=CONF.network_manager)
address = '192.168.0.100'
db.floating_ip_create(context.get_admin_context(),
{'address': address,
'pool': 'nova',
'project_id': self.project_id})
# auto allocated addresses should not be counted
self.assertRaises(exception.NoMoreFloatingIps,
self.network.allocate_floating_ip,
self.context,
self.project_id,
True)
db.floating_ip_destroy(context.get_admin_context(), address)
def test_too_many_metadata_items(self): def test_too_many_metadata_items(self):
metadata = {} metadata = {}
for i in range(CONF.quota.metadata_items + 1): for i in range(CONF.quota.metadata_items + 1):

6
nova/tests/unit/utils.py
View File

@ -166,15 +166,11 @@ def get_test_network_info(count=1):
vlan=FAKE_NETWORK_VLAN, vlan=FAKE_NETWORK_VLAN,
bridge_interface=FAKE_NETWORK_INTERFACE, bridge_interface=FAKE_NETWORK_INTERFACE,
injected=False) injected=False)
if CONF.use_neutron:
vif_type = network_model.VIF_TYPE_OVS
else:
vif_type = network_model.VIF_TYPE_BRIDGE
vif = network_model.VIF( vif = network_model.VIF(
id=FAKE_VIF_UUID, id=FAKE_VIF_UUID,
address=FAKE_VIF_MAC, address=FAKE_VIF_MAC,
network=network, network=network,
type=vif_type, type=network_model.VIF_TYPE_OVS,
devname=None, devname=None,
ovs_interfaceid=None) ovs_interfaceid=None)

6
nova/tests/unit/virt/test_virt_drivers.py
View File

@ -895,9 +895,9 @@ class LibvirtConnTestCase(_VirtDriverTestCase, test.TestCase):
# This is needed for the live migration tests which spawn off the # This is needed for the live migration tests which spawn off the
# operation for monitoring. # operation for monitoring.
self.useFixture(nova_fixtures.SpawnIsSynchronousFixture()) self.useFixture(nova_fixtures.SpawnIsSynchronousFixture())
# When using CONF.use_neutron=True and destroying an instance os-vif # When destroying an instance, os-vif will try to execute some commands
# will try to execute some commands which hangs tests so let's just # which hang tests so let's just stub out the unplug call to os-vif
# stub out the unplug call to os-vif since we don't care about it. # since we don't care about it.
self.stub_out('os_vif.unplug', lambda a, kw: None) self.stub_out('os_vif.unplug', lambda a, kw: None)
self.stub_out('nova.compute.utils.get_machine_ips', lambda: []) self.stub_out('nova.compute.utils.get_machine_ips', lambda: [])