openstack/nova
Code Issues Proposed changes

Make test_compute pass with CONF.use_neutron=True by default

Browse Source 
More of the same, the nova-network API methods are mocked
out directly rather than the ComputeManager.network_api variable.
We also use the NeutronFixture globally here rather than the
global nova-network method stubs in setUp.

The NeutronFixture also needs a few additional stub methods.

For the compute API tests, the changes are mostly just around
knowing that the security group API tests rely on nova-network.

Part of blueprint use-neutron-by-default

Change-Id: I9b670a2f1f1f67ad46a5ac4a9696eaf94b1058ce
This commit is contained in:
Matt Riedemann 2016-11-19 20:48:46 -05:00
parent 34f4b1bd68
commit 5b4758cc77
2 changed files with 173 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
nova/tests/fixtures.py
View File

@ -969,6 +969,15 @@ class NeutronFixture(fixtures.Fixture):
'nova.network.neutronv2.api.API.'
'create_pci_requests_for_sriov_ports',
lambda *args, **kwargs: None)
self.test.stub_out(
'nova.network.neutronv2.api.API.setup_networks_on_host',
lambda *args, **kwargs: None)
self.test.stub_out(
'nova.network.neutronv2.api.API.migrate_instance_start',
lambda *args, **kwargs: None)
self.test.stub_out(
'nova.network.neutronv2.api.API.migrate_instance_finish',
lambda *args, **kwargs: None)
self.test.stub_out(
'nova.network.security_group.neutron_driver.SecurityGroupAPI.'
'get_instances_security_groups_bindings',

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

@ -58,9 +58,7 @@ from nova import db
from nova import exception
from nova.image import api as image_api
from nova.image import glance
from nova.network import api as network_api
from nova.network import model as network_model
from nova.network.security_group import openstack_driver
from nova import objects
from nova.objects import block_device as block_device_obj
from nova.objects import fields as obj_fields
@ -241,16 +239,19 @@ class BaseTestCase(test.TestCase):
self.stub_out(
'nova.network.neutronv2.api.API.get_instance_nw_info',
fake_get_nw_info)
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)
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.stub_out('nova.network.api.API.allocate_for_instance',
fake_allocate_for_instance)
self.compute_api = compute.API()
# Just to make long lines short
@ -1697,6 +1698,20 @@ class ComputeTestCase(BaseTestCase):
orig_update = self.compute._instance_update
# Make this work for both neutron and nova-network by stubbing out
# allocate_for_instance to return a fake network_info list of VIFs.
ipv4_ip = network_model.IP(version=4, address='192.168.1.100')
ipv4_subnet = network_model.Subnet(ips=[ipv4_ip])
ipv6_ip = network_model.IP(version=6, address='2001:db8:0:1::1')
ipv6_subnet = network_model.Subnet(ips=[ipv6_ip])
network = network_model.Network(
label='test1', subnets=[ipv4_subnet, ipv6_subnet])
network_info = [network_model.VIF(network=network)]
allocate_for_inst_mock = mock.Mock(return_value=network_info)
self.compute.network_api.allocate_for_instance = allocate_for_inst_mock
# mock out deallocate_for_instance since we don't need it now
self.compute.network_api.deallocate_for_instance = mock.Mock()
# Make sure the access_ip_* updates happen in the same DB
# update as the set to ACTIVE.
def _instance_update(self, ctxt, instance_uuid, **kwargs):
@ -1715,8 +1730,7 @@ class ComputeTestCase(BaseTestCase):
instance = instances[0]
self.assertEqual(instance['access_ip_v4'], '192.168.1.100')
self.assertEqual(instance['access_ip_v6'],
'2001:db8:0:1:dcad:beff:feef:1')
self.assertEqual(instance['access_ip_v6'], '2001:db8:0:1::1')
finally:
db.instance_destroy(self.context, instance['uuid'])
@ -2768,14 +2782,13 @@ class ComputeTestCase(BaseTestCase):
@mock.patch.object(compute_manager.ComputeManager,
'_get_instance_block_device_info')
@mock.patch.object(network_api.API, 'get_instance_nw_info')
@mock.patch.object(compute_manager.ComputeManager,
'_notify_about_instance_usage')
@mock.patch.object(compute_manager.ComputeManager, '_instance_update')
@mock.patch.object(db, 'instance_update_and_get_original')
@mock.patch.object(compute_manager.ComputeManager, '_get_power_state')
def _test_reboot(self, soft, mock_get_power, mock_get_orig,
mock_update, mock_notify, mock_get_nw, mock_get_blk,
mock_update, mock_notify, mock_get_blk,
test_delete=False, test_unrescue=False,
fail_reboot=False, fail_running=False):
reboot_type = soft and 'SOFT' or 'HARD'
@ -2842,7 +2855,9 @@ class ComputeTestCase(BaseTestCase):
# Beginning of calls we expect.
self.stub_out('nova.context.RequestContext.elevated', _fake_elevated)
mock_get_blk.return_value = fake_block_dev_info
mock_get_nw = mock.Mock()
mock_get_nw.return_value = fake_nw_model
self.compute.network_api.get_instance_nw_info = mock_get_nw
mock_get_power.side_effect = [fake_power_state1]
mock_get_orig.side_effect = [(None, updated_dbinstance1),
(None, updated_dbinstance1)]
@ -3908,8 +3923,6 @@ class ComputeTestCase(BaseTestCase):
def dummy(*args, **kwargs):
pass
self.stub_out('nova.network.api.API.add_fixed_ip_to_instance',
dummy)
self.stub_out('nova.compute.manager.ComputeManager.'
'inject_network_info', dummy)
self.stub_out('nova.compute.manager.ComputeManager.'
@ -3918,8 +3931,10 @@ class ComputeTestCase(BaseTestCase):
instance = self._create_fake_instance_obj()
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 0)
self.compute.add_fixed_ip_to_instance(self.context, network_id=1,
instance=instance)
with mock.patch.object(self.compute.network_api,
'add_fixed_ip_to_instance', dummy):
self.compute.add_fixed_ip_to_instance(self.context, network_id=1,
instance=instance)
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 2)
self.compute.terminate_instance(self.context, instance, [], [])
@ -3928,8 +3943,6 @@ class ComputeTestCase(BaseTestCase):
def dummy(*args, **kwargs):
pass
self.stub_out('nova.network.api.API.remove_fixed_ip_from_instance',
dummy)
self.stub_out('nova.compute.manager.ComputeManager.'
'inject_network_info', dummy)
self.stub_out('nova.compute.manager.ComputeManager.'
@ -3938,8 +3951,10 @@ class ComputeTestCase(BaseTestCase):
instance = self._create_fake_instance_obj()
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 0)
self.compute.remove_fixed_ip_from_instance(self.context, 1,
instance=instance)
with mock.patch.object(self.compute.network_api,
'remove_fixed_ip_from_instance', dummy):
self.compute.remove_fixed_ip_from_instance(self.context, 1,
instance=instance)
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 2)
self.compute.terminate_instance(self.context, instance, [], [])
@ -4111,25 +4126,28 @@ class ComputeTestCase(BaseTestCase):
image_ref_url = glance.generate_image_url(FAKE_IMAGE_REF)
self.assertEqual(payload['image_ref_url'], image_ref_url)
@mock.patch.object(network_api.API, "allocate_for_instance")
@mock.patch.object(fake.FakeDriver, "macs_for_instance")
def test_run_instance_queries_macs(self, mock_mac, mock_allocate):
def test_run_instance_queries_macs(self, mock_mac):
# run_instance should ask the driver for node mac addresses and pass
# that to the network_api in use.
fake_network.unset_stub_network_methods(self)
instance = self._create_fake_instance_obj()
macs = set(['01:23:45:67:89:ab'])
mock_allocate.return_value = fake_network.fake_get_instance_nw_info(
self, 1, 1)
mock_mac.return_value = macs
self.compute._build_networks_for_instance(self.context, instance,
requested_networks=None, security_groups=None)
with mock.patch.object(self.compute.network_api,
'allocate_for_instance') as mock_allocate:
mock_allocate.return_value = (
fake_network.fake_get_instance_nw_info(self, 1, 1))
self.compute._build_networks_for_instance(self.context, instance,
requested_networks=None, security_groups=None)
security_groups = None if CONF.use_neutron else []
mock_allocate.assert_called_once_with(self.context, instance,
vpn=False, requested_networks=None, macs=macs,
security_groups=[], dhcp_options=None,
security_groups=security_groups, dhcp_options=None,
bind_host_id=self.compute.host)
mock_mac.assert_called_once_with(test.MatchType(instance_obj.Instance))
@ -5765,15 +5783,17 @@ class ComputeTestCase(BaseTestCase):
{'block_device_mapping': []},
mock.ANY, mock.ANY, mock.ANY)
@mock.patch.object(network_api.API, 'setup_networks_on_host')
@mock.patch.object(fake.FakeDriver, 'ensure_filtering_rules_for_instance')
@mock.patch.object(fake.FakeDriver, 'pre_live_migration')
def test_pre_live_migration_works_correctly(self, mock_pre, mock_ensure,
mock_setup):
def test_pre_live_migration_works_correctly(self, mock_pre, mock_ensure):
# Confirm setup_compute_volume is called when volume is mounted.
def stupid(*args, **kwargs):
return fake_network.fake_get_instance_nw_info(self)
self.stub_out('nova.network.api.API.get_instance_nw_info', stupid)
if CONF.use_neutron:
self.stub_out(
'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
instance = self._create_fake_instance_obj({'host': 'dummy'})
@ -5783,9 +5803,12 @@ class ComputeTestCase(BaseTestCase):
migrate_data = {'is_shared_instance_path': False}
mock_pre.return_value = None
ret = self.compute.pre_live_migration(c, instance=instance,
block_migration=False, disk=None,
migrate_data=migrate_data)
with mock.patch.object(self.compute.network_api,
'setup_networks_on_host') as mock_setup:
ret = self.compute.pre_live_migration(c, instance=instance,
block_migration=False,
disk=None,
migrate_data=migrate_data)
self.assertIsNone(ret)
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 2)
msg = fake_notifier.NOTIFICATIONS[0]
@ -5812,13 +5835,12 @@ class ComputeTestCase(BaseTestCase):
@mock.patch.object(fake.FakeDriver, 'get_instance_disk_info')
@mock.patch.object(compute_rpcapi.ComputeAPI, 'pre_live_migration')
@mock.patch.object(objects.BlockDeviceMappingList, 'get_by_instance_uuid')
@mock.patch.object(network_api.API, 'setup_networks_on_host')
@mock.patch.object(compute_rpcapi.ComputeAPI, 'remove_volume_connection')
@mock.patch.object(compute_rpcapi.ComputeAPI,
'rollback_live_migration_at_destination')
@mock.patch('nova.objects.Migration.save')
def test_live_migration_exception_rolls_back(self, mock_save,
mock_rollback, mock_remove, mock_setup,
mock_rollback, mock_remove,
mock_get_uuid, mock_pre, mock_get_disk):
# Confirm exception when pre_live_migration fails.
c = context.get_admin_context()
@ -5853,11 +5875,13 @@ class ComputeTestCase(BaseTestCase):
# start test
migration = objects.Migration()
self.assertRaises(test.TestingException,
self.compute.live_migration,
c, dest=dest_host, block_migration=True,
instance=instance, migration=migration,
migrate_data=migrate_data)
with mock.patch.object(self.compute.network_api,
'setup_networks_on_host') as mock_setup:
self.assertRaises(test.TestingException,
self.compute.live_migration,
c, dest=dest_host, block_migration=True,
instance=instance, migration=migration,
migrate_data=migrate_data)
instance.refresh()
self.assertEqual('src_host', instance.host)
@ -5879,16 +5903,14 @@ class ComputeTestCase(BaseTestCase):
migrate_data_obj.XenapiLiveMigrateData))
@mock.patch.object(compute_rpcapi.ComputeAPI, 'pre_live_migration')
@mock.patch.object(network_api.API, 'migrate_instance_start')
@mock.patch.object(compute_rpcapi.ComputeAPI,
'post_live_migration_at_destination')
@mock.patch.object(network_api.API, 'setup_networks_on_host')
@mock.patch.object(compute_manager.InstanceEvents,
'clear_events_for_instance')
@mock.patch.object(compute_utils, 'EventReporter')
@mock.patch('nova.objects.Migration.save')
def test_live_migration_works_correctly(self, mock_save, mock_event,
mock_clear, mock_setup, mock_post, mock_migrate, mock_pre):
mock_clear, mock_post, mock_pre):
# Confirm live_migration() works as expected correctly.
# creating instance testdata
c = context.get_admin_context()
@ -5903,11 +5925,19 @@ class ComputeTestCase(BaseTestCase):
# start test
migration = objects.Migration()
ret = self.compute.live_migration(c, dest=dest,
instance=instance,
block_migration=False,
migration=migration,
migrate_data=migrate_data)
with test.nested(
mock.patch.object(
self.compute.network_api, 'migrate_instance_start'),
mock.patch.object(
self.compute.network_api, 'setup_networks_on_host')
) as (
mock_migrate, mock_setup
):
ret = self.compute.live_migration(c, dest=dest,
instance=instance,
block_migration=False,
migration=migration,
migrate_data=migrate_data)
self.assertIsNone(ret)
mock_event.assert_called_with(
@ -5925,14 +5955,12 @@ class ComputeTestCase(BaseTestCase):
mock_clear.assert_called_once_with(mock.ANY)
@mock.patch.object(fake.FakeDriver, 'unfilter_instance')
@mock.patch.object(network_api.API, 'migrate_instance_start')
@mock.patch.object(compute_rpcapi.ComputeAPI,
'post_live_migration_at_destination')
@mock.patch.object(network_api.API, 'setup_networks_on_host')
@mock.patch.object(compute_manager.InstanceEvents,
'clear_events_for_instance')
def test_post_live_migration_no_shared_storage_working_correctly(self,
mock_clear, mock_setup, mock_post, mock_migrate, mock_unfilter):
mock_clear, mock_post, mock_unfilter):
"""Confirm post_live_migration() works correctly as expected
for non shared storage migration.
"""
@ -5964,8 +5992,16 @@ class ComputeTestCase(BaseTestCase):
is_shared_block_storage=False,
block_migration=False)
self.compute._post_live_migration(c, instance, dest,
migrate_data=migrate_data)
with test.nested(
mock.patch.object(
self.compute.network_api, 'migrate_instance_start'),
mock.patch.object(
self.compute.network_api, 'setup_networks_on_host')
) as (
mock_migrate, mock_setup
):
self.compute._post_live_migration(c, instance, dest,
migrate_data=migrate_data)
self.assertIn('cleanup', result)
self.assertTrue(result['cleanup'])
@ -6185,21 +6221,22 @@ class ComputeTestCase(BaseTestCase):
self.assertEqual('fake', migration.status)
migration.save.assert_called_once_with()
@mock.patch.object(network_api.API, 'setup_networks_on_host')
@mock.patch.object(fake.FakeDriver,
'rollback_live_migration_at_destination')
def test_rollback_live_migration_at_destination_correctly(self,
mock_rollback, mock_setup):
mock_rollback):
# creating instance testdata
c = context.get_admin_context()
instance = self._create_fake_instance_obj({'host': 'dummy'})
fake_notifier.NOTIFICATIONS = []
# start test
ret = self.compute.rollback_live_migration_at_destination(c,
instance=instance,
destroy_disks=True,
migrate_data=None)
with mock.patch.object(self.compute.network_api,
'setup_networks_on_host') as mock_setup:
ret = self.compute.rollback_live_migration_at_destination(c,
instance=instance,
destroy_disks=True,
migrate_data=None)
self.assertIsNone(ret)
self.assertEqual(len(fake_notifier.NOTIFICATIONS), 2)
msg = fake_notifier.NOTIFICATIONS[0]
@ -6216,19 +6253,21 @@ class ComputeTestCase(BaseTestCase):
'block_device_mapping': []},
destroy_disks=True, migrate_data=None)
@mock.patch('nova.network.api.API.setup_networks_on_host',
side_effect=test.TestingException)
@mock.patch('nova.virt.driver.ComputeDriver.'
'rollback_live_migration_at_destination')
@mock.patch('nova.objects.migrate_data.LiveMigrateData.'
'detect_implementation')
def test_rollback_live_migration_at_destination_network_fails(
self, mock_detect, mock_rollback, net_mock):
self, mock_detect, mock_rollback):
c = context.get_admin_context()
instance = self._create_fake_instance_obj()
self.assertRaises(test.TestingException,
self.compute.rollback_live_migration_at_destination,
c, instance, destroy_disks=True, migrate_data={})
with mock.patch.object(self.compute.network_api,
'setup_networks_on_host',
side_effect=test.TestingException):
self.assertRaises(
test.TestingException,
self.compute.rollback_live_migration_at_destination,
c, instance, destroy_disks=True, migrate_data={})
mock_rollback.assert_called_once_with(
c, instance, mock.ANY, mock.ANY,
destroy_disks=True,
@ -6619,8 +6658,13 @@ class ComputeTestCase(BaseTestCase):
fake_instance_get_all_by_host)
self.stub_out('nova.db.instance_get_by_uuid',
fake_instance_get_by_uuid)
self.stub_out('nova.network.api.API.get_instance_nw_info',
if CONF.use_neutron:
self.stub_out(
'nova.network.neutronv2.api.API.get_instance_nw_info',
fake_get_instance_nw_info)
else:
self.stub_out('nova.network.api.API.get_instance_nw_info',
fake_get_instance_nw_info)
# Make an instance appear to be still Building
instances[0]['vm_state'] = vm_states.BUILDING
@ -6935,7 +6979,6 @@ class ComputeTestCase(BaseTestCase):
@mock.patch.object(compute_manager.ComputeManager,
'_get_instances_on_driver')
@mock.patch.object(network_api.API, 'get_instance_nw_info')
@mock.patch.object(compute_manager.ComputeManager,
'_get_instance_block_device_info')
@mock.patch.object(compute_manager.ComputeManager,
@ -6945,7 +6988,7 @@ class ComputeTestCase(BaseTestCase):
@mock.patch('nova.objects.Migration.save')
def test_destroy_evacuated_instance_on_shared_storage(self, mock_save,
mock_get_filter, mock_destroy, mock_is_inst, mock_get_blk,
mock_get_nw, mock_get_inst):
mock_get_inst):
fake_context = context.get_admin_context()
# instances in central db
@ -6967,11 +7010,13 @@ class ComputeTestCase(BaseTestCase):
mock_get_filter.return_value = [migration]
instances.append(evacuated_instance)
mock_get_inst.return_value = instances
mock_get_nw.return_value = 'fake_network_info'
mock_get_blk.return_value = 'fake_bdi'
mock_is_inst.return_value = True
self.compute._destroy_evacuated_instances(fake_context)
with mock.patch.object(
self.compute.network_api, 'get_instance_nw_info',
return_value='fake_network_info') as mock_get_nw:
self.compute._destroy_evacuated_instances(fake_context)
mock_get_filter.assert_called_once_with(fake_context,
{'source_compute': self.compute.host,
@ -6987,7 +7032,6 @@ class ComputeTestCase(BaseTestCase):
@mock.patch.object(compute_manager.ComputeManager,
'_get_instances_on_driver')
@mock.patch.object(network_api.API, 'get_instance_nw_info')
@mock.patch.object(compute_manager.ComputeManager,
'_get_instance_block_device_info')
@mock.patch.object(fake.FakeDriver,
@ -7001,7 +7045,7 @@ class ComputeTestCase(BaseTestCase):
@mock.patch('nova.objects.Migration.save')
def test_destroy_evacuated_instance_with_disks(self, mock_save,
mock_get_filter, mock_destroy, mock_check_clean, mock_check,
mock_check_local, mock_get_blk, mock_get_nw, mock_get_drv):
mock_check_local, mock_get_blk, mock_get_drv):
fake_context = context.get_admin_context()
# instances in central db
@ -7023,12 +7067,14 @@ class ComputeTestCase(BaseTestCase):
mock_get_filter.return_value = [migration]
instances.append(evacuated_instance)
mock_get_drv.return_value = instances
mock_get_nw.return_value = 'fake_network_info'
mock_get_blk.return_value = 'fake-bdi'
mock_check_local.return_value = {'filename': 'tmpfilename'}
mock_check.return_value = False
self.compute._destroy_evacuated_instances(fake_context)
with mock.patch.object(
self.compute.network_api, 'get_instance_nw_info',
return_value='fake_network_info') as mock_get_nw:
self.compute._destroy_evacuated_instances(fake_context)
mock_get_drv.assert_called_once_with(fake_context, {'deleted': False})
mock_get_nw.assert_called_once_with(fake_context, evacuated_instance)
@ -7046,7 +7092,6 @@ class ComputeTestCase(BaseTestCase):
@mock.patch.object(compute_manager.ComputeManager,
'_get_instances_on_driver')
@mock.patch.object(network_api.API, 'get_instance_nw_info')
@mock.patch.object(compute_manager.ComputeManager,
'_get_instance_block_device_info')
@mock.patch.object(fake.FakeDriver, 'check_instance_shared_storage_local')
@ -7059,7 +7104,7 @@ class ComputeTestCase(BaseTestCase):
@mock.patch('nova.objects.Migration.save')
def test_destroy_evacuated_instance_not_implemented(self, mock_save,
mock_get_filter, mock_destroy, mock_check_clean, mock_check,
mock_check_local, mock_get_blk, mock_get_nw, mock_get_inst):
mock_check_local, mock_get_blk, mock_get_inst):
fake_context = context.get_admin_context()
# instances in central db
@ -7081,11 +7126,13 @@ class ComputeTestCase(BaseTestCase):
mock_get_filter.return_value = [migration]
instances.append(evacuated_instance)
mock_get_inst.return_value = instances
mock_get_nw.return_value = 'fake_network_info'
mock_get_blk.return_value = 'fake_bdi'
mock_check_local.side_effect = NotImplementedError
self.compute._destroy_evacuated_instances(fake_context)
with mock.patch.object(
self.compute.network_api, 'get_instance_nw_info',
return_value='fake_network_info') as mock_get_nw:
self.compute._destroy_evacuated_instances(fake_context)
mock_get_inst.assert_called_once_with(fake_context, {'deleted': False})
mock_get_nw.assert_called_once_with(fake_context, evacuated_instance)
@ -7205,20 +7252,20 @@ class ComputeTestCase(BaseTestCase):
def _noop(*args, **kwargs):
pass
self.stub_out('nova.network.api.API.'
'add_fixed_ip_to_instance', _noop)
self.stub_out('nova.network.api.API.'
'remove_fixed_ip_from_instance', _noop)
instance = self._create_fake_instance_obj()
updated_at_1 = instance['updated_at']
self.compute.add_fixed_ip_to_instance(self.context, 'fake', instance)
with mock.patch.object(
self.compute.network_api, 'add_fixed_ip_to_instance', _noop):
self.compute.add_fixed_ip_to_instance(
self.context, 'fake', instance)
updated_at_2 = db.instance_get_by_uuid(self.context,
instance['uuid'])['updated_at']
self.compute.remove_fixed_ip_from_instance(self.context, 'fake',
instance)
with mock.patch.object(
self.compute.network_api, 'remove_fixed_ip_from_instance', _noop):
self.compute.remove_fixed_ip_from_instance(
self.context, 'fake', instance)
updated_at_3 = db.instance_get_by_uuid(self.context,
instance['uuid'])['updated_at']
@ -7732,8 +7779,11 @@ class ComputeAPITestCase(BaseTestCase):
self.useFixture(fixtures.SpawnIsSynchronousFixture())
self.stub_out('nova.network.api.API.get_instance_nw_info',
fake_get_nw_info)
self.security_group_api = (
openstack_driver.get_openstack_security_group_driver())
# NOTE(mriedem): Everything in here related to the security group API
# is written for nova-network and using the database. Neutron-specific
# security group API tests are covered in
# nova.tests.unit.network.security_group.test_neutron_driver.
self.security_group_api = compute_api.SecurityGroupAPI()
self.compute_api = compute.API(
security_group_api=self.security_group_api)
@ -7770,6 +7820,16 @@ class ComputeAPITestCase(BaseTestCase):
self.compute_api.compute_task_api.rebuild_instance = \
self.rebuild_instance_mock
# Assume that we're always OK for network quota.
def fake_validate_networks(context, requested_networks, num_instances):
return num_instances
validate_nets_patch = mock.patch.object(
self.compute_api.network_api, 'validate_networks',
fake_validate_networks)
validate_nets_patch.start()
self.addCleanup(validate_nets_patch.stop)
def _run_instance(self, params=None):
instance = self._create_fake_instance_obj(params, services=True)
instance_uuid = instance['uuid']
@ -7981,12 +8041,16 @@ class ComputeAPITestCase(BaseTestCase):
groups_for_instance = db.security_group_get_by_instance(
self.context, ref[0]['uuid'])
self.assertEqual(1, len(groups_for_instance))
self.assertEqual(group.id, groups_for_instance[0].id)
group_with_instances = db.security_group_get(self.context,
group.id,
columns_to_join=['instances'])
self.assertEqual(1, len(group_with_instances.instances))
# For Neutron we don't store the security groups in the nova database.
if CONF.use_neutron:
self.assertEqual(0, len(groups_for_instance))
else:
self.assertEqual(1, len(groups_for_instance))
self.assertEqual(group.id, groups_for_instance[0].id)
group_with_instances = db.security_group_get(self.context,
group.id,
columns_to_join=['instances'])
self.assertEqual(1, len(group_with_instances.instances))
def test_create_instance_with_invalid_security_group_raises(self):
instance_type = flavors.get_default_flavor()
@ -9195,14 +9259,14 @@ class ComputeAPITestCase(BaseTestCase):
def test_add_remove_fixed_ip(self):
instance = self._create_fake_instance_obj(params={'host': CONF.host})
self.stub_out('nova.network.api.API.deallocate_for_instance',
lambda *a, **kw: None)
self.compute_api.add_fixed_ip(self.context, instance, '1')
self.compute_api.remove_fixed_ip(self.context,
instance, '192.168.1.1')
with mock.patch.object(self.compute_api, '_lookup_instance',
return_value=instance):
self.compute_api.delete(self.context, instance)
with mock.patch.object(self.compute_api.network_api,
'deallocate_for_instance'):
self.compute_api.delete(self.context, instance)
def test_attach_volume_invalid(self):
instance = fake_instance.fake_instance_obj(None, **{
@ -9532,8 +9596,7 @@ class ComputeAPITestCase(BaseTestCase):
self.compute_api.get_console_output,
self.context, instance)
@mock.patch.object(network_api.API, 'allocate_port_for_instance')
def test_attach_interface(self, mock_allocate):
def test_attach_interface(self):
new_type = flavors.get_flavor_by_flavor_id('4')
instance = objects.Instance(image_ref=uuids.image_instance,
system_metadata={},
@ -9543,7 +9606,8 @@ class ComputeAPITestCase(BaseTestCase):
network_id = nwinfo[0]['network']['id']
port_id = nwinfo[0]['id']
req_ip = '1.2.3.4'
mock_allocate.return_value = nwinfo
mock_allocate = mock.Mock(return_value=nwinfo)
self.compute.network_api.allocate_port_for_instance = mock_allocate
with mock.patch.dict(self.compute.driver.capabilities,
supports_attach_interface=True):