"""For this model this method will be delegated to vswitch plugin."""
pass
+ def _check_nexus_net_create_needed(self, new_port, old_port):
+ """Check if nexus plugin should be invoked for net_create.
+
+ In the following cases, the plugin should be invoked:
+ -- a port is attached to a VM instance. The old host id is None
+ -- VM migration. The old host id has a valid value
+
+ When the plugin needs to be invoked, return the old_host_id,
+ and a list of calling arguments.
+ Otherwise, return '' for old host id and an empty list
+ """
+ old_device_id = old_port['device_id']
+ new_device_id = new_port.get('device_id')
+ new_host_id = self._get_port_host_id_from_bindings(new_port)
+ tenant_id = old_port['tenant_id']
+ net_id = old_port['network_id']
+ old_host_id = self._get_port_host_id_from_bindings(old_port)
+
+ LOG.debug(_("tenant_id: %(tid)s, net_id: %(nid)s, "
+ "old_device_id: %(odi)s, new_device_id: %(ndi)s, "
+ "old_host_id: %(ohi)s, new_host_id: %(nhi)s, "
+ "old_device_owner: %(odo)s, new_device_owner: %(ndo)s"),
+ {'tid': tenant_id, 'nid': net_id,
+ 'odi': old_device_id, 'ndi': new_device_id,
+ 'ohi': old_host_id, 'nhi': new_host_id,
+ 'odo': old_port.get('device_owner'),
+ 'ndo': new_port.get('device_owner')})
+
+ # A port is attached to an instance
+ if (new_device_id and not old_device_id and new_host_id and
+ self._check_valid_port_device_owner(new_port)):
+ return '', [tenant_id, net_id, new_device_id, new_host_id]
+
+ # An instance is being migrated
+ if (old_device_id and old_host_id and new_host_id != old_host_id and
+ self._check_valid_port_device_owner(old_port)):
+ return old_host_id, [tenant_id, net_id, old_device_id, new_host_id]
+
+ # no need to invoke the plugin
+ return '', []
+
def update_port(self, context, id, port):
"""Update port.
"""
LOG.debug(_("update_port() called"))
old_port = self.get_port(context, id)
- old_device = old_port['device_id']
args = [context, id, port]
ovs_output = self._invoke_plugin_per_device(const.VSWITCH_PLUGIN,
self._func_name(),
args)
- net_id = old_port['network_id']
- instance_id = ''
- if 'device_id' in port['port']:
- instance_id = port['port']['device_id']
-
- # Check if there's a new device_id
try:
- host_id = self._get_port_host_id_from_bindings(port['port'])
- if (instance_id and not old_device and host_id and
- self._check_valid_port_device_owner(port['port'])):
- tenant_id = old_port['tenant_id']
- self._invoke_nexus_for_net_create(
- context, tenant_id, net_id, instance_id, host_id)
+ # Check if the nexus plugin needs to be invoked
+ old_host_id, create_args = self._check_nexus_net_create_needed(
+ port['port'], old_port)
+
+ # In the case of migration, invoke it to remove
+ # the previous port binding
+ if old_host_id:
+ vlan_id = self._get_segmentation_id(old_port['network_id'])
+ delete_args = [old_port['device_id'], vlan_id]
+ self._invoke_plugin_per_device(const.NEXUS_PLUGIN,
+ "delete_port",
+ delete_args)
+
+ # Invoke the Nexus plugin to create a net and/or new port binding
+ if create_args:
+ self._invoke_nexus_for_net_create(context, *create_args)
return ovs_output[0]
except Exception:
import webob.exc as wexc
from neutron.api import extensions
+from neutron.api.v2 import attributes
from neutron.api.v2 import base
from neutron.common import exceptions as q_exc
from neutron import context
VLAN_START = 1000
VLAN_END = 1100
COMP_HOST_NAME = 'testhost'
+COMP_HOST_NAME_2 = 'testhost_2'
NEXUS_IP_ADDR = '1.1.1.1'
NEXUS_DEV_ID = 'NEXUS_SWITCH'
NEXUS_USERNAME = 'admin'
NEXUS_PASSWORD = 'mySecretPassword'
NEXUS_SSH_PORT = 22
NEXUS_INTERFACE = '1/1'
+NEXUS_INTERFACE_2 = '1/2'
+NEXUS_PORT_1 = 'ethernet:1/1'
+NEXUS_PORT_2 = 'ethernet:1/2'
NETWORK_NAME = 'test_network'
CIDR_1 = '10.0.0.0/24'
CIDR_2 = '10.0.1.0/24'
(NEXUS_DEV_ID, NEXUS_IP_ADDR, 'password'): NEXUS_PASSWORD,
(NEXUS_DEV_ID, NEXUS_IP_ADDR, 'ssh_port'): NEXUS_SSH_PORT,
(NEXUS_DEV_ID, NEXUS_IP_ADDR, COMP_HOST_NAME): NEXUS_INTERFACE,
+ (NEXUS_DEV_ID, NEXUS_IP_ADDR, COMP_HOST_NAME_2): NEXUS_INTERFACE_2,
}
nexus_patch = mock.patch.dict(cisco_config.device_dictionary,
nexus_config)
NEXUS_IP_ADDR)
self.assertEqual(start_rows, end_rows)
+ def test_model_update_port_attach(self):
+ """Test the model for update_port in attaching to an instance.
+
+ Mock the routines that call into the plugin code, and make sure they
+ are called with correct arguments.
+
+ """
+ with contextlib.nested(
+ self.port(),
+ mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
+ '_invoke_plugin_per_device'),
+ mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
+ '_invoke_nexus_for_net_create')
+ ) as (port, invoke_plugin_per_device, invoke_nexus_for_net_create):
+ data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME,
+ 'device_id': DEVICE_ID_1,
+ 'device_owner': DEVICE_OWNER}}
+
+ req = self.new_update_request('ports', data, port['port']['id'])
+ # Note, due to mocking out the two model routines, response won't
+ # contain any useful data
+ req.get_response(self.api)
+
+ # Note that call_args_list is used instead of
+ # assert_called_once_with which requires exact match of arguments.
+ # This is because the mocked routines contain variable number of
+ # arguments and/or dynamic objects.
+ self.assertEqual(invoke_plugin_per_device.call_count, 1)
+ self.assertEqual(
+ invoke_plugin_per_device.call_args_list[0][0][0:2],
+ (const.VSWITCH_PLUGIN, 'update_port'))
+ self.assertEqual(invoke_nexus_for_net_create.call_count, 1)
+ self.assertEqual(
+ invoke_nexus_for_net_create.call_args_list[0][0][1:],
+ (port['port']['tenant_id'], port['port']['network_id'],
+ data['port']['device_id'],
+ data['port'][portbindings.HOST_ID],))
+
+ def test_model_update_port_migrate(self):
+ """Test the model for update_port in migrating an instance.
+
+ Mock the routines that call into the plugin code, and make sure they
+ are called with correct arguments.
+
+ """
+ arg_list = (portbindings.HOST_ID,)
+ data = {portbindings.HOST_ID: COMP_HOST_NAME,
+ 'device_id': DEVICE_ID_1,
+ 'device_owner': DEVICE_OWNER}
+
+ with contextlib.nested(
+ self.port(arg_list=arg_list, **data),
+ mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
+ '_invoke_plugin_per_device'),
+ mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
+ '_invoke_nexus_for_net_create')
+ ) as (port, invoke_plugin_per_device, invoke_nexus_for_net_create):
+ data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME_2}}
+ req = self.new_update_request('ports', data, port['port']['id'])
+ # Note, due to mocking out the two model routines, response won't
+ # contain any useful data
+ req.get_response(self.api)
+
+ # Note that call_args_list is used instead of
+ # assert_called_once_with which requires exact match of arguments.
+ # This is because the mocked routines contain variable number of
+ # arguments and/or dynamic objects.
+ self.assertEqual(invoke_plugin_per_device.call_count, 2)
+ self.assertEqual(
+ invoke_plugin_per_device.call_args_list[0][0][0:2],
+ (const.VSWITCH_PLUGIN, 'update_port'))
+ self.assertEqual(
+ invoke_plugin_per_device.call_args_list[1][0][0:2],
+ (const.NEXUS_PLUGIN, 'delete_port'))
+ self.assertEqual(invoke_nexus_for_net_create.call_count, 1)
+ self.assertEqual(
+ invoke_nexus_for_net_create.call_args_list[0][0][1:],
+ (port['port']['tenant_id'], port['port']['network_id'],
+ port['port']['device_id'],
+ data['port'][portbindings.HOST_ID],))
+
+ def test_model_update_port_net_create_not_needed(self):
+ """Test the model for update_port when no action is needed.
+
+ Mock the routines that call into the plugin code, and make sure that
+ VSWITCH plugin is called with correct arguments, while NEXUS plugin is
+ not called at all.
+
+ """
+ arg_list = (portbindings.HOST_ID,)
+ data = {portbindings.HOST_ID: COMP_HOST_NAME,
+ 'device_id': DEVICE_ID_1,
+ 'device_owner': DEVICE_OWNER}
+
+ with contextlib.nested(
+ self.port(arg_list=arg_list, **data),
+ mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
+ '_invoke_plugin_per_device'),
+ mock.patch.object(virt_phy_sw_v2.VirtualPhysicalSwitchModelV2,
+ '_invoke_nexus_for_net_create')
+ ) as (port, invoke_plugin_per_device, invoke_nexus_for_net_create):
+ data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME,
+ 'device_id': DEVICE_ID_1,
+ 'device_owner': DEVICE_OWNER}}
+ req = self.new_update_request('ports', data, port['port']['id'])
+ # Note, due to mocking out the two model routines, response won't
+ # contain any useful data
+ req.get_response(self.api)
+
+ # Note that call_args_list is used instead of
+ # assert_called_once_with which requires exact match of arguments.
+ # This is because the mocked routines contain variable number of
+ # arguments and/or dynamic objects.
+ self.assertEqual(invoke_plugin_per_device.call_count, 1)
+ self.assertEqual(
+ invoke_plugin_per_device.call_args_list[0][0][0:2],
+ (const.VSWITCH_PLUGIN, 'update_port'))
+ self.assertFalse(invoke_nexus_for_net_create.called)
+
+ def verify_portbinding(self, host_id1, host_id2,
+ vlan, device_id, binding_port):
+ """Verify a port binding entry in the DB is correct."""
+ self.assertEqual(host_id1, host_id2)
+ pb = nexus_db_v2.get_nexusvm_bindings(vlan, device_id)
+ self.assertEqual(len(pb), 1)
+ self.assertEqual(pb[0].port_id, binding_port)
+ self.assertEqual(pb[0].switch_ip, NEXUS_IP_ADDR)
+
+ def test_db_update_port_attach(self):
+ """Test DB for update_port in attaching to an instance.
+
+ Query DB for the port binding entry corresponding to the search key
+ (vlan, device_id), and make sure that it's bound to correct switch port
+
+ """
+ with self.port() as port:
+ data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME,
+ 'device_id': DEVICE_ID_1,
+ 'device_owner': DEVICE_OWNER}}
+
+ req = self.new_update_request('ports', data, port['port']['id'])
+ res = self.deserialize(self.fmt, req.get_response(self.api))
+ ctx = context.get_admin_context()
+ net = self._show('networks', res['port']['network_id'],
+ neutron_context=ctx)['network']
+ self.assertTrue(attributes.is_attr_set(
+ net.get(provider.SEGMENTATION_ID)))
+ vlan = net[provider.SEGMENTATION_ID]
+ self.assertEqual(vlan, VLAN_START)
+ self.verify_portbinding(res['port'][portbindings.HOST_ID],
+ data['port'][portbindings.HOST_ID],
+ vlan,
+ data['port']['device_id'],
+ NEXUS_PORT_1)
+
+ def test_db_update_port_migrate(self):
+ """Test DB for update_port in migrating an instance.
+
+ Query DB for the port binding entry corresponding to the search key
+ (vlan, device_id), and make sure that it's bound to correct switch port
+ before and after the migration.
+
+ """
+ arg_list = (portbindings.HOST_ID,)
+ data = {portbindings.HOST_ID: COMP_HOST_NAME,
+ 'device_id': DEVICE_ID_1,
+ 'device_owner': DEVICE_OWNER}
+
+ with self.port(arg_list=arg_list, **data) as port:
+ ctx = context.get_admin_context()
+ net = self._show('networks', port['port']['network_id'],
+ neutron_context=ctx)['network']
+ self.assertTrue(attributes.is_attr_set(
+ net.get(provider.SEGMENTATION_ID)))
+ vlan = net[provider.SEGMENTATION_ID]
+ self.assertEqual(vlan, VLAN_START)
+ self.verify_portbinding(port['port'][portbindings.HOST_ID],
+ data[portbindings.HOST_ID],
+ vlan,
+ data['device_id'],
+ NEXUS_PORT_1)
+
+ new_data = {'port': {portbindings.HOST_ID: COMP_HOST_NAME_2}}
+ req = self.new_update_request('ports',
+ new_data, port['port']['id'])
+ res = self.deserialize(self.fmt, req.get_response(self.api))
+ self.verify_portbinding(res['port'][portbindings.HOST_ID],
+ new_data['port'][portbindings.HOST_ID],
+ vlan,
+ data['device_id'],
+ NEXUS_PORT_2)
+
class TestCiscoNetworksV2(CiscoNetworkPluginV2TestCase,
test_db_plugin.TestNetworksV2):
Code Review / openstack-build / neutron-build.git / commitdiff