from neutron.common import constants as n_const
from neutron.extensions import portbindings
-from neutron.openstack.common import excutils
from neutron.openstack.common import log as logging
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers.cisco import config as conf
self.credentials = {}
self.driver = nexus_network_driver.CiscoNexusDriver()
- # Initialize credential store after database initialization
+ # Initialize credential store after database initialization.
cred.Store.initialize()
def _valid_network_segment(self, segment):
def _is_status_active(self, port):
return port['status'] == n_const.PORT_STATUS_ACTIVE
- def _get_credential(self, nexus_ip):
- """Return credential information for a given Nexus IP address.
-
- If credential doesn't exist then also add to local dictionary.
- """
- if nexus_ip not in self.credentials:
- _nexus_username = cred.Store.get_username(nexus_ip)
- _nexus_password = cred.Store.get_password(nexus_ip)
- self.credentials[nexus_ip] = {
- 'username': _nexus_username,
- 'password': _nexus_password
- }
- return self.credentials[nexus_ip]
-
- def _manage_port(self, vlan_name, vlan_id, host, instance):
- """Called during create and update port events.
-
- Create a VLAN in the appropriate switch/port and configure the
- appropriate interfaces for this VLAN.
- """
-
- # Grab the switch IP and port for this host
+ def _get_switch_info(self, host_id):
for switch_ip, attr in self._nexus_switches:
- if str(attr) == str(host):
+ if str(attr) == str(host_id):
port_id = self._nexus_switches[switch_ip, attr]
- break
+ return port_id, switch_ip
else:
- raise excep.NexusComputeHostNotConfigured(host=host)
-
- # Check if this network is already in the DB
- vlan_created = False
- vlan_trunked = False
+ raise excep.NexusComputeHostNotConfigured(host=host_id)
- try:
- nxos_db.get_port_vlan_switch_binding(port_id, vlan_id, switch_ip)
- except excep.NexusPortBindingNotFound:
- # Check for vlan/switch binding
- try:
- nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
- except excep.NexusPortBindingNotFound:
- # Create vlan and trunk vlan on the port
- LOG.debug(_("Nexus: create & trunk vlan %s"), vlan_name)
- self.driver.create_and_trunk_vlan(switch_ip, vlan_id,
- vlan_name, port_id)
- vlan_created = True
- vlan_trunked = True
- else:
- # Only trunk vlan on the port
- LOG.debug(_("Nexus: trunk vlan %s"), vlan_name)
- self.driver.enable_vlan_on_trunk_int(switch_ip, vlan_id,
- port_id)
- vlan_trunked = True
+ def _configure_nxos_db(self, context, vlan_id, device_id, host_id):
+ """Create the nexus database entry.
- try:
- nxos_db.add_nexusport_binding(port_id, str(vlan_id),
- switch_ip, instance)
- except Exception:
- with excutils.save_and_reraise_exception():
- # Add binding failed, roll back any vlan creation/enabling
- if vlan_created and vlan_trunked:
- LOG.debug(_("Nexus: delete & untrunk vlan %s"), vlan_name)
- self.driver.delete_and_untrunk_vlan(switch_ip, vlan_id,
- port_id)
- elif vlan_created:
- LOG.debug(_("Nexus: delete vlan %s"), vlan_name)
- self.driver.delete_vlan(switch_ip, vlan_id)
- elif vlan_trunked:
- LOG.debug(_("Nexus: untrunk vlan %s"), vlan_name)
- self.driver.disable_vlan_on_trunk_int(switch_ip, vlan_id,
- port_id)
-
- def _invoke_nexus_on_port_event(self, context):
- vlan_id = self._get_vlanid(context)
- host_id = context.current.get(portbindings.HOST_ID)
+ Called during update precommit port event.
- if vlan_id and host_id:
- vlan_name = cfg.CONF.ml2_cisco.vlan_name_prefix + str(vlan_id)
- instance_id = context.current.get('device_id')
- self._manage_port(vlan_name, vlan_id, host_id, instance_id)
- else:
- LOG.debug(_("Vlan ID %(vlan_id)s or Host ID %(host_id)s missing."),
- {'vlan_id': vlan_id, 'host_id': host_id})
+ """
+ port_id, switch_ip = self._get_switch_info(host_id)
+ nxos_db.add_nexusport_binding(port_id, str(vlan_id), switch_ip,
+ device_id)
- def update_port_postcommit(self, context):
- """Update port post-database commit event."""
- port = context.current
+ def _configure_switch_entry(self, context, vlan_id, device_id, host_id):
+ """Create a nexus switch entry.
- if self._is_deviceowner_compute(port) and self._is_status_active(port):
- self._invoke_nexus_on_port_event(context)
+ if needed, create a VLAN in the appropriate switch/port and
+ configure the appropriate interfaces for this VLAN.
- def delete_port_precommit(self, context):
- """Delete port pre-database commit event.
+ Called during update postcommit port event.
- Delete port bindings from the database and scan whether the network
- is still required on the interfaces trunked.
"""
+ port_id, switch_ip = self._get_switch_info(host_id)
+ vlan_name = cfg.CONF.ml2_cisco.vlan_name_prefix + str(vlan_id)
+
+ # Check to see if this is the first binding to use this vlan on the
+ # switch/port. Configure switch accordingly.
+ bindings = nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
+ if len(bindings) == 1:
+ LOG.debug(_("Nexus: create & trunk vlan %s"), vlan_name)
+ self.driver.create_and_trunk_vlan(switch_ip, vlan_id, vlan_name,
+ port_id)
+ else:
+ LOG.debug(_("Nexus: trunk vlan %s"), vlan_name)
+ self.driver.enable_vlan_on_trunk_int(switch_ip, vlan_id, port_id)
- if not self._is_deviceowner_compute(context.current):
- return
-
- port = context.current
- device_id = port['device_id']
- vlan_id = self._get_vlanid(context)
+ def _delete_nxos_db(self, context, vlan_id, device_id, host_id):
+ """Delete the nexus database entry.
- if not vlan_id or not device_id:
- return
+ Called during delete precommit port event.
- # Delete DB row for this port
+ """
try:
row = nxos_db.get_nexusvm_binding(vlan_id, device_id)
+ nxos_db.remove_nexusport_binding(row.port_id, row.vlan_id,
+ row.switch_ip, row.instance_id)
except excep.NexusPortBindingNotFound:
return
- switch_ip = row.switch_ip
- nexus_port = None
- if row.port_id != 'router':
- nexus_port = row.port_id
+ def _delete_switch_entry(self, context, vlan_id, device_id, host_id):
+ """Delete the nexus switch entry.
+
+ By accessing the current db entries determine if switch
+ configuration can be removed.
+
+ Called during update postcommit port event.
- nxos_db.remove_nexusport_binding(row.port_id, row.vlan_id,
- row.switch_ip, row.instance_id)
+ """
+ port_id, switch_ip = self._get_switch_info(host_id)
- # Check for any other bindings with the same vlan_id and switch_ip
+ # if there are no remaining db entries using this vlan on this nexus
+ # switch port then remove vlan from the switchport trunk.
try:
- nxos_db.get_nexusvlan_binding(row.vlan_id, row.switch_ip)
+ nxos_db.get_port_vlan_switch_binding(port_id, vlan_id, switch_ip)
except excep.NexusPortBindingNotFound:
+ self.driver.disable_vlan_on_trunk_int(switch_ip, vlan_id, port_id)
+
+ # if there are no remaining db entries using this vlan on this
+ # nexus switch then remove the vlan.
try:
- # Delete this vlan from this switch
- if nexus_port:
- self.driver.disable_vlan_on_trunk_int(switch_ip,
- row.vlan_id,
- nexus_port)
- self.driver.delete_vlan(switch_ip, row.vlan_id)
- except Exception:
- # The delete vlan operation on the Nexus failed,
- # so this delete_port request has failed. For
- # consistency, roll back the Nexus database to what
- # it was before this request.
- with excutils.save_and_reraise_exception():
- nxos_db.add_nexusport_binding(row.port_id,
- row.vlan_id,
- row.switch_ip,
- row.instance_id)
+ nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
+ except excep.NexusPortBindingNotFound:
+ self.driver.delete_vlan(switch_ip, vlan_id)
+
+ def _port_action(self, context, func):
+ """Verify configuration and then process event."""
+ device_id = context.current.get('device_id')
+ host_id = context.current.get(portbindings.HOST_ID)
+
+ # Workaround until vlan can be retrieved during delete_port_postcommit
+ # (or prehaps unbind_port) event.
+ if func == self._delete_switch_entry:
+ vlan_id = self._delete_port_postcommit_vlan
+ else:
+ vlan_id = self._get_vlanid(context)
+
+ if vlan_id and device_id and host_id:
+ func(context, vlan_id, device_id, host_id)
+ else:
+ fields = "vlan_id " if not vlan_id else ""
+ fields += "device_id " if not device_id else ""
+ fields += "host_id" if not host_id else ""
+ raise excep.NexusMissingRequiredFields(fields=fields)
+
+ # Workaround until vlan can be retrieved during delete_port_postcommit
+ # (or prehaps unbind_port) event.
+ if func == self._delete_nxos_db:
+ self._delete_port_postcommit_vlan = vlan_id
+ else:
+ self._delete_port_postcommit_vlan = 0
+
+ def update_port_precommit(self, context):
+ """Update port pre-database transaction commit event."""
+ port = context.current
+ if self._is_deviceowner_compute(port) and self._is_status_active(port):
+ self._port_action(context, self._configure_nxos_db)
+
+ def update_port_postcommit(self, context):
+ """Update port non-database commit event."""
+ port = context.current
+ if self._is_deviceowner_compute(port) and self._is_status_active(port):
+ self._port_action(context, self._configure_switch_entry)
+
+ def delete_port_precommit(self, context):
+ """Delete port pre-database commit event."""
+ if self._is_deviceowner_compute(context.current):
+ self._port_action(context, self._delete_nxos_db)
+
+ def delete_port_postcommit(self, context):
+ """Delete port non-database commit event."""
+ if self._is_deviceowner_compute(context.current):
+ self._port_action(context, self._delete_switch_entry)
from neutron.plugins.ml2.drivers.cisco import config as cisco_config
from neutron.plugins.ml2.drivers.cisco import exceptions as c_exc
from neutron.plugins.ml2.drivers.cisco import mech_cisco_nexus
-from neutron.plugins.ml2.drivers.cisco import nexus_db_v2
from neutron.plugins.ml2.drivers.cisco import nexus_network_driver
from neutron.plugins.ml2.drivers import type_vlan as vlan_config
from neutron.tests.unit import test_db_plugin
ML2_PLUGIN = 'neutron.plugins.ml2.plugin.Ml2Plugin'
PHYS_NET = 'physnet1'
COMP_HOST_NAME = 'testhost'
+COMP_HOST_NAME_2 = 'testhost_2'
VLAN_START = 1000
VLAN_END = 1100
NEXUS_IP_ADDR = '1.1.1.1'
+NETWORK_NAME = 'test_network'
+NETWORK_NAME_2 = 'test_network_2'
+NEXUS_INTERFACE = '1/1'
+NEXUS_INTERFACE_2 = '1/2'
CIDR_1 = '10.0.0.0/24'
CIDR_2 = '10.0.1.0/24'
DEVICE_ID_1 = '11111111-1111-1111-1111-111111111111'
DEVICE_ID_2 = '22222222-2222-2222-2222-222222222222'
+DEVICE_OWNER = 'compute:None'
class CiscoML2MechanismTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
(NEXUS_IP_ADDR, 'username'): 'admin',
(NEXUS_IP_ADDR, 'password'): 'mySecretPassword',
(NEXUS_IP_ADDR, 'ssh_port'): 22,
- (NEXUS_IP_ADDR, COMP_HOST_NAME): '1/1'}
+ (NEXUS_IP_ADDR, COMP_HOST_NAME): NEXUS_INTERFACE,
+ (NEXUS_IP_ADDR, COMP_HOST_NAME_2): NEXUS_INTERFACE_2}
nexus_patch = mock.patch.dict(
cisco_config.ML2MechCiscoConfig.nexus_dict,
nexus_config)
'_import_ncclient',
return_value=self.mock_ncclient).start()
- # Mock port values for 'status' and 'binding:segmenation_id'
+ # Mock port values for 'status' and 'binding:segmentation_id'
mock_status = mock.patch.object(
mech_cisco_nexus.CiscoNexusMechanismDriver,
'_is_status_active').start()
mock_status.return_value = n_const.PORT_STATUS_ACTIVE
def _mock_get_vlanid(context):
- port = context.current
- if port['device_id'] == DEVICE_ID_1:
+ network = context.network.current
+ if network['name'] == NETWORK_NAME:
return VLAN_START
else:
return VLAN_START + 1
config = {attr: None}
self.mock_ncclient.configure_mock(**config)
+ def _is_in_nexus_cfg(self, words):
+ """Check if any config sent to Nexus contains all words in a list."""
+ for call in (self.mock_ncclient.connect.return_value.
+ edit_config.mock_calls):
+ configlet = call[2]['config']
+ if all(word in configlet for word in words):
+ return True
+ return False
+
def _is_in_last_nexus_cfg(self, words):
"""Confirm last config sent to Nexus contains specified keywords."""
last_cfg = (self.mock_ncclient.connect.return_value.
edit_config.mock_calls[-1][2]['config'])
return all(word in last_cfg for word in words)
+ def _is_vlan_configured(self, vlan_creation_expected=True,
+ add_keyword_expected=False):
+ vlan_created = self._is_in_nexus_cfg(['vlan', 'vlan-name'])
+ add_appears = self._is_in_last_nexus_cfg(['add'])
+ return (self._is_in_last_nexus_cfg(['allowed', 'vlan']) and
+ vlan_created == vlan_creation_expected and
+ add_appears == add_keyword_expected)
+
+ def _is_vlan_unconfigured(self, vlan_deletion_expected=True):
+ vlan_deleted = self._is_in_last_nexus_cfg(
+ ['no', 'vlan', 'vlan-id-create-delete'])
+ return (self._is_in_nexus_cfg(['allowed', 'vlan', 'remove']) and
+ vlan_deleted == vlan_deletion_expected)
+
class TestCiscoBasicGet(CiscoML2MechanismTestCase,
test_db_plugin.TestBasicGet):
test_db_plugin.TestPortsV2):
@contextlib.contextmanager
- def _create_resources(self, name='myname', cidr=CIDR_1,
+ def _create_resources(self, name=NETWORK_NAME, cidr=CIDR_1,
device_id=DEVICE_ID_1,
- host_id=COMP_HOST_NAME,
- expected_exception=None):
+ host_id=COMP_HOST_NAME):
"""Create network, subnet, and port resources for test cases.
- Create a network, subnet, port and then update port, yield the result,
- then delete the port, subnet, and network.
+ Create a network, subnet, port and then update the port, yield the
+ result, then delete the port, subnet and network.
:param name: Name of network to be created.
:param cidr: cidr address of subnetwork to be created.
:param device_id: Device ID to use for port to be created/updated.
:param host_id: Host ID to use for port create/update.
- :param expected_exception: Expected HTTP code.
"""
- ctx = context.get_admin_context()
with self.network(name=name) as network:
with self.subnet(network=network, cidr=cidr) as subnet:
- net_id = subnet['subnet']['network_id']
- args = (portbindings.HOST_ID, 'device_id', 'device_owner',
- 'admin_state_up')
- port_dict = {portbindings.HOST_ID: host_id,
- 'device_id': device_id,
- 'device_owner': 'compute:none',
- 'admin_state_up': True}
-
- res = self._create_port(self.fmt, net_id, arg_list=args,
- context=ctx, **port_dict)
- port = self.deserialize(self.fmt, res)
-
- expected_exception = self._expectedHTTP(expected_exception)
- data = {'port': port_dict}
- self._update('ports', port['port']['id'], data,
- expected_code=expected_exception,
- neutron_context=ctx)
-
- try:
- yield port
- finally:
- self._delete('ports', port['port']['id'])
-
- def _expectedHTTP(self, exc):
- """Map a Cisco exception to the HTTP status equivalent.
-
- :param exc: Expected Cisco exception
+ with self.port(subnet=subnet, cidr=cidr) as port:
+ data = {'port': {portbindings.HOST_ID: host_id,
+ 'device_id': device_id,
+ 'device_owner': 'compute:none',
+ 'admin_state_up': True}}
+ req = self.new_update_request('ports', data,
+ port['port']['id'])
+ res = req.get_response(self.api)
+ yield res.status_int
+
+ def _assertExpectedHTTP(self, status, exc):
+ """Confirm that an HTTP status corresponds to an expected exception.
+
+ Confirm that an HTTP status which has been returned for an
+ neutron API request matches the HTTP status corresponding
+ to an expected exception.
+
+ :param status: HTTP status
+ :param exc: Expected exception
"""
- if exc == None:
- expected_http = wexc.HTTPOk.code
- elif exc in base.FAULT_MAP:
+ if exc in base.FAULT_MAP:
expected_http = base.FAULT_MAP[exc].code
else:
expected_http = wexc.HTTPInternalServerError.code
-
- return expected_http
+ self.assertEqual(status, expected_http)
def test_create_ports_bulk_emulated_plugin_failure(self):
real_has_attr = hasattr
keyword 'add'.
Confirm that for the second VLAN configured on a Nexus interface,
- the command staring sent to the switch contains the keyword 'add'.
+ the command string sent to the switch contains the keyword 'add'.
"""
- with self._create_resources(name='net1', cidr=CIDR_1):
- self.assertTrue(self._is_in_last_nexus_cfg(['allowed', 'vlan']))
- self.assertFalse(self._is_in_last_nexus_cfg(['add']))
- with self._create_resources(name='net2', device_id=DEVICE_ID_2,
+ # First vlan should be configured without 'add' keyword
+ with self._create_resources():
+ self.assertTrue(self._is_vlan_configured(
+ vlan_creation_expected=True,
+ add_keyword_expected=False))
+ self.mock_ncclient.reset_mock()
+
+ # Second vlan should be configured with 'add' keyword
+ with self._create_resources(name=NETWORK_NAME_2,
+ device_id=DEVICE_ID_2,
cidr=CIDR_2):
- self.assertTrue(
- self._is_in_last_nexus_cfg(['allowed', 'vlan', 'add']))
+ self.assertTrue(self._is_vlan_configured(
+ vlan_creation_expected=True,
+ add_keyword_expected=True))
def test_nexus_connect_fail(self):
"""Test failure to connect to a Nexus switch.
"""
with self._patch_ncclient('connect.side_effect',
AttributeError):
- self._create_resources(expected_exception=c_exc.NexusConnectFailed)
+ with self._create_resources() as result_status:
+ self._assertExpectedHTTP(result_status,
+ c_exc.NexusConnectFailed)
+
+ def test_nexus_vlan_config_two_hosts(self):
+ """Verify config/unconfig of vlan on two compute hosts."""
+
+ @contextlib.contextmanager
+ def _create_port_check_vlan(comp_host_name, device_id,
+ vlan_creation_expected=True):
+ with self.port(subnet=subnet, fmt=self.fmt) as port:
+ data = {'port': {portbindings.HOST_ID: comp_host_name,
+ 'device_id': device_id,
+ 'device_owner': DEVICE_OWNER,
+ 'admin_state_up': True}}
+ req = self.new_update_request('ports', data,
+ port['port']['id'])
+ req.get_response(self.api)
+ self.assertTrue(self._is_vlan_configured(
+ vlan_creation_expected=vlan_creation_expected,
+ add_keyword_expected=False))
+ self.mock_ncclient.reset_mock()
+ yield
+
+ # Create network and subnet
+ with self.network(name=NETWORK_NAME) as network:
+ with self.subnet(network=network, cidr=CIDR_1) as subnet:
+
+ # Create an instance on first compute host
+ with _create_port_check_vlan(COMP_HOST_NAME, DEVICE_ID_1,
+ vlan_creation_expected=True):
+ # Create an instance on second compute host
+ with _create_port_check_vlan(COMP_HOST_NAME_2, DEVICE_ID_2,
+ vlan_creation_expected=False):
+ pass
+
+ # Instance on second host is now terminated.
+ # Vlan should be untrunked from port, but vlan should
+ # still exist on the switch.
+ self.assertTrue(self._is_vlan_unconfigured(
+ vlan_deletion_expected=False))
+ self.mock_ncclient.reset_mock()
+
+ # Instance on first host is now terminated.
+ # Vlan should be untrunked from port and vlan should have
+ # been deleted from the switch.
+ self.assertTrue(self._is_vlan_unconfigured(
+ vlan_deletion_expected=True))
def test_nexus_config_fail(self):
"""Test a Nexus switch configuration failure.
with self._patch_ncclient(
'connect.return_value.edit_config.side_effect',
AttributeError):
- self._create_resources(expected_exception=c_exc.NexusConfigFailed)
+ with self._create_resources() as result_status:
+ self._assertExpectedHTTP(result_status,
+ c_exc.NexusConfigFailed)
def test_nexus_extended_vlan_range_failure(self):
"""Test that extended VLAN range config errors are ignored.
with self._patch_ncclient(
'connect.return_value.edit_config.side_effect',
mock_edit_config_a):
- self._create_resources(name='myname')
+ with self._create_resources() as result_status:
+ self.assertEqual(result_status, wexc.HTTPOk.code)
def mock_edit_config_b(target, config):
if all(word in config for word in ['no', 'shutdown']):
with self._patch_ncclient(
'connect.return_value.edit_config.side_effect',
mock_edit_config_b):
- self._create_resources(name='myname')
+ with self._create_resources() as result_status:
+ self.assertEqual(result_status, wexc.HTTPOk.code)
def test_nexus_vlan_config_rollback(self):
"""Test rollback following Nexus VLAN state config failure.
with self._patch_ncclient(
'connect.return_value.edit_config.side_effect',
mock_edit_config):
- with self._create_resources(
- name='myname',
- expected_exception=c_exc.NexusConfigFailed):
+ with self._create_resources() as result_status:
# Confirm that the last configuration sent to the Nexus
# switch was deletion of the VLAN.
self.assertTrue(self._is_in_last_nexus_cfg(['<no>', '<vlan>']))
+ self._assertExpectedHTTP(result_status,
+ c_exc.NexusConfigFailed)
def test_nexus_host_not_configured(self):
"""Test handling of a NexusComputeHostNotConfigured exception.
a fictitious host name during port creation.
"""
- self._create_resources(
- host_id='fake_host',
- expected_exception=c_exc.NexusComputeHostNotConfigured)
-
- def test_nexus_bind_fail_rollback(self):
- """Test for proper rollback following add Nexus DB binding failure.
-
- Test that the Cisco Nexus plugin correctly rolls back the vlan
- configuration on the Nexus switch when add_nexusport_binding fails
- within the plugin's create_port() method.
-
- """
- with mock.patch.object(nexus_db_v2,
- 'add_nexusport_binding',
- side_effect=KeyError):
- with self._create_resources(expected_exception=KeyError):
- # Confirm that the last configuration sent to the Nexus
- # switch was a removal of vlan from the test interface.
- self.assertTrue(
- self._is_in_last_nexus_cfg(['<vlan>', '<remove>'])
- )
+ with self._create_resources(host_id='fake_host') as result_status:
+ self._assertExpectedHTTP(result_status,
+ c_exc.NexusComputeHostNotConfigured)
- def test_nexus_delete_port_rollback(self):
- """Test for proper rollback for nexus plugin delete port failure.
+ def test_nexus_missing_fields(self):
+ """Test handling of a NexusMissingRequiredFields exception.
- Test for rollback (i.e. restoration) of a VLAN entry in the
- nexus database whenever the nexus plugin fails to reconfigure the
- nexus switch during a delete_port operation.
+ Test the Cisco NexusMissingRequiredFields exception by using
+ empty host_id and device_id values during port creation.
"""
- with self._create_resources() as port:
- # Check that there is only one binding in the nexus database
- # for this VLAN/nexus switch.
- start_rows = nexus_db_v2.get_nexusvlan_binding(VLAN_START,
- NEXUS_IP_ADDR)
- self.assertEqual(len(start_rows), 1)
-
- # Simulate a Nexus switch configuration error during
- # port deletion.
- with self._patch_ncclient(
- 'connect.return_value.edit_config.side_effect',
- AttributeError):
- self._delete('ports', port['port']['id'],
- wexc.HTTPInternalServerError.code)
-
- # Confirm that the binding has been restored (rolled back).
- end_rows = nexus_db_v2.get_nexusvlan_binding(VLAN_START,
- NEXUS_IP_ADDR)
- self.assertEqual(start_rows, end_rows)
+ with self._create_resources(device_id='', host_id='') as result_status:
+ self._assertExpectedHTTP(result_status,
+ c_exc.NexusMissingRequiredFields)
class TestCiscoNetworksV2(CiscoML2MechanismTestCase,