From 4b8bbe572840a602719494e406b9a07febca3a6a Mon Sep 17 00:00:00 2001 From: Dane LeBlanc Date: Thu, 19 Sep 2013 18:06:37 -0400 Subject: [PATCH] Port Cisco nexus subplugin UT to ML2 mech driver UT Fixes bug 1227889 This fix ports the Cisco Nexus sub-plugin unit tests (neutron/tests/unit/cisco/test_nexus_plugin.py) to the Cisco ML2 mechanism driver. The new test script will be neutron/tests/unit/ml2/drivers/test_cisco_nexus.py. Test cases which will be excluded from this port: - Test cases involving provider networks. - Test cases involving add/remove router networks since these features are not currently supported in the Cisco ML2 mechanism driver. Change-Id: I7aed85031cf402c418089d263c7e949c4631b1eb --- .../unit/ml2/drivers/test_cisco_nexus.py | 201 ++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 neutron/tests/unit/ml2/drivers/test_cisco_nexus.py diff --git a/neutron/tests/unit/ml2/drivers/test_cisco_nexus.py b/neutron/tests/unit/ml2/drivers/test_cisco_nexus.py new file mode 100644 index 000000000..93b4758b8 --- /dev/null +++ b/neutron/tests/unit/ml2/drivers/test_cisco_nexus.py @@ -0,0 +1,201 @@ +# Copyright (c) 2013 OpenStack Foundation. +# +# 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 collections +import mock +import testtools + +from neutron.db import api as db +from neutron.openstack.common import importutils +from neutron.plugins.ml2.drivers.cisco import constants +from neutron.plugins.ml2.drivers.cisco import exceptions +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.tests import base + + +NEXUS_IP_ADDRESS = '1.1.1.1' +NEXUS_IP_ADDRESS_PC = '2.2.2.2' +HOST_NAME_1 = 'testhost1' +HOST_NAME_2 = 'testhost2' +HOST_NAME_PC = 'testpchost' +INSTANCE_1 = 'testvm1' +INSTANCE_2 = 'testvm2' +INSTANCE_PC = 'testpcvm' +NEXUS_PORT_1 = '1/10' +NEXUS_PORT_2 = '1/20' +NEXUS_PORTCHANNELS = 'portchannel:2' +VLAN_ID_1 = 267 +VLAN_ID_2 = 265 +VLAN_ID_PC = 268 +DEVICE_OWNER = 'compute:test' +NEXUS_SSH_PORT = '22' +NEXUS_DRIVER = ('neutron.plugins.ml2.drivers.cisco.' + 'nexus_network_driver.CiscoNexusDriver') + + +class FakeNetworkContext(object): + + """Network context for testing purposes only.""" + + def __init__(self, segment_id): + self._network_segments = [{'segmentation_id': segment_id}] + + @property + def network_segments(self): + return self._network_segments + + +class FakePortContext(object): + + """Port context for testing purposes only.""" + + def __init__(self, device_id, network_context): + self._port = { + 'device_id': device_id, + 'device_owner': DEVICE_OWNER + } + self._network = network_context + + @property + def current(self): + return self._port + + @property + def network(self): + return self._network + + +class TestCiscoNexusDevice(base.BaseTestCase): + + """Unit tests for Cisco ML2 Nexus device driver.""" + + TestConfigObj = collections.namedtuple( + 'TestConfigObj', + 'nexus_ip_addr host_name nexus_port instance_id vlan_id') + + test_configs = { + 'test_config1': TestConfigObj( + NEXUS_IP_ADDRESS, + HOST_NAME_1, + NEXUS_PORT_1, + INSTANCE_1, + VLAN_ID_1), + 'test_config2': TestConfigObj( + NEXUS_IP_ADDRESS, + HOST_NAME_2, + NEXUS_PORT_2, + INSTANCE_2, + VLAN_ID_2), + 'test_config_portchannel': TestConfigObj( + NEXUS_IP_ADDRESS_PC, + HOST_NAME_PC, + NEXUS_PORTCHANNELS, + INSTANCE_PC, + VLAN_ID_PC), + } + + test_credentials = { + NEXUS_IP_ADDRESS: { + constants.USERNAME: 'admin', + constants.PASSWORD: 'pass1234' + }, + NEXUS_IP_ADDRESS_PC: { + constants.USERNAME: 'admin', + constants.PASSWORD: 'password' + }, + } + + def setUp(self): + """Sets up mock ncclient, and switch and credentials dictionaries.""" + super(TestCiscoNexusDevice, self).setUp() + + # Use a mock netconf client + mock_ncclient = mock.Mock() + mock.patch.object(nexus_network_driver.CiscoNexusDriver, + '_import_ncclient', + return_value=mock_ncclient).start() + + def new_nexus_init(mech_instance): + mech_instance.driver = importutils.import_object(NEXUS_DRIVER) + + mech_instance._nexus_switches = {} + for name, config in TestCiscoNexusDevice.test_configs.items(): + ip_addr = config.nexus_ip_addr + host_name = config.host_name + nexus_port = config.nexus_port + mech_instance._nexus_switches[(ip_addr, + host_name)] = nexus_port + mech_instance._nexus_switches[(ip_addr, + 'ssh_port')] = NEXUS_SSH_PORT + mech_instance.driver.nexus_switches = ( + mech_instance._nexus_switches) + + mech_instance.driver.credentials = ( + TestCiscoNexusDevice.test_credentials) + + db.configure_db() + + mock.patch.object(mech_cisco_nexus.CiscoNexusMechanismDriver, + '__init__', new=new_nexus_init).start() + self._cisco_mech_driver = (mech_cisco_nexus. + CiscoNexusMechanismDriver()) + + self.addCleanup(mock.patch.stopall) + self.addCleanup(db.clear_db) + + def _create_delete_port(self, port_config): + """Tests creation and 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 + + network_context = FakeNetworkContext(vlan_id) + port_context = FakePortContext(instance_id, network_context) + + with mock.patch.object(mech_cisco_nexus.CiscoNexusMechanismDriver, + '_get_instance_host') as mock_host: + mock_host.return_value = host_name + + self._cisco_mech_driver.create_port_postcommit(port_context) + bindings = nexus_db_v2.get_nexusport_binding(nexus_port, + vlan_id, + nexus_ip_addr, + instance_id) + self.assertEqual(len(bindings), 1) + + self._cisco_mech_driver.delete_port_precommit(port_context) + with testtools.ExpectedException( + exceptions.NexusPortBindingNotFound): + nexus_db_v2.get_nexusport_binding(nexus_port, + vlan_id, + nexus_ip_addr, + instance_id) + + def test_create_delete_ports(self): + """Tests creation and deletion of two new virtual Ports.""" + self._create_delete_port( + TestCiscoNexusDevice.test_configs['test_config1']) + + self._create_delete_port( + TestCiscoNexusDevice.test_configs['test_config2']) + + def test_create_delete_portchannel(self): + """Tests creation of a port over a portchannel.""" + self._create_delete_port( + TestCiscoNexusDevice.test_configs['test_config_portchannel']) -- 2.45.2