x/vmware-nsx
Code Issues Proposed changes

Merge "ML2 Cisco Nexus MD: Improve Unit Test Coverage"

Browse Source
This commit is contained in:
Jenkins 2014-04-22 03:09:21 +00:00 committed by Gerrit Code Review
commit 5141c1b0a0
5 changed files with 140 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
neutron/plugins/ml2/drivers/cisco/nexus/nexus_db_v2.py
View File

@ -109,12 +109,6 @@ def get_port_switch_bindings(port_id, switch_ip):
pass
def get_nexussvi_bindings():
"""Lists nexus svi bindings."""
LOG.debug(_("get_nexussvi_bindings() called"))
return _lookup_all_nexus_bindings(port_id='router')
def _lookup_nexus_bindings(query_type, session=None, **bfilter):
"""Look up 'query_type' Nexus bindings matching the filter.

20
neutron/plugins/ml2/drivers/cisco/nexus/nexus_network_driver.py
View File

@ -169,23 +169,3 @@ class CiscoNexusDriver(object):
if nexus_port:
self.enable_vlan_on_trunk_int(nexus_host, vlan_id, intf_type,
nexus_port)
def delete_and_untrunk_vlan(self, nexus_host, vlan_id, intf_type,
nexus_port):
"""Delete VLAN and untrunk it from the specified ports."""
self.delete_vlan(nexus_host, vlan_id)
if nexus_port:
self.disable_vlan_on_trunk_int(nexus_host, vlan_id, intf_type,
nexus_port)
def create_vlan_svi(self, nexus_host, vlan_id, gateway_ip):
confstr = snipp.CMD_VLAN_SVI_SNIPPET % (vlan_id, gateway_ip)
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
self._edit_config(nexus_host, target='running', config=confstr)
def delete_vlan_svi(self, nexus_host, vlan_id):
confstr = snipp.CMD_NO_VLAN_SVI_SNIPPET % vlan_id
confstr = self.create_xml_snippet(confstr)
LOG.debug(_("NexusDriver: %s"), confstr)
self._edit_config(nexus_host, target='running', config=confstr)

75
neutron/tests/unit/ml2/drivers/cisco/nexus/test_cisco_config.py Normal file
View File

@ -0,0 +1,75 @@
# Copyright (c) 2014 Cisco Systems, Inc.
# All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import mock
from oslo.config import cfg
from neutron.common import config as neutron_config
from neutron.plugins.ml2.drivers.cisco.nexus import config as cisco_config
from neutron.tests import base
from neutron.tests.unit import test_api_v2
class TestCiscoNexusPluginConfig(base.BaseTestCase):
def setUp(self):
# Point neutron config file to: neutron/tests/etc/neutron.conf.test
args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
neutron_config.parse(args=args)
super(TestCiscoNexusPluginConfig, self).setUp()
def test_config_parse_error(self):
"""Check that config error is raised upon config parser failure."""
with mock.patch.object(cfg, 'MultiConfigParser') as parser:
parser.return_value.read.return_value = []
self.assertRaises(cfg.Error, cisco_config.ML2MechCiscoConfig)
def test_create_device_dictionary(self):
"""Test creation of the device dictionary based on nexus config."""
test_config = {
'ml2_mech_cisco_nexus:1.1.1.1': {
'username': ['admin'],
'password': ['mySecretPassword'],
'ssh_port': [22],
'compute1': ['1/1'],
'compute2': ['1/2'],
},
'ml2_mech_cisco_nexus:2.2.2.2': {
'username': ['admin'],
'password': ['mySecretPassword'],
'ssh_port': [22],
'compute3': ['1/1'],
'compute4': ['1/2'],
},
}
expected_dev_dict = {
('1.1.1.1', 'username'): 'admin',
('1.1.1.1', 'password'): 'mySecretPassword',
('1.1.1.1', 'ssh_port'): 22,
('1.1.1.1', 'compute1'): '1/1',
('1.1.1.1', 'compute2'): '1/2',
('2.2.2.2', 'username'): 'admin',
('2.2.2.2', 'password'): 'mySecretPassword',
('2.2.2.2', 'ssh_port'): 22,
('2.2.2.2', 'compute3'): '1/1',
('2.2.2.2', 'compute4'): '1/2',
}
with mock.patch.object(cfg, 'MultiConfigParser') as parser:
parser.return_value.read.return_value = cfg.CONF.config_file
parser.return_value.parsed = [test_config]
cisco_config.ML2MechCiscoConfig()
self.assertEqual(expected_dev_dict,
cisco_config.ML2MechCiscoConfig.nexus_dict)

77
neutron/tests/unit/ml2/drivers/cisco/nexus/test_cisco_mech.py
View File

@ -152,6 +152,29 @@ class CiscoML2MechanismTestCase(test_db_plugin.NeutronDbPluginV2TestCase):
config = {attr: None}
self.mock_ncclient.configure_mock(**config)
@staticmethod
def _config_dependent_side_effect(match_config, exc):
"""Generates a config-dependent side effect for ncclient edit_config.
This method generates a mock side-effect function which can be
configured on the mock ncclient module for the edit_config method.
This side effect will cause a given exception to be raised whenever
the XML config string that is passed to edit_config contains all
words in a given match config string.
:param match_config: String containing keywords to be matched
:param exc: Exception to be raised when match is found
:return: Side effect function for the mock ncclient module's
edit_config method.
"""
keywords = match_config.split()
def _side_effect_function(target, config):
if all(word in config for word in keywords):
raise exc
return _side_effect_function
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.
@ -333,6 +356,35 @@ class TestCiscoPortsV2(CiscoML2MechanismTestCase,
# Return to first segment for delete port calls.
self.mock_bound_segment.return_value = BOUND_SEGMENT1
def test_nexus_add_trunk(self):
"""Verify syntax to enable a vlan on an interface.
Test also verifies that the vlan interface is not created.
Test of the following ml2_conf_cisco_ini config:
[ml2_mech_cisco_nexus:1.1.1.1]
hostA=1/1
hostB=1/2
where vlan_id = 100
Confirm that for the first host configured on a Nexus interface,
the command string sent to the switch does not contain the
keyword 'add'.
Confirm that for the second host configured on a Nexus interface,
the command staring sent to the switch contains does not contain
the keyword 'name' [signifies vlan intf creation].
"""
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',
cidr=CIDR_2, host_id=COMP_HOST_NAME_2):
self.assertTrue(
self._is_in_last_nexus_cfg(['allowed', 'vlan']))
self.assertFalse(self._is_in_last_nexus_cfg(['name']))
def test_nexus_connect_fail(self):
"""Test failure to connect to a Nexus switch.
@ -505,18 +557,19 @@ class TestCiscoPortsV2(CiscoML2MechanismTestCase,
for the extended VLAN range).
"""
def mock_edit_config(target, config):
if all(word in config for word in ['state', 'active']):
raise ValueError
with self._patch_ncclient(
'connect.return_value.edit_config.side_effect',
mock_edit_config):
with self._create_resources() as result:
# 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_int,
c_exc.NexusConfigFailed)
vlan_state_configs = ['state active', 'no shutdown']
for config in vlan_state_configs:
with self._patch_ncclient(
'connect.return_value.edit_config.side_effect',
self._config_dependent_side_effect(config, ValueError)):
with self._create_resources() as result:
# 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_int,
c_exc.NexusConfigFailed)
def test_nexus_host_not_configured(self):
"""Test handling of a NexusComputeHostNotConfigured exception.

15
neutron/tests/unit/ml2/drivers/cisco/nexus/test_cisco_nexus_db.py
View File

@ -181,21 +181,6 @@ class CiscoNexusDbTest(base.BaseTestCase):
npb = nexus_db_v2.get_port_switch_bindings(npb21.port, "dummySwitch")
self.assertIsNone(npb)
def test_nexussvibinding_get(self):
"""Tests get of switch virtual interface port bindings."""
npbr1 = self._npb_test_obj('router', 100)
npb21 = self._npb_test_obj(20, 100)
self._add_bindings_to_db([npbr1, npb21])
npb_svi = nexus_db_v2.get_nexussvi_bindings()
self.assertEqual(len(npb_svi), 1)
self._assert_bindings_match(npb_svi[0], npbr1)
npbr2 = self._npb_test_obj('router', 200)
self._add_binding_to_db(npbr2)
npb_svi = nexus_db_v2.get_nexussvi_bindings()
self.assertEqual(len(npb_svi), 2)
def test_nexusbinding_update(self):
"""Tests update of vlan IDs for port bindings."""
npb11 = self._npb_test_obj(10, 100, switch='1.1.1.1', instance='test')