x
/
vmware-nsx
Code Issues Proposed changes

Adding Unit Test Cases Now

This commit is contained in:
Shweta P 2011-08-08 00:23:44 -07:00
parent 0e0631bf3e
commit a5e988e2d6
6 changed files with 2152 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

301
quantum/plugins/cisco/run_tests.py Normal file
View File

@ -0,0 +1,301 @@
#!/usr/bin/env python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2010 OpenStack, LLC
# 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.
# Colorizer Code is borrowed from Twisted:
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""Unittest runner for quantum
To run all test::
python run_tests.py
To run all unit tests::
python run_tests.py unit
To run all functional tests::
python run_tests.py functional
To run a single unit test::
python run_tests.py unit.test_stores:TestSwiftBackend.test_get
To run a single functional test::
python run_tests.py functional.test_service:TestController.test_create
To run a single unit test module::
python run_tests.py unit.test_stores
To run a single functional test module::
python run_tests.py functional.test_stores
"""
import gettext
import logging
import os
import unittest
import sys
from nose import config
from nose import result
from nose import core
class _AnsiColorizer(object):
"""
A colorizer is an object that loosely wraps around a stream, allowing
callers to write text to the stream in a particular color.
Colorizer classes must implement C{supported()} and C{write(text, color)}.
"""
_colors = dict(black=30, red=31, green=32, yellow=33,
blue=34, magenta=35, cyan=36, white=37)
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
"""
A class method that returns True if the current platform supports
coloring terminal output using this method. Returns False otherwise.
"""
if not stream.isatty():
return False # auto color only on TTYs
try:
import curses
except ImportError:
return False
else:
try:
try:
return curses.tigetnum("colors") > 2
except curses.error:
curses.setupterm()
return curses.tigetnum("colors") > 2
except:
raise
# guess false in case of error
return False
supported = classmethod(supported)
def write(self, text, color):
"""
Write the given text to the stream in the given color.
@param text: Text to be written to the stream.
@param color: A string label for a color. e.g. 'red', 'white'.
"""
color = self._colors[color]
self.stream.write('\x1b[%s;1m%s\x1b[0m' % (color, text))
class _Win32Colorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
from win32console import GetStdHandle, STD_OUT_HANDLE, \
FOREGROUND_RED, FOREGROUND_BLUE, FOREGROUND_GREEN, \
FOREGROUND_INTENSITY
red, green, blue, bold = (FOREGROUND_RED, FOREGROUND_GREEN,
FOREGROUND_BLUE, FOREGROUND_INTENSITY)
self.stream = stream
self.screenBuffer = GetStdHandle(STD_OUT_HANDLE)
self._colors = {
'normal': red | green | blue,
'red': red | bold,
'green': green | bold,
'blue': blue | bold,
'yellow': red | green | bold,
'magenta': red | blue | bold,
'cyan': green | blue | bold,
'white': red | green | blue | bold}
def supported(cls, stream=sys.stdout):
try:
import win32console
screenBuffer = win32console.GetStdHandle(
win32console.STD_OUT_HANDLE)
except ImportError:
return False
import pywintypes
try:
screenBuffer.SetConsoleTextAttribute(
win32console.FOREGROUND_RED |
win32console.FOREGROUND_GREEN |
win32console.FOREGROUND_BLUE)
except pywintypes.error:
return False
else:
return True
supported = classmethod(supported)
def write(self, text, color):
color = self._colors[color]
self.screenBuffer.SetConsoleTextAttribute(color)
self.stream.write(text)
self.screenBuffer.SetConsoleTextAttribute(self._colors['normal'])
class _NullColorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
return True
supported = classmethod(supported)
def write(self, text, color):
self.stream.write(text)
class QuantumTestResult(result.TextTestResult):
def __init__(self, *args, **kw):
result.TextTestResult.__init__(self, *args, **kw)
self._last_case = None
self.colorizer = None
# NOTE(vish, tfukushima): reset stdout for the terminal check
stdout = sys.__stdout__
for colorizer in [_Win32Colorizer, _AnsiColorizer, _NullColorizer]:
if colorizer.supported():
self.colorizer = colorizer(self.stream)
break
sys.stdout = stdout
def getDescription(self, test):
return str(test)
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addSuccess(self, test):
unittest.TestResult.addSuccess(self, test)
if self.showAll:
self.colorizer.write("OK", 'green')
self.stream.writeln()
elif self.dots:
self.stream.write('.')
self.stream.flush()
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addFailure(self, test, err):
unittest.TestResult.addFailure(self, test, err)
if self.showAll:
self.colorizer.write("FAIL", 'red')
self.stream.writeln()
elif self.dots:
self.stream.write('F')
self.stream.flush()
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addError(self, test, err):
"""Overrides normal addError to add support for errorClasses.
If the exception is a registered class, the error will be added
to the list for that class, not errors.
"""
stream = getattr(self, 'stream', None)
ec, ev, tb = err
try:
exc_info = self._exc_info_to_string(err, test)
except TypeError:
# This is for compatibility with Python 2.3.
exc_info = self._exc_info_to_string(err)
for cls, (storage, label, isfail) in self.errorClasses.items():
if result.isclass(ec) and issubclass(ec, cls):
if isfail:
test.passwd = False
storage.append((test, exc_info))
# Might get patched into a streamless result
if stream is not None:
if self.showAll:
message = [label]
detail = result._exception_details(err[1])
if detail:
message.append(detail)
stream.writeln(": ".join(message))
elif self.dots:
stream.write(label[:1])
return
self.errors.append((test, exc_info))
test.passed = False
if stream is not None:
if self.showAll:
self.colorizer.write("ERROR", 'red')
self.stream.writeln()
elif self.dots:
stream.write('E')
def startTest(self, test):
unittest.TestResult.startTest(self, test)
current_case = test.test.__class__.__name__
if self.showAll:
if current_case != self._last_case:
self.stream.writeln(current_case)
self._last_case = current_case
self.stream.write(
' %s' % str(test.test._testMethodName).ljust(60))
self.stream.flush()
class QuantumTestRunner(core.TextTestRunner):
def _makeResult(self):
return QuantumTestResult(self.stream,
self.descriptions,
self.verbosity,
self.config)
if __name__ == '__main__':
# Set up test logger.
logger = logging.getLogger()
hdlr = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)
logger.setLevel(logging.DEBUG)
working_dir = os.path.abspath("tests")
c = config.Config(stream=sys.stdout,
env=os.environ,
verbosity=3,
workingDir=working_dir)
runner = QuantumTestRunner(stream=c.stream,
verbosity=c.verbosity,
config=c)
sys.exit(not core.run(config=c, testRunner=runner))

32
quantum/plugins/cisco/tests/unit/__init__.py Normal file
View File

@ -0,0 +1,32 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2011 OpenStack LLC.
# 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.
# See http://code.google.com/p/python-nose/issues/detail?id=373
# The code below enables nosetests to work with i18n _() blocks
import __builtin__
import unittest
setattr(__builtin__, '_', lambda x: x)
class BaseTest(unittest.TestCase):
def setUp(self):
pass
def setUp():
pass

892
quantum/plugins/cisco/tests/unit/test_l2networkApi.py Normal file
View File

@ -0,0 +1,892 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2011 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.
#
# @author: Shweta Padubidri, Cisco Systems, Inc.
#
import logging
import unittest
from quantum.common import exceptions as exc
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.common import cisco_exceptions as cexc
from quantum.plugins.cisco import l2network_plugin
from quantum.plugins.cisco import l2network_plugin_configuration as conf
LOG = logging.getLogger('quantum.tests.test_core_api_func')
class CoreAPITestFunc(unittest.TestCase):
def test_create_network(self, net_tenant_id=None, net_name=None):
"""
Tests creation of new Virtual Network.
"""
LOG.debug("test_create_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if net_name:
network_name = net_name
else:
network_name = self.network_name
new_net_dict = self._l2network_plugin.create_network(
tenant_id, network_name)
self.assertEqual(new_net_dict[const.NET_NAME], network_name)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_create_network - END")
def test_delete_network(self, net_tenant_id=None):
"""
Tests deletion of a Virtual Network.
"""
LOG.debug("test_delete_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
delete_net_dict = self._l2network_plugin.delete_network(
tenant_id, new_net_dict[const.NET_ID])
self.assertEqual(
new_net_dict[const.NET_ID], delete_net_dict[const.NET_ID])
LOG.debug("test_delete_network - END")
def test_delete_networkDNE(self, net_tenant_id=None, net_id='0005'):
"""
Tests deletion of a Virtual Network when Network does not exist.
"""
LOG.debug("test_delete_network_not_found - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(
exc.NetworkNotFound, self._l2network_plugin.delete_network,
tenant_id, net_id)
LOG.debug("test_delete_network_not_found - END")
def test_delete_networkInUse(self, tenant_id='test_network'):
"""
Tests deletion of a Virtual Network when Network is in Use.
"""
LOG.debug("test_delete_networkInUse - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], self.remote_interface)
self.assertRaises(exc.NetworkInUse,
self._l2network_plugin.delete_network, tenant_id,
new_net_dict[const.NET_ID])
self.tearDownNetworkPortInterface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_delete_networkInUse - END")
def test_show_network(self, net_tenant_id=None):
"""
Tests display of details of a Virtual Network .
"""
LOG.debug("test_show_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
result_net_dict = self._l2network_plugin.get_network_details(
tenant_id, new_net_dict[const.NET_ID])
self.assertEqual(
new_net_dict[const.NET_ID], result_net_dict[const.NET_ID])
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_show_network - END")
def test_show_networkDNE(self, net_tenant_id=None, net_id='0005'):
"""
Tests display of a Virtual Network when Network does not exist.
"""
LOG.debug("test_show_network_not_found - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.get_network_details,
tenant_id, net_id)
LOG.debug("test_show_network_not_found - END")
def test_rename_network(self, net_tenant_id=None,
new_name='new_test_network'):
"""
Tests rename of a Virtual Network .
"""
LOG.debug("test_rename_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
rename_net_dict = self._l2network_plugin.rename_network(
tenant_id, new_net_dict[const.NET_ID], new_name)
self.assertEqual(new_name, rename_net_dict[const.NET_NAME])
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_rename_network - END")
def test_rename_networkDNE(self, net_tenant_id=None,
net_id='0005', new_name='new_test_network'):
"""
Tests rename of a Virtual Network when Network does not exist.
"""
LOG.debug("test_rename_network_not_found - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.rename_network,
tenant_id, net_id, new_name)
LOG.debug("test_rename_network_not_found - END")
def test_list_networks(self, tenant_id='test_network'):
"""
Tests listing of all the Virtual Networks .
"""
LOG.debug("test_list_networks - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
new_net_dict2 = self._l2network_plugin.create_network(
tenant_id, 'test_net2')
net_list = self._l2network_plugin.get_all_networks(tenant_id)
net_temp_list = [new_net_dict, new_net_dict2]
self.assertEqual(len(net_list), 2)
self.assertTrue(net_list[0] in net_temp_list)
self.assertTrue(net_list[1] in net_temp_list)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
self.tearDownNetwork(tenant_id, new_net_dict2[const.NET_ID])
LOG.debug("test_list_networks - END")
def test_list_ports(self, tenant_id='test_network'):
"""
Tests listing of all the Ports.
"""
LOG.debug("test_list_ports - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
port_dict2 = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
port_list = self._l2network_plugin.get_all_ports(
tenant_id, new_net_dict[const.NET_ID])
port_temp_list = [port_dict, port_dict2]
self.assertEqual(len(port_list), 2)
self.assertTrue(port_list[0] in port_temp_list)
self.assertTrue(port_list[1] in port_temp_list)
self.tearDownPortOnly(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict2[const.PORT_ID])
LOG.debug("test_list_ports - END")
def test_create_port(self, tenant_id='test_network',
port_state=const.PORT_UP):
"""
Tests creation of Ports.
"""
LOG.debug("test_create_port - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], port_state)
self.assertEqual(port_dict[const.PORT_STATE], port_state)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_create_port - END")
def test_create_port_network_DNE(
self, net_tenant_id=None, net_id='0005', port_state=const.PORT_UP):
"""
Tests creation of Ports when network does not exist.
"""
LOG.debug("test_create_port_network_DNE - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.create_port,
tenant_id, net_id, port_state)
LOG.debug("test_create_port_network_DNE - END:")
def test_delete_port(self, tenant_id='test_tenant',
port_state=const.PORT_UP):
"""
Tests deletion of Ports
"""
LOG.debug("test_delete_port - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], port_state)
delete_port_dict = self._l2network_plugin.delete_port(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
self.assertEqual(delete_port_dict, None)
LOG.debug("test_delete_port - END")
def test_delete_port_networkDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p0005'):
"""
Tests deletion of Ports when network does not exist.
"""
LOG.debug("test_delete_port_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.delete_port, tenant_id,
net_id, port_id)
LOG.debug("test_delete_port_networkDNE - END")
def test_delete_portDNE(self, tenant_id='test_tenant', port_id='p0005'):
"""
Tests deletion of Ports when port does not exist.
"""
LOG.debug("test_delete_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
self.assertRaises(exc.PortNotFound, self._l2network_plugin.delete_port,
tenant_id, new_net_dict[const.NET_ID], port_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_delete_portDNE - END")
def test_delete_portInUse(self, tenant_id='test_tenant'):
"""
Tests deletion of Ports when port is in Use.
"""
LOG.debug("test_delete_portInUse - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.port_state)
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], self.remote_interface)
self.assertRaises(exc.PortInUse,
self._l2network_plugin.delete_port, tenant_id,
new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
self.tearDownNetworkPortInterface(
tenant_id, new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
LOG.debug("test_delete_portInUse - END")
def test_update_port(self, tenant_id='test_tenant',
port_state=const.PORT_DOWN):
"""
Tests updation of Ports.
"""
LOG.debug("test_update_port - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
update_port_dict = self._l2network_plugin.update_port(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_state)
self.assertEqual(update_port_dict[const.PORT_STATE], port_state)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_update_port - END")
def test_update_port_networkDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p0005'):
"""
Tests updation of Ports when network does not exist.
"""
LOG.debug("test_update_port_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.update_port, tenant_id,
net_id, port_id, self.port_state)
LOG.debug("test_update_port_networkDNE - END")
def test_update_portDNE(self, tenant_id='test_tenant', port_id='p0005'):
"""
Tests updation of Ports when port does not exist.
"""
LOG.debug("test_update_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
self.assertRaises(
exc.PortNotFound, self._l2network_plugin.update_port, tenant_id,
new_net_dict[const.NET_ID], port_id, self.port_state)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_update_portDNE - END")
def test_show_port(self, tenant_id='test_tenant'):
"""
Tests display of Ports
"""
LOG.debug("test_show_port - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
get_port_dict = self._l2network_plugin.get_port_details(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertEqual(get_port_dict[const.PORT_STATE], self.port_state)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_show_port - END")
def test_show_port_networkDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p0005'):
"""
Tests display of Ports when network does not exist
"""
LOG.debug("test_show_port_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.get_port_details,
tenant_id, net_id, port_id)
LOG.debug("test_show_port_networkDNE - END")
def test_show_portDNE(self, tenant_id='test_tenant', port_id='p0005'):
"""
Tests display of Ports when port does not exist
"""
LOG.debug("test_show_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
self.assertRaises(exc.PortNotFound,
self._l2network_plugin.get_port_details, tenant_id,
new_net_dict[const.NET_ID], port_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_show_portDNE - END")
def test_plug_interface(self, tenant_id='test_tenant',
remote_interface='new_interface'):
"""
Tests attachment of interface to the port
"""
LOG.debug("test_plug_interface - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], remote_interface)
self.assertEqual(
self._l2network_plugin._networks[new_net_dict[const.NET_ID]]
[const.NET_PORTS][port_dict[const.PORT_ID]]
[const.ATTACHMENT], remote_interface)
self.tearDownNetworkPortInterface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_plug_interface - END")
def test_plug_interface_networkDNE(
self, tenant_id='test_tenant', net_id='0005',
port_id='p0005', remote_interface='new_interface'):
"""
Tests attachment of interface network does not exist
"""
LOG.debug("test_plug_interface_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.plug_interface, tenant_id,
net_id, port_id, remote_interface)
LOG.debug("test_plug_interface_networkDNE - END")
def test_plug_interface_portDNE(self, tenant_id='test_tenant',
port_id='p0005',
remote_interface='new_interface'):
"""
Tests attachment of interface port does not exist
"""
LOG.debug("test_plug_interface_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
self.assertRaises(
exc.PortNotFound, self._l2network_plugin.plug_interface, tenant_id,
new_net_dict[const.NET_ID], port_id, remote_interface)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_plug_interface_portDNE - END")
def test_plug_interface_portInUse(self, tenant_id='test_tenant',
remote_interface='new_interface'):
"""
Tests attachment of new interface to the port when there is an
existing attachment
"""
LOG.debug("test_plug_interface_portInUse - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.port_state)
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], remote_interface)
self.assertRaises(exc.AlreadyAttached,
self._l2network_plugin.plug_interface, tenant_id,
new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], remote_interface)
self.tearDownNetworkPortInterface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_plug_interface_portInUse - END")
def test_unplug_interface(self, tenant_id='test_tenant'):
"""
Tests detaachment of an interface to a port
"""
LOG.debug("test_unplug_interface - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.port_state)
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], self.remote_interface)
self._l2network_plugin.unplug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertEqual(self._l2network_plugin._networks
[new_net_dict[const.NET_ID]][const.NET_PORTS]
[port_dict[const.PORT_ID]][const.ATTACHMENT], None)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_unplug_interface - END")
def test_unplug_interface_networkDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p0005'):
"""
Tests detaachment of an interface to a port, when the network does
not exist
"""
LOG.debug("test_unplug_interface_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.unplug_interface,
tenant_id, net_id, port_id)
LOG.debug("test_unplug_interface_networkDNE - END")
def test_unplug_interface_portDNE(self, tenant_id='test_tenant',
port_id='p0005'):
"""
Tests detaachment of an interface to a port, when the port does
not exist
"""
LOG.debug("test_unplug_interface_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(tenant_id,
self.network_name)
self.assertRaises(exc.PortNotFound,
self._l2network_plugin.unplug_interface, tenant_id,
new_net_dict[const.NET_ID], port_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_unplug_interface_portDNE - END")
def test_create_portprofile(self, net_tenant_id=None,
net_profile_name=None, net_vlan_id=None):
"""
Tests creation of a port-profile
"""
LOG.debug("test_create_portprofile - tenant id: %s - START",
net_tenant_id)
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if net_profile_name:
profile_name = net_profile_name
else:
profile_name = self.profile_name
if net_vlan_id:
vlan_id = net_vlan_id
else:
vlan_id = self.vlan_id
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, profile_name, vlan_id)
port_profile_id = port_profile_dict['profile-id']
self.assertEqual(
self._l2network_plugin._portprofiles[port_profile_id]['vlan-id'],
vlan_id)
self.assertEqual(
self._l2network_plugin._portprofiles[port_profile_id]
['profile-name'], profile_name)
self.tearDownPortProfile(tenant_id, port_profile_id)
LOG.debug("test_create_portprofile - tenant id: %s - END",
net_tenant_id)
def test_delete_portprofile(self, net_tenant_id=None):
"""
Tests deletion of a port-profile
"""
LOG.debug("test_delete_portprofile - tenant id: %s - START",
net_tenant_id)
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
port_profile_id = port_profile_dict['profile-id']
self._l2network_plugin.delete_portprofile(tenant_id, port_profile_id)
self.assertEqual(self._l2network_plugin._portprofiles, {})
LOG.debug("test_delete_portprofile - tenant id: %s - END",
net_tenant_id)
def test_delete_portprofileDNE(self, tenant_id='test_tenant',
profile_id='pr0005'):
"""
Tests deletion of a port-profile when netowrk does not exist
"""
LOG.debug("test_delete_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self._l2network_plugin.delete_portprofile,
tenant_id, profile_id)
LOG.debug("test_delete_portprofileDNE - END")
def test_delete_portprofileAssociated(self, tenant_id='test_tenant'):
"""
Tests deletion of an associatedport-profile
"""
LOG.debug("test_delete_portprofileAssociated - START")
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
port_profile_id = port_profile_dict['profile-id']
self._l2network_plugin.associate_portprofile(
tenant_id, self.net_id, self.port_id, port_profile_id)
self.assertRaises(cexc.PortProfileInvalidDelete,
self._l2network_plugin.delete_portprofile,
tenant_id, port_profile_id)
self.tearDownAssociatePortProfile(tenant_id, self.net_id,
self.port_id, port_profile_id)
LOG.debug("test_delete_portprofileAssociated - END")
def test_list_portprofile(self, tenant_id='test_tenant'):
"""
Tests listing of port-profiles
"""
LOG.debug("test_list_portprofile - tenant id: %s - START", tenant_id)
profile_name2 = tenant_id + '_port_profile2'
vlan_id2 = tenant_id + '201'
port_profile_dict1 = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
port_profile_dict2 = self._l2network_plugin.create_portprofile(
tenant_id, profile_name2, vlan_id2)
port_profile_id1 = port_profile_dict1['profile-id']
port_profile_id2 = port_profile_dict2['profile-id']
list_all_portprofiles = self._l2network_plugin.get_all_portprofiles(
tenant_id)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id1]['vlan-id'], self.vlan_id)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id1]['profile-name'], self.profile_name)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id2]['vlan-id'], vlan_id2)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id2]['profile-name'], profile_name2)
LOG.debug("test_create_portprofile - tenant id: %s - END", tenant_id)
def test_show_portprofile(self, net_tenant_id=None):
"""
Tests display of a port-profile
"""
LOG.debug("test_show_portprofile - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
port_profile_id = port_profile_dict['profile-id']
result_port_profile = self._l2network_plugin.get_portprofile_details(
tenant_id, port_profile_id)
self.assertEqual(result_port_profile[const.PROFILE_VLAN_ID],
self.vlan_id)
self.assertEqual(result_port_profile[const.PROFILE_NAME],
self.profile_name)
self.tearDownPortProfile(tenant_id, port_profile_id)
LOG.debug("test_show_portprofile - tenant id: %s - END", net_tenant_id)
def test_show_portprofileDNE(self, tenant_id='test_tenant',
profile_id='pr0005'):
"""
Tests display of a port-profile when network does not exist
"""
LOG.debug("test_show_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self._l2network_plugin.get_portprofile_details,
tenant_id, profile_id)
LOG.debug("test_show_portprofileDNE - END")
def test_rename_portprofile(self, tenant_id='test_tenant',
new_profile_name='new_profile_name'):
"""
Tests rename of a port-profile
"""
LOG.debug("test_rename_portprofile - START")
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
port_profile_id = port_profile_dict['profile-id']
result_port_profile_dict = self._l2network_plugin.rename_portprofile(
tenant_id, port_profile_id, new_profile_name)
self.assertEqual(result_port_profile_dict[const.PROFILE_NAME],
new_profile_name)
self.tearDownPortProfile(tenant_id, port_profile_id)
LOG.debug("test_show_portprofile - tenant id: %s - END")
def test_rename_portprofileDNE(self, tenant_id='test_tenant',
profile_id='pr0005',
new_profile_name='new_profile_name'):
"""
Tests rename of a port-profile when network does not exist
"""
LOG.debug("test_rename_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self._l2network_plugin.rename_portprofile,
tenant_id, profile_id, new_profile_name)
LOG.debug("test_rename_portprofileDNE - END")
def test_associate_portprofile(self, tenant_id='test_tenant',
net_id='0005', port_id='p00005'):
"""
Tests association of a port-profile
"""
LOG.debug("test_associate_portprofile - START")
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
port_profile_id = port_profile_dict['profile-id']
self._l2network_plugin.associate_portprofile(
tenant_id, net_id, port_id, port_profile_id)
self.assertEqual(
self._l2network_plugin._portprofiles[port_profile_id]
[const.PROFILE_ASSOCIATIONS][0], port_id)
self.tearDownAssociatePortProfile(tenant_id, net_id,
port_id, port_profile_id)
LOG.debug("test_associate_portprofile - END")
def test_associate_portprofileDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p00005',
profile_id='pr0005'):
"""
Tests association of a port-profile when a network does not exist
"""
LOG.debug("test_associate_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self._l2network_plugin.associate_portprofile,
tenant_id, net_id, port_id, profile_id)
LOG.debug("test_associate_portprofileDNE - END")
def test_disassociate_portprofile(self, tenant_id='test_tenant',
net_id='0005', port_id='p00005'):
"""
Tests disassociation of a port-profile
"""
LOG.debug("test_disassociate_portprofile - START")
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
port_profile_id = port_profile_dict['profile-id']
self._l2network_plugin.associate_portprofile(tenant_id, net_id,
port_id, port_profile_id)
self._l2network_plugin.disassociate_portprofile(
tenant_id, net_id, port_id, port_profile_id)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id][const.PROFILE_ASSOCIATIONS], [])
self.tearDownPortProfile(tenant_id, port_profile_id)
LOG.debug("test_disassociate_portprofile - END")
def test_disassociate_portprofileDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p00005', profile_id='pr0005'):
"""
Tests disassociation of a port-profile when network does not exist
"""
LOG.debug("test_disassociate_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self._l2network_plugin.disassociate_portprofile,
tenant_id, net_id, port_id, profile_id)
LOG.debug("test_disassociate_portprofileDNE - END")
# def test_disassociate_portprofile_Unassociated
def test_get_tenant(self, net_tenant_id=None):
"""
Tests get tenant
"""
LOG.debug("test_get_tenant - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
tenant_dict = self._l2network_plugin._get_tenant(tenant_id)
self.assertEqual(tenant_dict[const.TENANT_ID], tenant_id)
self.assertEqual(tenant_dict[const.TENANT_NAME], tenant_id)
LOG.debug("test_get_tenant - END")
def test_get_vlan_name(self, net_tenant_id=None, vlan_name="NewVlan",
vlan_prefix=conf.VLAN_NAME_PREFIX):
"""
Tests get vlan name
"""
LOG.debug("test_get_vlan_name - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
result_vlan_name = self._l2network_plugin._get_vlan_name(tenant_id,
vlan_name)
expected_output = vlan_prefix + tenant_id + "-" + vlan_name
self.assertEqual(result_vlan_name, expected_output)
LOG.debug("test_get_vlan_name - END")
def test_validate_port_state(self, port_state=const.PORT_UP):
"""
Tests validate port state
"""
LOG.debug("test_validate_port_state - START")
result = self._l2network_plugin._validate_port_state(port_state)
self.assertEqual(result, True)
LOG.debug("test_validate_port_state - END")
def test_invalid_port_state(self, port_state="BADSTATE"):
"""
Tests invalidate port state
"""
LOG.debug("test_validate_port_state - START")
self.assertRaises(exc.StateInvalid,
self._l2network_plugin._validate_port_state,
port_state)
LOG.debug("test_validate_port_state - END")
def test_validate_attachment(self, net_tenant_id=None,
remote_interface_id="new_interface"):
"""
Tests validate attachment
"""
LOG.debug("test_validate_attachment - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
net_name = self.network_name
new_network_dict = self._l2network_plugin.create_network(tenant_id,
net_name)
network_id = new_network_dict[const.NET_ID]
new_port_dict = self._l2network_plugin.create_port(tenant_id,
network_id)
port_id = new_port_dict[const.PORT_ID]
self._l2network_plugin.plug_interface(
tenant_id, new_network_dict[const.NET_ID], port_id,
remote_interface_id)
self.assertRaises(exc.AlreadyAttached,
self._l2network_plugin._validate_attachment,
tenant_id, network_id, port_id, remote_interface_id)
self.tearDownNetworkPortInterface(
tenant_id, new_network_dict[const.NET_ID], port_id)
LOG.debug("test_validate_attachment - END")
def setUp(self):
self.tenant_id = "test_tenant"
self.network_name = "test_network"
self.profile_name = "test_tenant_port_profile"
self.vlan_id = "test_tenant_vlanid300"
self.port_state = const.PORT_UP
self.net_id = '00005'
self.port_id = 'p0005'
self.remote_interface = 'new_interface'
self._l2network_plugin = l2network_plugin.L2Network()
"""
Clean up functions after the tests
"""
def tearDownNetwork(self, tenant_id, network_dict_id):
self._l2network_plugin.delete_network(tenant_id, network_dict_id)
def tearDownPortOnly(self, tenant_id, network_dict_id, port_id):
self._l2network_plugin.delete_port(tenant_id, network_dict_id, port_id)
def tearDownNetworkPort(self, tenant_id, network_dict_id, port_id):
self._l2network_plugin.delete_port(tenant_id, network_dict_id, port_id)
self.tearDownNetwork(tenant_id, network_dict_id)
def tearDownNetworkPortInterface(self, tenant_id, network_dict_id,
port_id):
self._l2network_plugin.unplug_interface(tenant_id,
network_dict_id, port_id)
self.tearDownNetworkPort(tenant_id, network_dict_id, port_id)
def tearDownPortProfile(self, tenant_id, port_profile_id):
self._l2network_plugin.delete_portprofile(tenant_id, port_profile_id)
def tearDownAssociatePortProfile(self, tenant_id, net_id, port_id,
port_profile_id):
self._l2network_plugin.disassociate_portprofile(
tenant_id, net_id, port_id, port_profile_id)
self.tearDownPortProfile(tenant_id, port_profile_id)

282
quantum/plugins/cisco/tests/unit/test_nexus_plugin.py Normal file
View File

@ -0,0 +1,282 @@
# copyright 2011 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.
#
# @author: Shweta Padubidri, Peter Strunk, Cisco Systems, Inc.
#
import unittest
import logging
from quantum.common import exceptions as exc
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.nexus import cisco_nexus_plugin
LOG = logging.getLogger('quantum.tests.test_nexus')
class TestNexusPlugin(unittest.TestCase):
def setUp(self):
self.tenant_id = "test_tenant_cisco1"
self.net_name = "test_network_cisco1"
self.net_id = 000007
self.vlan_name = "q-" + str(self.net_id) + "vlan"
self.vlan_id = 267
self.port_id = "9"
self._cisco_nexus_plugin = cisco_nexus_plugin.NexusPlugin()
def test_create_network(self, net_tenant_id=None, network_name=None,
network_id=None, net_vlan_name=None,
net_vlan_id=None):
"""
Tests creation of new Virtual Network.
"""
LOG.debug("test_create_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if network_name:
net_name = network_name
else:
net_name = self.net_name
if network_id:
net_id = network_id
else:
net_id = self.net_id
if net_vlan_name:
vlan_name = net_vlan_name
else:
vlan_name = self.vlan_name
if net_vlan_id:
vlan_id = net_vlan_id
else:
vlan_id = self.vlan_id
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, net_name, net_id, vlan_name, vlan_id)
self.assertEqual(new_net_dict[const.NET_ID], self.net_id)
self.assertEqual(new_net_dict[const.NET_NAME], self.net_name)
self.assertEqual(new_net_dict[const.NET_VLAN_NAME], self.vlan_name)
self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.vlan_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_create_network - END")
def test_delete_network(self, net_tenant_id=None, network_id=None):
"""
Tests deletion of a Virtual Network.
"""
LOG.debug("test_delete_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if network_id:
net_id = network_id
else:
net_id = self.net_id
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, self.net_name, net_id, self.vlan_name, self.vlan_id)
deleted_net_dict = self._cisco_nexus_plugin.delete_network(
tenant_id, new_net_dict[const.NET_ID])
self.assertEqual(deleted_net_dict[const.NET_ID], net_id)
LOG.debug("test_delete_network - END")
def test_delete_network_DNE(self, net_tenant_id=None, net_id='0005'):
"""
Tests deletion of a Virtual Network when Network does not exist.
"""
LOG.debug("test_delete_network_DNE - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
self._cisco_nexus_plugin.delete_network,
tenant_id, net_id)
LOG.debug("test_delete_network_DNE - END")
def test_get_network_details(self, net_tenant_id=None, network_id=None):
"""
Tests displays details of a Virtual Network .
"""
LOG.debug("test_get_network_details - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if network_id:
net_id = network_id
else:
net_id = self.net_id
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, self.net_name, net_id, self.vlan_name, self.vlan_id)
check_net_dict = self._cisco_nexus_plugin.get_network_details(
tenant_id, net_id)
self.assertEqual(check_net_dict[const.NET_ID], net_id)
self.assertEqual(check_net_dict[const.NET_NAME], self.net_name)
self.assertEqual(check_net_dict[const.NET_VLAN_NAME], self.vlan_name)
self.assertEqual(check_net_dict[const.NET_VLAN_ID], self.vlan_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_get_network_details - END")
def test_get_networkDNE(self, net_tenant_id=None, net_id='0005'):
"""
Tests display of a Virtual Network when Network does not exist.
"""
LOG.debug("test_get_network_details_network_does_not_exist - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
self._cisco_nexus_plugin.get_network_details,
tenant_id, net_id)
LOG.debug("test_get_network_details_network_does_not_exist - END")
def test_rename_network(self, new_name="new_network_name",
net_tenant_id=None, network_id=None):
"""
Tests rename of a Virtual Network .
"""
LOG.debug("test_rename_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if network_id:
net_id = network_id
else:
net_id = self.net_id
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, self.net_name, net_id, self.vlan_name,
self.vlan_id)
rename_net_dict = self._cisco_nexus_plugin.rename_network(
tenant_id, new_net_dict[const.NET_ID], new_name)
self.assertEqual(rename_net_dict[const.NET_NAME], new_name)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_rename_network - END")
def test_rename_network_DNE(self, new_name="new_network_name",
net_tenant_id=None, network_id='0005'):
"""
Tests rename of a Virtual Network when Network does not exist.
"""
LOG.debug("test_rename_network_DNE - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if network_id:
net_id = network_id
else:
net_id = self.net_id
self.assertRaises(exc.NetworkNotFound,
self._cisco_nexus_plugin.rename_network,
new_name, tenant_id, net_id)
LOG.debug("test_rename_network_DNE - END")
def test_list_all_networks(self, net_tenant_id=None):
"""
Tests listing of all the Virtual Networks .
"""
LOG.debug("test_list_all_networks - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
new_net_dict1 = self._cisco_nexus_plugin.create_network(
tenant_id, self.net_name, self.net_id,
self.vlan_name, self.vlan_id)
new_net_dict2 = self._cisco_nexus_plugin.create_network(
tenant_id, "New_Network2", "0011",
"second_vlan", "2003")
list_net_dict = self._cisco_nexus_plugin.get_all_networks(tenant_id)
net_temp_list = [new_net_dict1, new_net_dict2]
self.assertEqual(len(list_net_dict), 2)
self.assertTrue(list_net_dict[0] in net_temp_list)
self.assertTrue(list_net_dict[1] in net_temp_list)
self.tearDownNetwork(tenant_id, new_net_dict1[const.NET_ID])
self.tearDownNetwork(tenant_id, new_net_dict2[const.NET_ID])
LOG.debug("test_list_all_networks - END")
def test_get_vlan_id_for_network(self, net_tenant_id=None,
network_id=None):
"""
Tests retrieval of vlan id for a Virtual Networks .
"""
LOG.debug("test_get_vlan_id_for_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if network_id:
net_id = network_id
else:
net_id = self.net_id
new_net_dict = self._cisco_nexus_plugin.create_network(
tenant_id, self.net_name, net_id, self.vlan_name,
self.vlan_id)
result_vlan_id = self._cisco_nexus_plugin._get_vlan_id_for_network(
tenant_id, net_id)
self.assertEqual(result_vlan_id, self.vlan_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_get_vlan_id_for_network - END")
"""
Clean up functions after the tests
"""
def tearDownNetwork(self, tenant_id, network_dict_id):
self._cisco_nexus_plugin.delete_network(tenant_id, network_dict_id)
# def test_create_network(self):
# _test_create_network(self._cisco_nexus_plugin)
# def test_delete_network(self):
# _test_delete_network(self._cisco_nexus_plugin)
# def test_rename_network(self):
# _test_rename_network(self._cisco_nexus_plugin)
# def test_show_network(self):
# _test_get_network_details(self._cisco_nexus_plugin)
# def test_list_networks(self):
# _test_list_all_networks(self._cisco_nexus_plugin)

165
quantum/plugins/cisco/tests/unit/test_ucs_driver.py Normal file
View File

@ -0,0 +1,165 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2011 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.
#
# @author: Shweta Padubidri, Cisco Systems, Inc.
#
import logging
import unittest
from quantum.plugins.cisco.ucs import cisco_ucs_network_driver
LOG = logging.getLogger('quantum.tests.test_ucs_driver')
create_vlan_output = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs><pair key=\"fabric/lan/net-New Vlan\"> "\
"<fabricVlan defaultNet=\"no\" dn=\"fabric/lan/net-New Vlan\" id=\"200\" "\
"name=\"New Vlan\" status=\"created\"></fabricVlan> </pair> </inConfigs> "\
"</configConfMos>"
create_profile_output = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs><pair key=\"fabric/lan/profiles/vnic-"\
"New Profile\"> <vnicProfile descr=\"Profile created by Cisco OpenStack "\
"Quantum Plugin\" dn=\"fabric/lan/profiles/vnic-New Profile\" maxPorts="\
"\"64\" name=\"New Profile\" nwCtrlPolicyName=\"\" pinToGroupName=\"\" "\
"qosPolicyName=\"\" status=\"created\"> <vnicEtherIf defaultNet=\"yes\" "\
"name=\"New Vlan\" rn=\"if-New Vlan\" > </vnicEtherIf> </vnicProfile> "\
"</pair> </inConfigs> </configConfMos>"
change_vlan_output = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs><pair key=\""\
"fabric/lan/profiles/vnic-New Profile\"> <vnicProfile descr=\"Profile "\
"created by Cisco OpenStack Quantum Plugin\" "\
"dn=\"fabric/lan/profiles/vnic-New Profile\" maxPorts=\"64\" "\
"name=\"New Profile\" nwCtrlPolicyName=\"\" pinToGroupName=\"\" "\
"qosPolicyName=\"\" status=\"created,modified\"><vnicEtherIf "\
"rn=\"if-Old Vlan\" status=\"deleted\"> </vnicEtherIf> "\
"<vnicEtherIf defaultNet=\"yes\" name=\"New Vlan\" rn=\"if-New Vlan\" > "\
"</vnicEtherIf> </vnicProfile> </pair></inConfigs> </configConfMos>"
delete_vlan_output = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs><pair key=\"fabric/lan/net-New Vlan\"> "\
"<fabricVlan dn=\"fabric/lan/net-New Vlan\" status=\"deleted\"> "\
"</fabricVlan> </pair> </inConfigs></configConfMos>"
delete_profile_output = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"false\"> <inConfigs><pair key=\""\
"fabric/lan/profiles/vnic-New Profile\"> <vnicProfile "\
"dn=\"fabric/lan/profiles/vnic-New Profile\" status=\"deleted\"> "\
"</vnicProfile></pair> </inConfigs> </configConfMos>"
associate_profile_output = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs> <pair key="\
"\"fabric/lan/profiles/vnic-New Profile/cl-New Profile Client\">"\
" <vmVnicProfCl dcName=\".*\" descr=\"\" dn=\"fabric/lan/profiles/vnic-"\
"New Profile/cl-New Profile Client\"name=\"New Profile Client\" "\
"orgPath=\".*\" status=\"created\" swName=\"default$\"> </vmVnicProfCl>" \
"</pair> </inConfigs> </configConfMos>"
class TestUCSDriver(unittest.TestCase):
def setUp(self):
self._ucsmDriver = cisco_ucs_network_driver.CiscoUCSMDriver()
self.vlan_name = 'New Vlan'
self.vlan_id = '200'
self.profile_name = 'New Profile'
self.old_vlan_name = 'Old Vlan'
self.profile_client_name = 'New Profile Client'
def test_create_vlan_post_data(self, expected_output=create_vlan_output):
"""
Tests creation of vlan post Data
"""
LOG.debug("test_create_vlan")
vlan_details = self._ucsmDriver._create_vlan_post_data(
self.vlan_name, self.vlan_id)
self.assertEqual(vlan_details, expected_output)
LOG.debug("test_create_vlan - END")
def test_create_profile_post_data(
self, expected_output=create_profile_output):
"""
Tests creation of profile post Data
"""
LOG.debug("test_create_profile_post_data - START")
profile_details = self._ucsmDriver._create_profile_post_data(
self.profile_name, self.vlan_name)
self.assertEqual(profile_details, expected_output)
LOG.debug("test_create_profile_post - END")
def test_change_vlan_in_profile_post_data(
self, expected_output=change_vlan_output):
"""
Tests creation of change vlan in profile post Data
"""
LOG.debug("test_create_profile_post_data - START")
profile_details = self._ucsmDriver._change_vlan_in_profile_post_data(
self.profile_name, self.old_vlan_name, self.vlan_name)
self.assertEqual(profile_details, expected_output)
LOG.debug("test_create_profile_post - END")
def test_delete_vlan_post_data(self, expected_output=delete_vlan_output):
LOG.debug("test_create_profile_post_data - START")
"""
Tests deletion of vlan post Data
"""
vlan_details = self._ucsmDriver._create_vlan_post_data(
self.vlan_name, self.vlan_id)
vlan_delete_details = self._ucsmDriver._delete_vlan_post_data(
self.vlan_name)
self.assertEqual(vlan_delete_details, expected_output)
LOG.debug("test_create_profile_post - END")
def test_delete_profile_post_data(
self, expected_output=delete_profile_output):
"""
Tests deletion of profile post Data
"""
LOG.debug("test_create_profile_post_data - START")
profile_details = self._ucsmDriver._create_profile_post_data(
self.profile_name, self.vlan_name)
profile_delete_details = self._ucsmDriver._delete_profile_post_data(
self.profile_name)
self.assertEqual(profile_delete_details, expected_output)
LOG.debug("test_create_profile_post - END")
def test_create_profile_client_post_data(
self, expected_output=associate_profile_output):
"""
Tests creation of profile client post Data
"""
LOG.debug("test_create_profile_client_post_data - START")
profile_details = self._ucsmDriver._create_profile_client_post_data(
self.profile_name, self.profile_client_name)
self.assertEqual(profile_details, expected_output)
LOG.debug("test_create_profile_post - END")
def test_get_next_dynamic_nic(self):
"""
Tests get next dynamic nic
"""
LOG.debug("test_get_next_dynamic_nic - START")
dynamic_nic_id = self._ucsmDriver._get_next_dynamic_nic()
self.assertTrue(len(dynamic_nic_id) > 0)
LOG.debug("test_get_next_dynamic_nic - END")

480
quantum/plugins/cisco/tests/unit/test_ucs_plugin.py Normal file
View File

@ -0,0 +1,480 @@
#vim: tabstop=4 shiftwidth=4 softtabstop=4
#copyright 2011 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.
#
# @author: Shubhangi Satras, Cisco Systems, Inc.
#
import unittest
import logging as LOG
from quantum.common import exceptions as exc
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.ucs import cisco_ucs_plugin
from quantum.plugins.cisco.ucs import cisco_ucs_configuration as conf
LOG.basicConfig(level=LOG.WARN)
LOG.getLogger("cisco_plugin")
class UCSVICTestPlugin(unittest.TestCase):
def setUp(self):
self.tenant_id = "test_tenant_cisco12"
self.net_name = "test_network_cisco12"
self.net_id = 000007
self.vlan_name = "q-" + str(self.net_id) + "vlan"
self.vlan_id = 266
self.port_id = "4"
self._cisco_ucs_plugin = cisco_ucs_plugin.UCSVICPlugin()
def test_create_network(self):
"""
Tests creation of new Virtual Network.
"""
LOG.debug("UCSVICTestPlugin:_test_create_network() called\n")
new_net_dict = self._cisco_ucs_plugin.create_network(
self.tenant_id, self.net_name, self.net_id,
self.vlan_name, self.vlan_id)
self.assertEqual(new_net_dict[const.NET_ID], self.net_id)
self.assertEqual(new_net_dict[const.NET_NAME], self.net_name)
self.assertEqual(new_net_dict[const.NET_VLAN_NAME], self.vlan_name)
self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.vlan_id)
self.tearDownNetwork(self.tenant_id, self.net_id)
def test_delete_network(self):
"""
Tests deletion of the network with the specified network identifier
belonging to the specified tenant.
"""
LOG.debug("UCSVICTestPlugin:test_delete_network() called\n")
self._cisco_ucs_plugin.create_network(
self.tenant_id, self.net_name, self.net_id,
self.vlan_name, self.vlan_id)
new_net_dict = self._cisco_ucs_plugin.delete_network(
self.tenant_id, self.net_id)
self.assertEqual(new_net_dict[const.NET_ID], self.net_id)
def test_get_network_details(self):
"""
Tests the deletion the Virtual Network belonging to a the
spec
"""
LOG.debug("UCSVICTestPlugin:test_get_network_details() called\n")
self._cisco_ucs_plugin.create_network(
self.tenant_id, self.net_name, self.net_id,
self.vlan_name, self.vlan_id)
new_net_dict = self._cisco_ucs_plugin.get_network_details(
self.tenant_id, self.net_id)
self.assertEqual(new_net_dict[const.NET_ID], self.net_id)
self.assertEqual(new_net_dict[const.NET_VLAN_NAME], self.vlan_name)
self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.vlan_id)
self.tearDownNetwork(self.tenant_id, self.net_id)
def test_get_all_networks(self):
"""
Tests whether dictionary is returned containing all
<network_uuid, network_name> for
the specified tenant.
"""
LOG.debug("UCSVICTestPlugin:test_get_all_networks() called\n")
new_net_dict1 = self._cisco_ucs_plugin.create_network(
self.tenant_id, self.net_name, self.net_id,
self.vlan_name, self.vlan_id)
new_net_dict2 = self._cisco_ucs_plugin.create_network(
self.tenant_id, "test_network2",
000006, "q-000006vlan", "6")
net_list = self._cisco_ucs_plugin.get_all_networks(self.tenant_id)
net_id_list = [new_net_dict1, new_net_dict2]
self.assertTrue(net_list[0] in net_id_list)
self.assertTrue(net_list[1] in net_id_list)
self.tearDownNetwork(self.tenant_id, new_net_dict1[const.NET_ID])
self.tearDownNetwork(self.tenant_id, new_net_dict2[const.NET_ID])
def test_get_all_ports(self):
"""
Retrieves all port identifiers belonging to the
specified Virtual Network.
"""
LOG.debug("UCSVICPlugin:get_all_ports() called\n")
new_net_dict = self._cisco_ucs_plugin.create_network(
self.tenant_id, self.net_name, self.net_id,
self.vlan_name, self.vlan_id)
port_dict1 = self._cisco_ucs_plugin.create_port(
self.tenant_id, self.net_id, const.PORT_UP,
self.port_id)
port_dict2 = self._cisco_ucs_plugin.create_port(
self.tenant_id, self.net_id,
const.PORT_UP, "10")
ports_on_net = self._cisco_ucs_plugin.get_all_ports(
self.tenant_id, self.net_id)
port_list = [port_dict1, port_dict2]
self.assertTrue(port_list[0] in ports_on_net)
self.assertTrue(port_list[1] in ports_on_net)
self._cisco_ucs_plugin.delete_port(self.tenant_id, self.net_id,
self.port_id)
self.tearDownNetworkPort(self.tenant_id, new_net_dict[const.NET_ID],
port_dict2[const.PORT_ID])
def _test_rename_network(self, new_name):
"""
Tests whether symbolic name is updated for the particular
Virtual Network.
"""
LOG.debug("UCSVICTestPlugin:_test_rename_network() called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
new_net_dict = self._cisco_ucs_plugin.rename_network(
self.tenant_id, self.net_id, new_name)
self.assertEqual(new_net_dict[const.NET_NAME], new_name)
self.tearDownNetwork(self.tenant_id, self.net_id)
def test_rename_network(self):
self._test_rename_network("new_test_network1")
def _test_create_port(self, port_state):
"""
Tests creation of a port on the specified Virtual Network.
"""
LOG.debug("UCSVICTestPlugin:_test_create_port() called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
new_port_dict = self._cisco_ucs_plugin.create_port(
self.tenant_id, self.net_id, port_state, self.port_id)
self.assertEqual(new_port_dict[const.PORT_ID], self.port_id)
self.assertEqual(new_port_dict[const.PORT_STATE], port_state)
self.assertEqual(new_port_dict[const.ATTACHMENT], None)
profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id)
new_port_profile = new_port_dict[const.PORT_PROFILE]
self.assertEqual(new_port_profile[const.PROFILE_NAME], profile_name)
self.assertEqual(new_port_profile[const.PROFILE_VLAN_NAME],
conf.DEFAULT_VLAN_NAME)
self.assertEqual(new_port_profile[const.PROFILE_VLAN_ID],
conf.DEFAULT_VLAN_ID)
self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id)
def test_create_port(self):
self._test_create_port(const.PORT_UP)
def _test_delete_port(self, port_state):
"""
Tests Deletion of a port on a specified Virtual Network,
if the port contains a remote interface attachment,
the remote interface should first be un-plugged and
then the port can be deleted.
"""
LOG.debug("UCSVICTestPlugin:_test_delete_port() called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id,
port_state, self.port_id)
self._cisco_ucs_plugin.delete_port(self.tenant_id, self.net_id,
self.port_id)
net = self._cisco_ucs_plugin._get_network(self.tenant_id, self.net_id)
self.assertEqual(net[const.NET_PORTS], {})
self.tearDownNetwork(self.tenant_id, self.net_id)
def test_delete_port(self):
self._test_delete_port(const.PORT_UP)
def _test_update_port(self, port_state):
"""
Tests Updation of the state of a port on the specified Virtual Network.
"""
LOG.debug("UCSVICTestPlugin:_test_update_port() called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id,
port_state, self.port_id)
port = self._cisco_ucs_plugin.update_port(
self.tenant_id, self.net_id,
self.port_id, port_state)
self.assertEqual(port[const.PORT_STATE], port_state)
self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id)
def test_update_port_state_up(self):
self._test_update_port(const.PORT_UP)
def test_update_port_state_down(self):
self._test_update_port(const.PORT_DOWN)
def _test_get_port_details_state_up(self, port_state):
"""
Tests whether user is able to retrieve a remote interface
that is attached to this particular port when port state is Up.
"""
LOG.debug("UCSVICTestPlugin:_test_get_port_details_state_up()" +
"called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id,
port_state, self.port_id)
port = self._cisco_ucs_plugin.get_port_details(
self.tenant_id, self.net_id, self.port_id)
self.assertEqual(port[const.PORT_ID], self.port_id)
self.assertEqual(port[const.PORT_STATE], port_state)
self.assertEqual(port[const.ATTACHMENT], None)
new_port_profile = port[const.PORT_PROFILE]
profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id)
self.assertEqual(new_port_profile[const.PROFILE_VLAN_NAME],
conf.DEFAULT_VLAN_NAME)
self.assertEqual(new_port_profile[const.PROFILE_VLAN_ID],
conf.DEFAULT_VLAN_ID)
self.assertEqual(new_port_profile[const.PROFILE_NAME], profile_name)
self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id)
def _test_get_port_details_state_down(self, port_state):
"""
Tests whether user is able to retrieve a remote interface
that is attached to this particular port when port state is down.
"""
LOG.debug("UCSVICTestPlugin:_test_get_port_details_state_down()" +
"called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id,
port_state, self.port_id)
port = self._cisco_ucs_plugin.get_port_details(self.tenant_id,
self.net_id,
self.port_id)
self.assertEqual(port[const.PORT_ID], self.port_id)
self.assertNotEqual(port[const.PORT_STATE], port_state)
self.assertEqual(port[const.ATTACHMENT], None)
new_port_profile = port[const.PORT_PROFILE]
profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id)
self.assertEqual(new_port_profile[const.PROFILE_VLAN_NAME],
conf.DEFAULT_VLAN_NAME)
self.assertEqual(new_port_profile[const.PROFILE_VLAN_ID],
conf.DEFAULT_VLAN_ID)
self.assertEqual(new_port_profile[const.PROFILE_NAME], profile_name)
self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id)
def test_get_port_details_state_up(self):
self._test_get_port_details_state_up(const.PORT_UP)
def test_get_port_details_state_down(self):
self._test_get_port_details_state_down(const.PORT_DOWN)
def test_create_port_profile(self):
LOG.debug("UCSVICTestPlugin:test_create_port_profile() called\n")
new_port_profile = self._cisco_ucs_plugin._create_port_profile(
self.tenant_id, self.net_id, self.port_id,
self.vlan_name, self.vlan_id)
profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id)
self.assertEqual(new_port_profile[const.PROFILE_NAME], profile_name)
self.assertEqual(new_port_profile[const.PROFILE_VLAN_NAME],
self.vlan_name)
self.assertEqual(new_port_profile[const.PROFILE_VLAN_ID], self.vlan_id)
self._cisco_ucs_plugin._delete_port_profile(self.port_id, profile_name)
def test_delete_port_profile(self):
LOG.debug("UCSVICTestPlugin:test_delete_port_profile() called\n")
self._cisco_ucs_plugin._create_port_profile(
self.tenant_id, self.net_id, self.port_id, self.vlan_name,
self.vlan_id)
profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id)
counter1 = self._cisco_ucs_plugin._port_profile_counter
self._cisco_ucs_plugin._delete_port_profile(self.port_id,
profile_name)
counter2 = self._cisco_ucs_plugin._port_profile_counter
self.assertNotEqual(counter1, counter2)
def _test_plug_interface(self, remote_interface_id):
"""
Attaches a remote interface to the specified port on the
specified Virtual Network.
"""
LOG.debug("UCSVICTestPlugin:_test_plug_interface() called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id,
const.PORT_UP, self.port_id)
self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id,
self.port_id,
remote_interface_id)
port = self._cisco_ucs_plugin._get_port(
self.tenant_id, self.net_id, self.port_id)
self.assertEqual(port[const.ATTACHMENT], remote_interface_id)
port_profile = port[const.PORT_PROFILE]
profile_name = port_profile[const.PROFILE_NAME]
new_vlan_name = self._cisco_ucs_plugin._get_vlan_name_for_network(
self.tenant_id, self.net_id)
new_vlan_id = self._cisco_ucs_plugin._get_vlan_id_for_network(
self.tenant_id, self.net_id)
self.assertEqual(port_profile[const.PROFILE_VLAN_NAME], new_vlan_name)
self.assertEqual(port_profile[const.PROFILE_VLAN_ID], new_vlan_id)
self.tearDownNetworkPortInterface(self.tenant_id, self.net_id,
self.port_id)
def test_plug_interface(self):
self._test_plug_interface("4")
def _test_unplug_interface(self, remote_interface_id):
"""
Tests whether remote interface detaches from the specified port on the
specified Virtual Network.
"""
LOG.debug("UCSVICTestPlugin:_test_unplug_interface() called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id,
const.PORT_UP, self.port_id)
self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id,
self.port_id,
remote_interface_id)
self._cisco_ucs_plugin.unplug_interface(self.tenant_id, self.net_id,
self.port_id)
port = self._cisco_ucs_plugin._get_port(
self.tenant_id, self.net_id, self.port_id)
self.assertEqual(port[const.ATTACHMENT], None)
port_profile = port[const.PORT_PROFILE]
profile_name = port_profile[const.PROFILE_NAME]
self.assertEqual(port_profile[const.PROFILE_VLAN_NAME],
conf.DEFAULT_VLAN_NAME)
self.assertEqual(port_profile[const.PROFILE_VLAN_ID],
conf.DEFAULT_VLAN_ID)
self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id)
def test_unplug_interface(self):
self._test_unplug_interface("4")
def test_get_vlan_name_for_network(self):
LOG.debug("UCSVICTestPlugin:test_get_vlan_name_for_network() called\n")
net = self._cisco_ucs_plugin.create_network(
self.tenant_id, self.net_name, self.net_id,
self.vlan_name, self.vlan_id)
self.assertEqual(net[const.NET_VLAN_NAME], self.vlan_name)
self.tearDownNetwork(self.tenant_id, self.net_id)
def test_get_vlan_id_for_network(self):
LOG.debug("UCSVICTestPlugin:test_get_vlan_id_for_network() called\n")
net = self._cisco_ucs_plugin.create_network(
self.tenant_id, self.net_name, self.net_id, self.vlan_name,
self.vlan_id)
self.assertEqual(net[const.NET_VLAN_ID], self.vlan_id)
self.tearDownNetwork(self.tenant_id, self.net_id)
def test_get_network(self):
LOG.debug("UCSVICTestPlugin:test_get_network() called\n")
net = self._cisco_ucs_plugin.create_network(
self.tenant_id, self.net_name, self.net_id, self.vlan_name,
self.vlan_id)
self.assertEqual(net[const.NET_ID], self.net_id)
self.tearDownNetwork(self.tenant_id, self.net_id)
def test_get_port(self):
LOG.debug("UCSVICTestPlugin:test_get_port() called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
new_port_dict = self._cisco_ucs_plugin.create_port(
self.tenant_id, self.net_id,
const.PORT_UP, self.port_id)
self.assertEqual(new_port_dict[const.PORT_ID], self.port_id)
self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id)
def test_get_network_NetworkNotFound(self):
self.assertRaises(exc.NetworkNotFound,
self._cisco_ucs_plugin._get_network,
*(self.tenant_id, self.net_id))
def test_delete_network_NetworkNotFound(self):
self.assertRaises(exc.NetworkNotFound,
self._cisco_ucs_plugin.delete_network,
*(self.tenant_id, self.net_id))
def test_delete_port_PortInUse(self):
self._test_delete_port_PortInUse("4")
def _test_delete_port_PortInUse(self, remote_interface_id):
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id,
const.PORT_UP, self.port_id)
self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id,
self.port_id,
remote_interface_id)
self.assertRaises(exc.PortInUse, self._cisco_ucs_plugin.delete_port,
*(self.tenant_id, self.net_id, self.port_id))
self.tearDownNetworkPortInterface(self.tenant_id, self.net_id,
self.port_id)
def test_delete_port_PortNotFound(self):
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self.assertRaises(exc.PortNotFound, self._cisco_ucs_plugin.delete_port,
*(self.tenant_id, self.net_id, self.port_id))
self.tearDownNetwork(self.tenant_id, self.net_id)
def test_plug_interface_PortInUse(self):
self._test_plug_interface_PortInUse("6", "5")
def _test_plug_interface_PortInUse(self, remote_interface_id1,
remote_interface_id2):
LOG.debug("UCSVICTestPlugin:_test_plug_interface_PortInUse() called\n")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id,
const.PORT_UP, self.port_id)
self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id,
self.port_id,
remote_interface_id1)
self.assertRaises(exc.PortInUse, self._cisco_ucs_plugin.plug_interface,
*(self.tenant_id, self.net_id, self.port_id,
remote_interface_id2))
self.tearDownNetworkPortInterface(self.tenant_id, self.net_id,
self.port_id)
def test_validate_attachment_AlreadyAttached(self):
LOG.debug("UCSVICTestPlugin:testValidateAttachmentAlreadyAttached")
self._test_validate_attachment_AlreadyAttached("4")
def _test_validate_attachment_AlreadyAttached(self, remote_interface_id):
LOG.debug("UCSVICTestPlugin:_test_validate_attachmentAlreadyAttached")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
self.vlan_id)
self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id,
const.PORT_UP, self.port_id)
self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id,
self.port_id,
remote_interface_id)
self.assertRaises(
exc.AlreadyAttached, self._cisco_ucs_plugin._validate_attachment,
*(self.tenant_id, self.net_id, self.port_id, remote_interface_id))
self.tearDownNetworkPortInterface(self.tenant_id, self.net_id,
self.port_id)
def tearDownNetwork(self, tenant_id, net_id):
self._cisco_ucs_plugin.delete_network(tenant_id, net_id)
def tearDownNetworkPort(self, tenant_id, net_id, port_id):
self._cisco_ucs_plugin.delete_port(tenant_id, net_id,
port_id)
self.tearDownNetwork(tenant_id, net_id)
def tearDownNetworkPortInterface(self, tenant_id, net_id, port_id):
self._cisco_ucs_plugin.unplug_interface(tenant_id, net_id,
port_id)
self.tearDownNetworkPort(tenant_id, net_id, port_id)