openstack
/
deb-networking-cisco
Code Issues Proposed changes

Make multi-switch host successful when 1 active 

If you have a compute host sending traffic to multiple switches,
the configuration must be successful when at least one switch is
active.

Change-Id: I0fe103b3da78bed0c0c66f91dae9fac4345ebc42
Closes-bug:  #1516679
This commit is contained in:
Carol Bouchard 2015-11-18 17:20:32 -05:00
parent b7bddc9cac
commit 536cf6cc91
3 changed files with 337 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
networking_cisco/plugins/ml2/drivers/cisco/nexus/mech_cisco_nexus.py
View File

@ -464,7 +464,34 @@ class CiscoNexusMechanismDriver(api.MechanismDriver):
intf_type, port = 'ethernet', port_id
host_list.append((switch_ip, intf_type, port))
def _get_switch_info(self, host_id):
def _get_host_switches(self, host_id):
all_switches = set()
active_switches = set()
for switch_ip, attr in self._nexus_switches:
if str(attr) == str(host_id):
all_switches.add(switch_ip)
if self.is_switch_active(switch_ip):
active_switches.add(switch_ip)
return list(all_switches), list(active_switches)
def _get_active_host_connections(self, host_id):
host_found = False
host_connections = []
for switch_ip, attr in self._nexus_switches:
if str(attr) == str(host_id):
host_found = True
if self.is_switch_active(switch_ip):
self._gather_configured_ports(
switch_ip, attr, host_connections)
if not host_found:
LOG.warn(HOST_NOT_FOUND, host_id)
return host_connections
def _get_host_connections(self, host_id):
host_connections = []
for switch_ip, attr in self._nexus_switches:
if str(attr) == str(host_id):
@ -531,12 +558,6 @@ class CiscoNexusMechanismDriver(api.MechanismDriver):
for switch_ip in host_nve_connections:
if not self.is_switch_active(switch_ip):
raise excep.NexusConfigFailed(
nexus_host=switch_ip, config="None",
exc="Update nve member Failed: Nexus Switch "
"is down or replay in progress")
# If configured to set global VXLAN values then
# If this is the first database entry for this switch_ip
# then configure the "interface nve" entry on the switch.
@ -575,12 +596,6 @@ class CiscoNexusMechanismDriver(api.MechanismDriver):
host_nve_connections = self._get_switch_nve_info(host_id)
for switch_ip in host_nve_connections:
if not self.is_switch_active(switch_ip):
raise excep.NexusConfigFailed(
nexus_host=switch_ip, config="None",
exc="Update nve member Failed: Nexus Switch "
"is down or replay in progress")
if not nxos_db.get_nve_vni_switch_bindings(vni, switch_ip):
self.driver.delete_nve_member(switch_ip,
const.NVE_INT_NUM, vni)
@ -594,7 +609,7 @@ class CiscoNexusMechanismDriver(api.MechanismDriver):
Called during update precommit port event.
"""
host_connections = self._get_switch_info(host_id)
host_connections = self._get_host_connections(host_id)
for switch_ip, intf_type, nexus_port in host_connections:
port_id = '%s:%s' % (intf_type, nexus_port)
try:
@ -772,20 +787,15 @@ class CiscoNexusMechanismDriver(api.MechanismDriver):
Called during update postcommit port event.
"""
host_connections = self._get_switch_info(host_id)
host_connections = self._get_active_host_connections(host_id)
# (nexus_port,switch_ip) will be unique in each iteration.
# But switch_ip will repeat if host has >1 connection to same switch.
# So track which switch_ips already have vlan created in this loop.
vlan_already_created = []
starttime = time.time()
for switch_ip, intf_type, nexus_port in host_connections:
if not self.is_switch_active(switch_ip):
raise excep.NexusConfigFailed(
nexus_host=switch_ip, config="None",
exc="Update Failed: Nexus Switch is down or "
"replay in progress")
for switch_ip, intf_type, nexus_port in host_connections:
all_bindings = nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
previous_bindings = [row for row in all_bindings
@ -970,7 +980,7 @@ class CiscoNexusMechanismDriver(api.MechanismDriver):
Called during delete postcommit port event.
"""
host_connections = self._get_switch_info(host_id)
host_connections = self._get_host_connections(host_id)
# (nexus_port,switch_ip) will be unique in each iteration.
# But switch_ip will repeat if host has >1 connection to same switch.
@ -1114,24 +1124,41 @@ class CiscoNexusMechanismDriver(api.MechanismDriver):
def create_port_postcommit(self, context):
"""Create port non-database commit event."""
port = context.current
host_id = port.get(portbindings.HOST_ID)
host_connections = self._get_switch_info(host_id)
if self._is_supported_deviceowner(port):
# For each unique switch, verify you can talk
# to it; otherwise, let exception bubble
# up so other dbs cleaned and no further retries.
verified = []
for switch_ip, intf_type, nexus_port in host_connections:
if not self.is_switch_active(switch_ip):
raise excep.NexusConfigFailed(
nexus_host=switch_ip, config="None",
exc="Create Failed: Port event can not "
"be processed at this time.")
# No new events are handled until replay
# thread has put the switch in active state.
# If a switch is in active state, verify
# the switch is still in active state
# before accepting this new event.
#
# If create_port_postcommit fails, it causes
# other openstack dbs to be cleared and
# retries for new VMs will stop. Subnet
# transactions will continue to be retried.
if switch_ip not in verified:
port = context.current
if self._is_supported_deviceowner(port):
host_id = port.get(portbindings.HOST_ID)
all_switches, active_switches = (
self._get_host_switches(host_id))
# Verify switch is still up before replay
# thread checks.
verified_active_switches = []
for switch_ip in active_switches:
try:
self.driver.get_nexus_type(switch_ip)
verified.append(switch_ip)
verified_active_switches.append(switch_ip)
except Exception:
pass
# if host_id is valid and there is no active
# switches remaining
if all_switches and not verified_active_switches:
raise excep.NexusConnectFailed(
nexus_host=all_switches[0], config="None",
exc="Create Failed: Port event can not "
"be processed at this time.")
@lockutils.synchronized('cisco-nexus-portlock')
def update_port_precommit(self, context):
@ -1178,6 +1205,15 @@ class CiscoNexusMechanismDriver(api.MechanismDriver):
else:
if (self._is_supported_deviceowner(context.current) and
self._is_status_active(context.current)):
host_id = context.current.get(portbindings.HOST_ID)
all_switches, active_switches = (
self._get_host_switches(host_id))
# if switches not active but host_id is valid
if not active_switches and all_switches:
raise excep.NexusConnectFailed(
nexus_host=all_switches[0], config="None",
exc="Update Port Failed: Nexus Switch "
"is down or replay in progress")
vni = self._port_action_vxlan(context.current, vxlan_segment,
self._configure_nve_member) if vxlan_segment else 0
self._port_action_vlan(context.current, vlan_segment,
@ -1216,7 +1252,7 @@ class CiscoNexusMechanismDriver(api.MechanismDriver):
# Find physical network setting for this host.
host_id = context.current.get(portbindings.HOST_ID)
host_connections = self._get_switch_info(host_id)
host_connections = self._get_host_connections(host_id)
if not host_connections:
return

5
networking_cisco/tests/unit/ml2/drivers/cisco/nexus/test_cisco_mech.py
View File

@ -190,6 +190,11 @@ class CiscoML2MechanismTestCase(test_plugin.Ml2PluginV2TestCase):
'_is_status_active').start()
self.mock_status.side_effect = _mock_check_bind_state
self.mock_switch_status = mock.patch.object(
mech_cisco_nexus.CiscoNexusMechanismDriver,
'is_switch_active').start()
self.mock_switch_status.return_value = True
super(CiscoML2MechanismTestCase, self).setUp()
self.port_create_status = 'DOWN'

295
networking_cisco/tests/unit/ml2/drivers/cisco/nexus/test_cisco_nexus.py
View File

@ -398,7 +398,8 @@ class TestCiscoNexusDevice(testlib_api.SqlTestCase):
return _side_effect_method
def _create_port_failure(self, attr, match_str, test_case, test_id):
def _create_port_failure(self, attr, match_str, test_case, test_id,
which_exc=exceptions.NexusConfigFailed):
"""Verifies exception handling during initial create object.
This method is a shared method to initiate an exception
@ -421,7 +422,7 @@ class TestCiscoNexusDevice(testlib_api.SqlTestCase):
Exception(test_id))}
self.mock_ncclient.configure_mock(**config)
e = self.assertRaises(
exceptions.NexusConfigFailed,
which_exc,
self._create_port,
TestCiscoNexusDevice.test_configs[test_case])
self.assertIn(test_id, six.u(str(e)))
@ -494,6 +495,14 @@ class TestCiscoNexusDevice(testlib_api.SqlTestCase):
self._create_port(
TestCiscoNexusDevice.test_configs['test_config3'])
# TODO(caboucha)
# Commented out until the correct fix for
# the following issue is resolved.
# https://review.openstack.org/#/c/241216/
#
# verify first config was indeed configured
# Original code was as follows:
# self._verify_results(duplicate_add_port_driver_result)
# Verify there are 2 port configs
bindings = nexus_db_v2.get_nexusvlan_binding(VLAN_ID_1,
@ -552,7 +561,8 @@ class TestCiscoNexusDevice(testlib_api.SqlTestCase):
'connect.return_value.get.side_effect',
'show inventory',
'test_config1',
__name__)
'Create Failed:',
which_exc=exceptions.NexusConnectFailed)
# Verify we attempt to connect once. get_nexus_type is a
# special case since replay code will retry
@ -663,15 +673,22 @@ class TestCiscoNexusDevice(testlib_api.SqlTestCase):
RP_NEXUS_IP_ADDRESS_1 = '1.1.1.1'
RP_NEXUS_IP_ADDRESS_2 = '2.2.2.2'
RP_NEXUS_IP_ADDRESS_3 = '3.3.3.3'
RP_NEXUS_IP_ADDRESS_DUAL = '4.4.4.4'
RP_NEXUS_IP_ADDRESS_DUAL2 = '5.5.5.5'
RP_HOST_NAME_1 = 'UniquePort'
RP_HOST_NAME_2 = 'DuplicateVlan'
RP_HOST_NAME_3 = 'DuplicatePort'
RP_HOST_NAME_DUAL = 'testdualhost'
RP_INSTANCE_1 = 'testvm1'
RP_INSTANCE_2 = 'testvm2'
RP_INSTANCE_DUAL = 'testdualvm'
RP_NEXUS_PORT_1 = 'ethernet:1/10'
RP_NEXUS_PORT_2 = 'ethernet:1/20'
RP_NEXUS_DUAL1 = 'ethernet:1/3'
RP_NEXUS_DUAL2 = 'ethernet:1/2'
RP_VLAN_ID_1 = 267
RP_VLAN_ID_2 = 265
RP_VLAN_ID_DUAL = 269
MAX_REPLAY_COUNT = 4
@ -738,6 +755,24 @@ class TestCiscoNexusReplay(testlib_api.SqlTestCase):
NO_VXLAN_ID,
None,
DEVICE_OWNER_COMPUTE),
'test_replay_dual': TestConfigObj(
RP_NEXUS_IP_ADDRESS_DUAL,
RP_HOST_NAME_DUAL,
RP_NEXUS_DUAL1,
RP_INSTANCE_DUAL,
RP_VLAN_ID_DUAL,
NO_VXLAN_ID,
None,
DEVICE_OWNER_COMPUTE),
'test_replay_dual2': TestConfigObj(
RP_NEXUS_IP_ADDRESS_DUAL2,
RP_HOST_NAME_DUAL,
RP_NEXUS_DUAL2,
RP_INSTANCE_DUAL,
RP_VLAN_ID_DUAL,
NO_VXLAN_ID,
None,
DEVICE_OWNER_COMPUTE),
'test_replay_vxlan_unique1': TestConfigObj(
RP_NEXUS_IP_ADDRESS_1,
RP_HOST_NAME_1,
@ -788,6 +823,34 @@ class TestCiscoNexusReplay(testlib_api.SqlTestCase):
'\<no\>\s+\<vlan\>\s+<vlan-id-create-delete\>'
'\s+\<__XML__PARAM_value\>267',
]
driver_result_dual_add_if1 = [
'\<ethernet\>\s+\<interface\>1\/3\<\/interface\>\s+'
'[\x20-\x7e]+\s+\<switchport\>\s+\<trunk\>\s+'
'\<allowed\>\s+\<vlan\>\s+\<add\>\s+\<vlan_id\>269',
]
driver_result_dual_add_vlan = [
'configure\>\s+\<vlan\>\s+<vlan-id-create-delete\>'
'\s+\<__XML__PARAM_value\>269',
]
driver_result_dual_add_if2 = [
'\<ethernet\>\s+\<interface\>1\/2\<\/interface\>\s+'
'[\x20-\x7e]+\s+\<switchport\>\s+\<trunk\>\s+'
'\<allowed\>\s+\<vlan\>\s+\<add\>\s+\<vlan_id\>269',
]
driver_result_dual_del1 = [
'\<ethernet\>\s+\<interface\>1\/3\<\/interface\>\s+'
'[\x20-\x7e\s]+\<switchport\>\s+\<trunk\>\s+'
'\<allowed\>\s+\<vlan\>\s+\<remove\>\s+\<vlan\>269',
'\<no\>\s+\<vlan\>\s+<vlan-id-create-delete\>'
'\s+\<__XML__PARAM_value\>269',
]
driver_result_dual_del2 = [
'\<ethernet\>\s+\<interface\>1\/2\<\/interface\>\s+'
'[\x20-\x7e\s]+\<switchport\>\s+\<trunk\>\s+'
'\<allowed\>\s+\<vlan\>\s+\<remove\>\s+\<vlan\>269',
'\<no\>\s+\<vlan\>\s+<vlan-id-create-delete\>'
'\s+\<__XML__PARAM_value\>269',
]
test_configs = collections.OrderedDict(sorted(test_configs.items()))
def setUp(self):
@ -851,11 +914,10 @@ class TestCiscoNexusReplay(testlib_api.SqlTestCase):
self._cisco_mech_driver.driver,
self._cisco_mech_driver))
def _create_port(self, port_config):
"""Tests creation of a virtual port."""
nexus_ip_addr = port_config.nexus_ip_addr
def _generate_port_context(self, port_config):
"""Returns port context from port_config."""
host_name = port_config.host_name
nexus_port = port_config.nexus_port
instance_id = port_config.instance_id
vlan_id = port_config.vlan_id
vxlan_id = port_config.vxlan_id
@ -872,46 +934,38 @@ class TestCiscoNexusReplay(testlib_api.SqlTestCase):
port_context = FakePortContext(instance_id, host_name,
device_owner, network_context)
return port_context
def _create_port(self, port_config):
"""Tests creation of a virtual port."""
port_context = self._generate_port_context(port_config)
self._cisco_mech_driver.create_port_postcommit(port_context)
self._cisco_mech_driver.update_port_precommit(port_context)
self._cisco_mech_driver.update_port_postcommit(port_context)
for port_id in nexus_port.split(','):
bindings = nexus_db_v2.get_nexusport_binding(port_id,
vlan_id,
nexus_ip_addr,
instance_id)
for port_id in port_config.nexus_port.split(','):
bindings = nexus_db_v2.get_nexusport_binding(
port_id,
port_config.vlan_id,
port_config.nexus_ip_addr,
port_config.instance_id)
self.assertEqual(len(bindings), 1)
def _delete_port(self, port_config):
"""Tests deletion of a virtual port."""
nexus_ip_addr = port_config.nexus_ip_addr
host_name = port_config.host_name
nexus_port = port_config.nexus_port
instance_id = port_config.instance_id
vlan_id = port_config.vlan_id
vxlan_id = port_config.vxlan_id
mcast_group = port_config.mcast_group
device_owner = port_config.device_owner
network_context = FakeNetworkContext(vlan_id, NETWORK_TYPE)
if vxlan_id != NO_VXLAN_ID:
vxlan_network_context = FakeNetworkContext(vlan_id,
NEXUS_VXLAN_NETWORK_TYPE, mcast_group)
port_context = FakePortContext(instance_id, host_name,
device_owner, vxlan_network_context, network_context)
else:
port_context = FakePortContext(instance_id, host_name,
device_owner, network_context)
port_context = self._generate_port_context(port_config)
self._cisco_mech_driver.delete_port_precommit(port_context)
self._cisco_mech_driver.delete_port_postcommit(port_context)
for port_id in nexus_port.split(','):
for port_id in port_config.nexus_port.split(','):
with testtools.ExpectedException(
exceptions.NexusPortBindingNotFound):
nexus_db_v2.get_nexusport_binding(port_id,
vlan_id,
nexus_ip_addr,
instance_id)
nexus_db_v2.get_nexusport_binding(
port_id,
port_config.vlan_id,
port_config.nexus_ip_addr,
port_config.instance_id)
def _verify_replay_results(self, driver_result):
"""Verifies correct entries sent to Nexus."""
@ -945,6 +999,9 @@ class TestCiscoNexusReplay(testlib_api.SqlTestCase):
assert(len(nexus_db_v2.get_nexusport_switch_bindings(
port_cfg.nexus_ip_addr)) == nbr_of_bindings + 1)
# Make sure there is only a single attempt to configure.
self._verify_replay_results(test_result)
# Clean all the ncclient mock_calls to clear exception
# and other mock_call history.
self.mock_ncclient.reset_mock()
@ -1062,7 +1119,8 @@ class TestCiscoNexusReplay(testlib_api.SqlTestCase):
Exception(__name__))}
self.mock_ncclient.configure_mock(**config)
def _create_port_failure(self, attr, match_str, test_case, test_id):
def _create_port_failure(self, attr, match_str, test_case, test_id,
which_exc=exceptions.NexusConfigFailed):
"""Verifies exception handling during initial create object.
This method is a shared method to initiate an exception
@ -1214,7 +1272,14 @@ class TestCiscoNexusReplay(testlib_api.SqlTestCase):
driver_result_duplvlan_add_vlan +
driver_result_duplvlan_add2),
'nbr_db_entries': 2}
second_add = {'driver_results': [],
# TODO(caboucha)
# 'driver_result': [], until the correct fix for
# the following issue is resolved.
# https://review.openstack.org/#/c/241216/
second_add = {'driver_results': (
driver_result_duplvlan_add_vlan +
driver_result_duplvlan_add1 +
driver_result_duplvlan_add2),
'nbr_db_entries': 4}
first_del = {'driver_results': [],
'nbr_db_entries': 2}
@ -1255,7 +1320,13 @@ class TestCiscoNexusReplay(testlib_api.SqlTestCase):
(driver_result_duplport_add_vlan +
driver_result_duplport_add_if),
'nbr_db_entries': 1}
second_add = {'driver_results': [],
# TODO(caboucha)
# 'driver_result': [], until the correct fix for
# the following issue is resolved.
# https://review.openstack.org/#/c/241216/
second_add = {'driver_results':
(driver_result_duplport_add_vlan +
driver_result_duplport_add_if),
'nbr_db_entries': 2}
first_del = {'driver_results': [],
'nbr_db_entries': 1}
@ -1349,6 +1420,154 @@ class TestCiscoNexusReplay(testlib_api.SqlTestCase):
'test_replay_unique1',
__name__)
def test_replay_new_port_success_if_one_switch_up(self):
"""Verifies create port successful if one multi-switch up."""
# Make sure port is not rejected when there are multiple
# switches and only one is active.
port_cfg1 = TestCiscoNexusReplay.test_configs['test_replay_dual']
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg1.nexus_ip_addr, const.SWITCH_ACTIVE)
port_cfg2 = TestCiscoNexusReplay.test_configs['test_replay_dual2']
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg2.nexus_ip_addr, const.SWITCH_INACTIVE)
# Set-up successful creation of port vlan config
self._basic_create_verify_port_vlan('test_replay_dual',
(self.driver_result_dual_add_vlan +
self.driver_result_dual_add_if1),
nbr_of_bindings=1)
# Even though 2nd entry is inactive, there should be
# a data base entry configured for it.
# 2 = One entry for port the other for reserved binding
assert(len(nexus_db_v2.get_nexusport_switch_bindings(
port_cfg2.nexus_ip_addr)) == 2)
# Clean-up the port entry
self._basic_delete_verify_port_vlan('test_replay_dual',
self.driver_result_dual_del1 + self.driver_result_dual_del2,
nbr_of_bindings=0)
def test_replay_port_success_if_one_switch_restored(self):
"""Verifies port restored after one of multi-switch restored."""
# Make sure port is not rejected when there are multiple
# switches and one is active. Then proceed to bring-up
# the other switch and it gets configured successfully.
# Then remove all.
port_cfg1 = TestCiscoNexusReplay.test_configs['test_replay_dual']
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg1.nexus_ip_addr, const.SWITCH_ACTIVE)
port_cfg2 = TestCiscoNexusReplay.test_configs['test_replay_dual2']
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg2.nexus_ip_addr, const.SWITCH_INACTIVE)
# Set-up successful creation of port vlan config
self._basic_create_verify_port_vlan('test_replay_dual',
(self.driver_result_dual_add_vlan +
self.driver_result_dual_add_if1),
nbr_of_bindings=1)
# Even though 2nd entry is inactive, there should be
# a data base entry configured for it.
# 2 = One entry for port the other for reserved binding
assert(len(nexus_db_v2.get_nexusport_switch_bindings(
port_cfg2.nexus_ip_addr)) == 2)
# Restore port data for that switch
self._cfg_monitor.check_connections()
self._verify_replay_results(
self.driver_result_dual_add_if2 +
self.driver_result_dual_add_vlan)
# Clear mock_call history.
self.mock_ncclient.reset_mock()
# Clean-up the port entries
self._basic_delete_verify_port_vlan('test_replay_dual',
self.driver_result_dual_del1 + self.driver_result_dual_del2,
nbr_of_bindings=0)
def test_replay_create_fails_if_single_switch_down(self):
"""Verifies port create fails if switch down."""
# Make sure create ethernet config fails when the
# switch state is inactive.
port_cfg = TestCiscoNexusReplay.test_configs['test_replay_unique1']
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg.nexus_ip_addr, const.SWITCH_INACTIVE)
port_context = self._generate_port_context(port_cfg)
self.assertRaises(
exceptions.NexusConnectFailed,
self._cisco_mech_driver.create_port_postcommit,
port_context)
def test_replay_update_fails_if_single_switch_down(self):
"""Verifies port update fails if switch down."""
# Make sure update ethernet config fails when the
# switch state is inactive.
port_cfg = TestCiscoNexusReplay.test_configs['test_replay_unique1']
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg.nexus_ip_addr, const.SWITCH_INACTIVE)
port_context = self._generate_port_context(port_cfg)
self.assertRaises(
exceptions.NexusConnectFailed,
self._cisco_mech_driver.update_port_postcommit,
port_context)
def test_replay_delete_success_if_switch_down(self):
"""Verifies port delete success if switch down."""
# Make sure delete config successful even when the
# switch state is inactive.
port_cfg = TestCiscoNexusReplay.test_configs['test_replay_unique1']
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg.nexus_ip_addr, const.SWITCH_ACTIVE)
# Set-up successful creation of port vlan config
self._basic_create_verify_port_vlan('test_replay_unique1',
self.driver_result_unique_add1)
# Make switch inactive before delete
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg.nexus_ip_addr, const.SWITCH_INACTIVE)
# Clean-up the port entry
self._basic_delete_verify_port_vlan('test_replay_unique1',
self.driver_result_unique_del2, nbr_of_bindings=0)
def test_replay_get_nexus_type_failure_two_switches(self):
"""Verifies exception during ncclient get inventory. """
# There are two switches, one active and the other inactive.
# Make sure 'get_nexus_type' fails so create_port_postcommit()
# will return an exception. 'get_nexus_type' is used as
# as ping so even if the switch is marked active then double
# check it is indeed still active. If not and thre are no
# other active switches, then raise exception.
port_cfg1 = TestCiscoNexusReplay.test_configs['test_replay_dual']
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg1.nexus_ip_addr, const.SWITCH_ACTIVE)
port_cfg2 = TestCiscoNexusReplay.test_configs['test_replay_dual2']
self._cisco_mech_driver.set_switch_ip_and_active_state(
port_cfg2.nexus_ip_addr, const.SWITCH_INACTIVE)
# Set-up so get_nexus_type driver fails on active switch
config = {'connect.return_value.get.side_effect':
self._config_side_effects('show inventory',
Exception(__name__))}
self.mock_ncclient.configure_mock(**config)
port_context = self._generate_port_context(port_cfg1)
self.assertRaises(
exceptions.NexusConnectFailed,
self._cisco_mech_driver.create_port_postcommit,
port_context)
def test_replay_get_nexus_type_failure(self):
"""Verifies exception during get nexus_type while replaying. """