x/vmware-nsx
Code Issues Proposed changes

Merge "Port Cisco nexus subplugin UT to ML2 mech driver UT"

Browse Source
This commit is contained in:
Jenkins
2013-09-25 03:04:45 +00:00
committed by Gerrit Code Review
parent 7cc4037524 4b8bbe5728
commit 8c7e791a9e
1 changed files with 201 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

201
neutron/tests/unit/ml2/drivers/test_cisco_nexus.py Normal file
View File

@@ -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'])