diff --git a/quantum/plugins/cisco/run_tests.py b/quantum/plugins/cisco/run_tests.py new file mode 100644 index 0000000000..d63cc34a42 --- /dev/null +++ b/quantum/plugins/cisco/run_tests.py @@ -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)) diff --git a/quantum/plugins/cisco/tests/unit/__init__.py b/quantum/plugins/cisco/tests/unit/__init__.py new file mode 100644 index 0000000000..5910e35494 --- /dev/null +++ b/quantum/plugins/cisco/tests/unit/__init__.py @@ -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 diff --git a/quantum/plugins/cisco/tests/unit/test_l2networkApi.py b/quantum/plugins/cisco/tests/unit/test_l2networkApi.py new file mode 100644 index 0000000000..9cd619ef64 --- /dev/null +++ b/quantum/plugins/cisco/tests/unit/test_l2networkApi.py @@ -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) diff --git a/quantum/plugins/cisco/tests/unit/test_nexus_plugin.py b/quantum/plugins/cisco/tests/unit/test_nexus_plugin.py new file mode 100644 index 0000000000..013d798940 --- /dev/null +++ b/quantum/plugins/cisco/tests/unit/test_nexus_plugin.py @@ -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) diff --git a/quantum/plugins/cisco/tests/unit/test_ucs_driver.py b/quantum/plugins/cisco/tests/unit/test_ucs_driver.py new file mode 100644 index 0000000000..ab93d778f9 --- /dev/null +++ b/quantum/plugins/cisco/tests/unit/test_ucs_driver.py @@ -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 = " "\ +" "\ +"" + +create_profile_output = " "\ +" " + +change_vlan_output = " "\ +" "\ +" " + +delete_vlan_output = " "\ +" "\ +" " + +delete_profile_output = " "\ +" " + +associate_profile_output = " "\ +" " \ +" " + + +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") diff --git a/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py b/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py new file mode 100644 index 0000000000..978c026482 --- /dev/null +++ b/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py @@ -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 + 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)