Improve unit test coverage for:
neutron/plugins/ml2/drivers/cisco/ modules.
Current coverage info:
neutron/plugins/ml2/drivers/cisco/nexus/nexus_models_v2 65%
neutron/plugins/ml2/drivers/cisco/nexus/nexus_network_driver 82%
neutron/plugins/ml2/drivers/cisco/nexus/credentials_v2 81%
neutron/plugins/ml2/drivers/cisco/nexus/mech_cisco_nexus 78%
neutron/plugins/ml2/drivers/cisco/nexus/config 75%
With these updates:
neutron/plugins/ml2/drivers/cisco/nexus/mech_cisco_nexus 99%
neutron/plugins/ml2/drivers/cisco/nexus/nexus_db_v2 99%
neutron/plugins/ml2/drivers/cisco/nexus/nexus_models_v2 65%
neutron/plugins/ml2/drivers/cisco/nexus/nexus_network_driver 98%
neutron/plugins/ml2/drivers/cisco/nexus/config 100%
NB: nexus_models_v2.py consists of database definitions and two
customization methods, __repr__ and __eq__.
Change-Id: I0de1cb9e71adffff698e37c8bec5497b88f0f0cd
Closes-Bug:
1267481
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.
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)
--- /dev/null
+# 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)
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.
# 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.
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.
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')
Code Review / openstack-build / neutron-build.git / commitdiff