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

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 (file)
index 0000000..93b4758
--- /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'])