--- /dev/null
+#!/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))
--- /dev/null
+# 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
--- /dev/null
+# 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)
--- /dev/null
+# 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)
--- /dev/null
+# 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")
--- /dev/null
+#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)
Code Review / openstack-build / neutron-build.git / commitdiff